import { randomUUID } from "node:crypto";import { EventEmitter } from "node:events";import fs from "node:fs";import path from "node:path";import * as Effect from "effect/Effect";import * as Fiber from "effect/Fiber";import * as Queue from "effect/Queue";import * as SqlClient from "effect/unstable/sql/SqlClient"; import {  DEFAULT_MODEL,  type AppSettings,  type AppSettingsUpdateInput,  type ProjectAddInput,  type ProjectAddResult,  type ProjectListResult,  type ProjectRemoveInput,  type ProjectUpdateScriptsInput,  type ProjectUpdateScriptsResult,  type ProviderEvent,  type ProviderSendTurnInput,  type StateBootstrapResult,  type StateBootstrapThread,  type StateCatchUpInput,  type StateCatchUpResult,  type StateEvent,  type StateListMessagesInput,  type StateListMessagesResult,  type StateMessage,  type StateProject,  type StateThread,  type StateTurnDiffFileChange,  type StateTurnSummary,  type ThreadsCreateInput,  type ThreadsDeleteInput,  type ThreadsMarkVisitedInput,  type ThreadsUpdateBranchInput,  type ThreadsUpdateModelInput,  type ThreadsUpdateResult,  type ThreadsUpdateTerminalStateInput,  type ThreadsUpdateTitleInput,  normalizeProjectScripts,  projectAddInputSchema,  projectRecordSchema,  projectRemoveInputSchema,  projectScriptsSchema,  projectUpdateScriptsInputSchema,  stateCatchUpInputSchema,  stateListMessagesInputSchema,  stateMessageSchema,  stateProjectSchema,  stateThreadSchema,  stateTurnSummarySchema,  threadsCreateInputSchema,  threadsDeleteInputSchema,  threadsMarkVisitedInputSchema,  threadsUpdateBranchInputSchema,  threadsUpdateModelInputSchema,  threadsUpdateTerminalStateInputSchema,  threadsUpdateTitleInputSchema,  threadsUpdateResultSchema,} from "@t3tools/contracts"; import {  countMessagesForThread as countMessagesForThreadEffect,  deleteDocumentByIdAndKind as deleteDocumentByIdAndKindEffect,  deleteDocumentsByProjectId as deleteDocumentsByProjectIdEffect,  deleteDocumentsByThreadId as deleteDocumentsByThreadIdEffect,  findThreadPayloadByRuntimeThreadId as findThreadPayloadByRuntimeThreadIdEffect,  getDocumentRowById as getDocumentRowByIdEffect,  listPaginatedMessagePayloadsForThread as listPaginatedMessagePayloadsForThreadEffect,  listProjectPayloads as listProjectPayloadsEffect,  listMessagePayloadsForThread as listMessagePayloadsForThreadEffect,  listMessagePayloadsForThreadDesc as listMessagePayloadsForThreadDescEffect,  listThreadPayloads as listThreadPayloadsEffect,  listThreadPayloadsByProject as listThreadPayloadsByProjectEffect,  listTurnSummaryPayloadsForThread as listTurnSummaryPayloadsForThreadEffect,  readNextSortKey as readNextSortKeyEffect,  upsertDocument as upsertDocumentEffect,  type DocumentRow,} from "./persistence/repos/documentsRepo";import {  insertProviderEvent as insertProviderEventEffect,  listCompletedItemEventsBySessionTurn as listCompletedItemEventsBySessionTurnEffect,  listCompletedItemEventsByThreadTurn as listCompletedItemEventsByThreadTurnEffect,  listCompletedItemEventsByTurn as listCompletedItemEventsByTurnEffect,  type CompletedProviderItemRow,} from "./persistence/repos/providerEventsRepo";import {  appendStateEvent as appendStateEventEffect,  listStateEventsAfterSeq as listStateEventsAfterSeqEffect,  readLastStateSeq as readLastStateSeqEffect,} from "./persistence/repos/stateEventsRepo";import {  readMetadataValue as readMetadataValueEffect,  writeMetadataValue as writeMetadataValueEffect,} from "./persistence/repos/metadataRepo";import {  buildUpdatedAppSettings,  resolveAppSettings,} from "./persistence/domain/appSettings";import { buildUserTurnMessage, messageDocId } from "./persistence/domain/messages";import {  inferProjectName,  isDirectory,  normalizeCwd,} from "./persistence/domain/projects";import {  asObject,  asString,  normalizeProviderItemType,  parseAssistantItemId,  parseThreadIdFromEventPayload,  parseTurnIdFromEvent,} from "./persistence/domain/providerProjection";import {  buildStateBootstrapResult,  buildStateCatchUpResult,  buildStateListMessagesResult,} from "./persistence/domain/stateSync";import { fallbackGroupId, normalizeTerminalIds, normalizeThread } from "./persistence/domain/threads";import { mergeTurnSummaryFiles, summarizeUnifiedDiff } from "./persistence/domain/turnSummaries";import { resolvePersistenceConfig } from "./persistence/config";import { PersistenceInitializationError } from "./persistence/errors";import { runPersistenceMigrations } from "./persistence/migrator";import { runWithSqlClient } from "./persistence/runtime";import { openPersistenceSqliteDatabase } from "./persistence/sqliteLayer";import type { SqliteDatabase } from "./persistence/sqliteAdapter"; const METADATA_KEY_PROJECTS_JSON_IMPORTED = "migration.projects_json_imported";const METADATA_KEY_APP_SETTINGS = "app.settings.v1";const DEFAULT_TERMINAL_ID = "default";const DEFAULT_TERMINAL_HEIGHT = 280; interface ProviderEventInsertResult {  inserted: boolean;  runtimeThreadId: string | null;} export interface PersistenceServiceOptions {  dbPath: string;  legacyProjectsJsonPath?: string;} export interface PersistenceServiceEvents {  stateEvent: [event: StateEvent];} interface UpsertDocumentInput {  id: string;  kind: "project" | "thread" | "message" | "turn_summary";  projectId: string | null;  threadId: string | null;  sortKey: number | null;  createdAt: string;  updatedAt: string;  data: unknown;} interface SafeParseSchema<T> {  safeParse(input: unknown): { success: true; data: T } | { success: false };} function nowIso(): string {  return new Date().toISOString();} function toSafeInteger(value: unknown, fallback = 0): number {  if (typeof value === "number" && Number.isFinite(value)) {    return Math.trunc(value);  }  if (typeof value === "bigint") {    return Number(value);  }  return fallback;} function projectDocId(projectId: string): string {  return `project:${projectId}`;} function threadDocId(threadId: string): string {  return `thread:${threadId}`;} function turnSummaryDocId(threadId: string, turnId: string): string {  return `turn_summary:${threadId}:${turnId}`;} export class PersistenceService extends EventEmitter<PersistenceServiceEvents> {  private readonly db: SqliteDatabase;  private readonly sessionThreadIds = new Map<string, string>();  private readonly runtimeThreadIds = new Map<string, string>();  private readonly stateEventsQueue = Effect.runSync(Queue.unbounded<StateEvent>());  private readonly stateEventsBridge = Effect.runFork(this.runStateEventsBridge());  private closed = false;   constructor(options: PersistenceServiceOptions) {    super();    const config = resolvePersistenceConfig(options);    this.db = openPersistenceSqliteDatabase(config.dbPath);    try {      runPersistenceMigrations(this.db);    } catch (error) {      try {        this.db.close();      } catch {        // Best effort close on failed initialization.      }      throw new PersistenceInitializationError("Failed to initialize persistence database", {        cause: error,      });    }    if (config.legacyProjectsJsonPath) {      this.importProjectsJsonIfNeeded(config.legacyProjectsJsonPath);    }  }   close(): void {    if (this.closed) {      return;    }    this.closed = true;    try {      Effect.runSync(Queue.shutdown(this.stateEventsQueue));    } catch {      // Best effort shutdown.    }    try {      Effect.runSync(Fiber.interrupt(this.stateEventsBridge));    } catch {      // Ignore bridge interruption failures during shutdown.    }    this.db.close();  }   getAppSettings(): AppSettings {    return resolveAppSettings(this.readMetadata(METADATA_KEY_APP_SETTINGS));  }   updateAppSettings(raw: AppSettingsUpdateInput): AppSettings {    const next = buildUpdatedAppSettings(this.getAppSettings(), raw);    this.writeMetadata(METADATA_KEY_APP_SETTINGS, next);    return next;  }   listProjects(): ProjectListResult {    const payloads = this.runWithEffectSql(listProjectPayloadsEffect());     const projects: StateProject[] = [];    for (const payload of payloads) {      const parsed = this.parseJson(payload, stateProjectSchema);      if (parsed) {        projects.push(parsed);      }    }    return projects;  }   addProject(raw: ProjectAddInput): ProjectAddResult {    const input = projectAddInputSchema.parse(raw);    const normalizedCwd = normalizeCwd(input.cwd);    if (!isDirectory(normalizedCwd)) {      throw new Error(`Project path does not exist: ${normalizedCwd}`);    }     const existing = this.findProjectByNormalizedCwd(normalizedCwd);    if (existing) {      return { project: existing, created: false };    }     const now = nowIso();    const project = stateProjectSchema.parse({      id: randomUUID(),      cwd: normalizedCwd,      name: inferProjectName(normalizedCwd),      scripts: [],      createdAt: now,      updatedAt: now,    });     this.withTransaction((pendingEvents) => {      this.upsertProjectDocument(project);      this.appendStateEvent(pendingEvents, "project.upsert", project.id, { project }, project.updatedAt);    });     return { project, created: true };  }   removeProject(raw: ProjectRemoveInput): void {    const input = projectRemoveInputSchema.parse(raw);    const existingProject = this.getProjectById(input.id);    if (!existingProject) {      return;    }     this.withTransaction((pendingEvents) => {      const threadPayloads = this.runWithEffectSql(listThreadPayloadsByProjectEffect(input.id));      const threadIds: string[] = [];      for (const payload of threadPayloads) {        const parsed = this.parseJson(payload, stateThreadSchema);        if (parsed) {          threadIds.push(parsed.id);        }      }       this.runWithEffectSql(deleteDocumentsByProjectIdEffect(input.id));       const eventTime = nowIso();      for (const threadId of threadIds) {        this.appendStateEvent(pendingEvents, "thread.delete", threadId, { threadId }, eventTime);      }      this.appendStateEvent(        pendingEvents,        "project.delete",        input.id,        { projectId: input.id },        eventTime,      );    });  }   updateProjectScripts(raw: ProjectUpdateScriptsInput): ProjectUpdateScriptsResult {    const input = projectUpdateScriptsInputSchema.parse(raw);    const existing = this.getProjectById(input.id);    if (!existing) {      throw new Error(`Project not found: ${input.id}`);    }     const nextScripts = normalizeProjectScripts(projectScriptsSchema.parse(input.scripts));    const updatedProject = stateProjectSchema.parse({      ...existing,      scripts: nextScripts,      updatedAt: nowIso(),    });     this.withTransaction((pendingEvents) => {      this.upsertProjectDocument(updatedProject);      this.appendStateEvent(        pendingEvents,        "project.upsert",        updatedProject.id,        { project: updatedProject },        updatedProject.updatedAt,      );    });     return {      project: updatedProject,    };  }   createThread(raw: ThreadsCreateInput): ThreadsUpdateResult {    const input = threadsCreateInputSchema.parse(raw);    const project = this.getProjectById(input.projectId);    if (!project) {      throw new Error(`Project not found: ${input.projectId}`);    }     const now = nowIso();    const terminalIds = normalizeTerminalIds(input.terminalIds ?? [DEFAULT_TERMINAL_ID]);    const activeTerminalId = input.activeTerminalId ?? terminalIds[0] ?? DEFAULT_TERMINAL_ID;    const thread = normalizeThread(      stateThreadSchema.parse({        id: randomUUID(),        codexThreadId: null,        projectId: project.id,        title: input.title ?? "New thread",        model: input.model ?? DEFAULT_MODEL,        terminalOpen: input.terminalOpen ?? false,        terminalHeight: input.terminalHeight ?? DEFAULT_TERMINAL_HEIGHT,        terminalIds,        activeTerminalId,        terminalGroups: input.terminalGroups ?? [],        activeTerminalGroupId:          input.activeTerminalGroupId ?? fallbackGroupId(activeTerminalId),        createdAt: now,        updatedAt: now,        lastVisitedAt: now,        branch: input.branch ?? null,        worktreePath: input.worktreePath ?? null,      }),    );     this.withTransaction((pendingEvents) => {      this.upsertThreadDocument(thread);      this.appendStateEvent(pendingEvents, "thread.upsert", thread.id, { thread }, thread.updatedAt);    });     return threadsUpdateResultSchema.parse({ thread });  }   updateThreadTitle(raw: ThreadsUpdateTitleInput): ThreadsUpdateResult {    const input = threadsUpdateTitleInputSchema.parse(raw);    return this.updateThreadWith(input.threadId, (thread) => ({      ...thread,      title: input.title,      updatedAt: nowIso(),    }));  }   updateThreadModel(raw: ThreadsUpdateModelInput): ThreadsUpdateResult {    const input = threadsUpdateModelInputSchema.parse(raw);    return this.updateThreadWith(input.threadId, (thread) => ({      ...thread,      model: input.model,      updatedAt: nowIso(),    }));  }   markThreadVisited(raw: ThreadsMarkVisitedInput): ThreadsUpdateResult {    const input = threadsMarkVisitedInputSchema.parse(raw);    const visitedAt = input.visitedAt ?? nowIso();    return this.updateThreadWith(input.threadId, (thread) => ({      ...thread,      lastVisitedAt: visitedAt,      updatedAt: nowIso(),    }));  }   updateThreadBranch(raw: ThreadsUpdateBranchInput): ThreadsUpdateResult {    const input = threadsUpdateBranchInputSchema.parse(raw);    return this.updateThreadWith(input.threadId, (thread) => ({      ...thread,      branch: input.branch,      worktreePath: input.worktreePath,      updatedAt: nowIso(),    }));  }   updateThreadTerminalState(raw: ThreadsUpdateTerminalStateInput): ThreadsUpdateResult {    const input = threadsUpdateTerminalStateInputSchema.parse(raw);    return this.updateThreadWith(input.threadId, (thread) =>      normalizeThread({        ...thread,        terminalOpen: input.terminalOpen ?? thread.terminalOpen,        terminalHeight: input.terminalHeight ?? thread.terminalHeight,        terminalIds: input.terminalIds ?? thread.terminalIds,        runningTerminalIds: input.runningTerminalIds ?? thread.runningTerminalIds,        activeTerminalId: input.activeTerminalId ?? thread.activeTerminalId,        terminalGroups: input.terminalGroups ?? thread.terminalGroups,        activeTerminalGroupId: input.activeTerminalGroupId ?? thread.activeTerminalGroupId,        updatedAt: nowIso(),      }),    );  }   deleteThread(raw: ThreadsDeleteInput): void {    const input = threadsDeleteInputSchema.parse(raw);    const thread = this.getThreadById(input.threadId);    if (!thread) {      return;    }     this.withTransaction((pendingEvents) => {      this.runWithEffectSql(deleteDocumentsByThreadIdEffect(thread.id));      this.runWithEffectSql(deleteDocumentByIdAndKindEffect(threadDocId(thread.id), "thread"));      this.appendStateEvent(pendingEvents, "thread.delete", thread.id, { threadId: thread.id }, nowIso());    });  }   loadSnapshot(): StateBootstrapResult {    const projects = this.listProjects();    const threadPayloads = this.runWithEffectSql(listThreadPayloadsEffect());     const threads: StateBootstrapThread[] = [];    for (const payload of threadPayloads) {      const parsedThread = this.parseJson(payload, stateThreadSchema);      if (!parsedThread) continue;      const messages = this.listMessagesForThread(parsedThread.id);      const turnDiffSummaries = this.listTurnSummariesForThread(parsedThread.id).map((summary) => {        if (summary.assistantMessageId) {          return summary;        }        const assistantMessageId = this.findAssistantMessageIdForTurn({          turnId: summary.turnId,          runtimeThreadId: parsedThread.codexThreadId,        });        if (!assistantMessageId) {          return summary;        }        return stateTurnSummarySchema.parse({          ...summary,          assistantMessageId,        });      });      threads.push({        ...parsedThread,        turnDiffSummaries,        messages,      });    }     const lastStateSeq = this.readLastStateSeq();    return buildStateBootstrapResult({      projects,      threads,      lastStateSeq,    });  }

Lines 1–500 of 1503. Load more