Centralize app persistence in SQLite and harden state/event flows

apps/server/src/persistenceService.test.ts

@@ -102,7 +102,7 @@ describe("PersistenceService", () => {
         sessionId: "sess-1",
       };
 
-      service.ingestProviderEvent({
+      const event1 = service.ingestProviderEvent({
         ...baseEvent,
         id: "evt-1",
         createdAt: iso(10),
@@ -115,8 +115,9 @@ describe("PersistenceService", () => {
           },
         },
       });
+      expect(event1?.seq).toBeGreaterThan(0);
 
-      service.ingestProviderEvent({
+      const event2 = service.ingestProviderEvent({
         ...baseEvent,
         id: "evt-2",
         createdAt: iso(20),
@@ -128,8 +129,9 @@ describe("PersistenceService", () => {
           delta: "hi",
         },
       });
+      expect((event2?.seq ?? 0) > (event1?.seq ?? 0)).toBe(true);
 
-      service.ingestProviderEvent({
+      const event3 = service.ingestProviderEvent({
         ...baseEvent,
         id: "evt-3",
         createdAt: iso(30),
@@ -142,8 +144,9 @@ describe("PersistenceService", () => {
           },
         },
       });
+      expect((event3?.seq ?? 0) > (event2?.seq ?? 0)).toBe(true);
 
-      service.ingestProviderEvent({
+      const event4 = service.ingestProviderEvent({
         ...baseEvent,
         id: "evt-4",
         createdAt: iso(40),
@@ -156,6 +159,7 @@ describe("PersistenceService", () => {
           },
         },
       });
+      expect((event4?.seq ?? 0) > (event3?.seq ?? 0)).toBe(true);
 
       const snapshot = service.loadSnapshot();
       expect(snapshot.projects).toHaveLength(1);
@@ -179,6 +183,24 @@ describe("PersistenceService", () => {
       expect(catchUp.events.some((event) => event.eventType === "project.upsert")).toBe(true);
       expect(catchUp.events.some((event) => event.eventType === "thread.upsert")).toBe(true);
       expect(catchUp.events.some((event) => event.eventType === "message.upsert")).toBe(true);
+
+      const providerCatchUp = service.providerCatchUp({ afterSeq: 0 });
+      expect(providerCatchUp.lastProviderSeq).toBe(event4?.seq ?? 0);
+      expect(providerCatchUp.events.map((event) => event.id)).toEqual([
+        "evt-1",
+        "evt-2",
+        "evt-3",
+        "evt-4",
+      ]);
+      const providerSeqs = providerCatchUp.events.map((event) => event.seq ?? 0);
+      for (let index = 1; index < providerSeqs.length; index += 1) {
+        expect(providerSeqs[index]).toBeGreaterThan(providerSeqs[index - 1] ?? Number.NEGATIVE_INFINITY);
+      }
+
+      const providerCatchUpAfterSecond = service.providerCatchUp({
+        afterSeq: event2?.seq ?? 0,
+      });
+      expect(providerCatchUpAfterSecond.events.map((event) => event.id)).toEqual(["evt-3", "evt-4"]);
     } finally {
       service.close();
     }

apps/server/src/persistenceService.ts

@@ -16,6 +16,8 @@ import {
   type ProjectRemoveInput,
   type ProjectUpdateScriptsInput,
   type ProjectUpdateScriptsResult,
+  type ProviderCatchUpInput,
+  type ProviderCatchUpResult,
   type ProviderEvent,
   type ProviderSendTurnInput,
   type StateBootstrapResult,
@@ -44,6 +46,9 @@ import {
   projectRemoveInputSchema,
   projectScriptsSchema,
   projectUpdateScriptsInputSchema,
+  providerCatchUpInputSchema,
+  providerCatchUpResultSchema,
+  providerEventSchema,
   stateBootstrapResultSchema,
   stateCatchUpInputSchema,
   stateCatchUpResultSchema,
@@ -94,6 +99,25 @@ interface StateEventRow {
 interface ProviderEventInsertResult {
   inserted: boolean;
   runtimeThreadId: string | null;
+  seq: number;
+}
+
+interface ProviderEventRow {
+  seq: number;
+  id: string;
+  session_id: string;
+  provider: string;
+  kind: string;
+  method: string;
+  thread_id: string | null;
+  turn_id: string | null;
+  item_id: string | null;
+  request_id: string | null;
+  request_kind: string | null;
+  text_delta: string | null;
+  message: string | null;
+  payload_json: string | null;
+  created_at: string;
 }
 
 export interface PersistenceServiceOptions {
@@ -772,6 +796,43 @@ export class PersistenceService extends EventEmitter<PersistenceServiceEvents> {
     });
   }
 
+  providerCatchUp(raw: ProviderCatchUpInput): ProviderCatchUpResult {
+    const input = providerCatchUpInputSchema.parse(raw);
+    const rows = this.db
+      .prepare(
+        `SELECT
+          seq,
+          id,
+          session_id,
+          provider,
+          kind,
+          method,
+          thread_id,
+          turn_id,
+          item_id,
+          request_id,
+          request_kind,
+          text_delta,
+          message,
+          payload_json,
+          created_at
+        FROM provider_events
+        WHERE seq > ?
+        ORDER BY seq ASC
+        LIMIT ?;`,
+      )
+      .all(input.afterSeq, input.limit) as unknown as ProviderEventRow[];
+
+    const events = rows
+      .map((row) => this.parseProviderEventRow(row))
+      .filter((event): event is ProviderEvent => event !== null);
+
+    return providerCatchUpResultSchema.parse({
+      events,
+      lastProviderSeq: this.readLastProviderSeq(),
+    });
+  }
+
   listMessages(raw: StateListMessagesInput): StateListMessagesResult {
     const input = stateListMessagesInputSchema.parse(raw);
     const rows = this.db
@@ -867,12 +928,18 @@ export class PersistenceService extends EventEmitter<PersistenceServiceEvents> {
     });
   }
 
-  ingestProviderEvent(event: ProviderEvent): void {
+  ingestProviderEvent(event: ProviderEvent): ProviderEvent | null {
+    let persistedEvent: ProviderEvent | null = null;
     this.withTransaction((pendingEvents) => {
       const insertResult = this.insertProviderEvent(event);
       if (!insertResult.inserted) {
         return;
       }
+      persistedEvent = providerEventSchema.parse({
+        ...event,
+        seq: insertResult.seq,
+        ...(insertResult.runtimeThreadId ? { threadId: insertResult.runtimeThreadId } : {}),
+      });
 
       const localThreadId = this.resolveThreadIdForEvent(event, insertResult.runtimeThreadId);
       if (!localThreadId) {
@@ -1138,6 +1205,7 @@ export class PersistenceService extends EventEmitter<PersistenceServiceEvents> {
         }
       }
     });
+    return persistedEvent;
   }
 
   persistTurnDiffSummaryFromCheckpoint(input: {
@@ -1360,6 +1428,13 @@ export class PersistenceService extends EventEmitter<PersistenceServiceEvents> {
     return row?.seq ?? 0;
   }
 
+  private readLastProviderSeq(): number {
+    const row = this.db
+      .prepare("SELECT COALESCE(MAX(seq), 0) AS seq FROM provider_events;")
+      .get() as { seq: number } | undefined;
+    return row?.seq ?? 0;
+  }
+
   private withTransaction<T>(fn: (pendingEvents: StateEvent[]) => T): T {
     const pendingEvents: StateEvent[] = [];
     const result = this.stateDb.transaction(() => fn(pendingEvents));
@@ -1766,13 +1841,40 @@ export class PersistenceService extends EventEmitter<PersistenceServiceEvents> {
         event.message ?? null,
         payloadJson,
         event.createdAt,
-      ) as { changes?: number | bigint };
+      ) as { changes?: number | bigint; lastInsertRowid?: number | bigint };
+    const inserted = toSafeInteger(result.changes, 0) > 0;
     return {
-      inserted: toSafeInteger(result.changes, 0) > 0,
+      inserted,
       runtimeThreadId: runtimeThreadId ?? null,
+      seq: inserted ? toSafeInteger(result.lastInsertRowid, 0) : 0,
     };
   }
 
+  private parseProviderEventRow(row: ProviderEventRow): ProviderEvent | null {
+    const payload = row.payload_json ? this.tryParseJson(row.payload_json) : undefined;
+    const parsed = providerEventSchema.safeParse({
+      seq: row.seq,
+      id: row.id,
+      kind: row.kind,
+      provider: row.provider,
+      sessionId: row.session_id,
+      createdAt: row.created_at,
+      method: row.method,
+      ...(row.thread_id ? { threadId: row.thread_id } : {}),
+      ...(row.turn_id ? { turnId: row.turn_id } : {}),
+      ...(row.item_id ? { itemId: row.item_id } : {}),
+      ...(row.request_id ? { requestId: row.request_id } : {}),
+      ...(row.request_kind ? { requestKind: row.request_kind } : {}),
+      ...(row.text_delta !== null ? { textDelta: row.text_delta } : {}),
+      ...(row.message ? { message: row.message } : {}),
+      ...(payload !== undefined ? { payload } : {}),
+    });
+    if (!parsed.success) {
+      return null;
+    }
+    return parsed.data;
+  }
+
   private parseJson<T>(json: string, schema: SafeParseSchema<T>): T | null {
     const candidate = this.tryParseJson(json);
     const parsed = schema.safeParse(candidate);

apps/server/src/providerManager.test.ts

@@ -3,6 +3,58 @@ import { describe, expect, it, vi } from "vitest";
 import { ProviderManager } from "./providerManager";
 
 describe("ProviderManager", () => {
+  it("emits persisted provider events with durable sequence metadata when available", () => {
+    const ingestProviderEvent = vi.fn((event: {
+      id: string;
+      kind: "notification";
+      provider: "codex";
+      sessionId: string;
+      createdAt: string;
+      method: string;
+      threadId: string;
+    }) => ({
+      ...event,
+      seq: 11,
+    }));
+    const persistenceService = {
+      ingestProviderEvent,
+      unbindSession: vi.fn(),
+    };
+    const manager = new ProviderManager({
+      persistenceService: persistenceService as never,
+    });
+    const events: Array<{ id: string; seq?: number }> = [];
+    manager.on("event", (event) => {
+      events.push({ id: event.id, seq: event.seq });
+    });
+
+    const internals = manager as unknown as {
+      onCodexEvent: (event: {
+        id: string;
+        kind: "notification";
+        provider: "codex";
+        sessionId: string;
+        createdAt: string;
+        method: string;
+        threadId: string;
+      }) => void;
+    };
+
+    internals.onCodexEvent({
+      id: "evt-seq-1",
+      kind: "notification",
+      provider: "codex",
+      sessionId: "sess-1",
+      createdAt: new Date().toISOString(),
+      method: "turn/started",
+      threadId: "thread-1",
+    });
+
+    expect(ingestProviderEvent).toHaveBeenCalledTimes(1);
+    expect(events).toEqual([{ id: "evt-seq-1", seq: 11 }]);
+    manager.dispose();
+  });
+
   it("detaches provider event listener and ends thread log streams on dispose", () => {
     const manager = new ProviderManager();
     const internals = manager as unknown as {

apps/server/src/providerManager.ts

@@ -180,18 +180,11 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
     if (this.disposed) {
       return;
     }
-
-    this.routeEventToThreadLog(event);
-    try {
-      this.persistenceService?.ingestProviderEvent(event);
-    } catch {
-      // Persistence failures should not break provider streaming.
-    }
-    if (event.method === "session/closed" || event.method === "session/exited") {
-      this.persistenceService?.unbindSession(event.sessionId);
+    const emittedEvent = this.publishProviderEvent(event);
+    if (!emittedEvent) {
+      return;
     }
-    this.emit("event", event);
-    this.maybeCaptureFilesystemCheckpoint(event);
+    this.maybeCaptureFilesystemCheckpoint(emittedEvent);
   };
 
   constructor(options: ProviderManagerOptions = {}) {
@@ -546,7 +539,7 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
   }
 
   private emitCheckpointCaptured(sessionId: string, threadId: string, turnCount: number): void {
-    this.emit("event", {
+    this.publishProviderEvent({
       id: randomUUID(),
       kind: "notification",
       provider: "codex",
@@ -566,7 +559,7 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
     message: string,
     threadId?: string,
   ): void {
-    this.emit("event", {
+    this.publishProviderEvent({
       id: randomUUID(),
       kind: "error",
       provider: "codex",
@@ -578,6 +571,26 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
     });
   }
 
+  private publishProviderEvent(event: ProviderEvent): ProviderEvent | null {
+    let emittedEvent = event;
+    try {
+      const persistedEvent = this.persistenceService?.ingestProviderEvent(event);
+      if (this.persistenceService && !persistedEvent) {
+        return null;
+      }
+      emittedEvent = persistedEvent ?? event;
+    } catch {
+      // Persistence failures should not break provider streaming.
+    }
+
+    this.routeEventToThreadLog(emittedEvent);
+    if (emittedEvent.method === "session/closed" || emittedEvent.method === "session/exited") {
+      this.persistenceService?.unbindSession(emittedEvent.sessionId);
+    }
+    this.emit("event", emittedEvent);
+    return emittedEvent;
+  }
+
   private withFilesystemLock<T>(sessionId: string, fn: () => Promise<T>): Promise<T> {
     const previous = this.filesystemLocks.get(sessionId) ?? Promise.resolve();
     const next = previous.catch(() => undefined).then(fn);