fix(server): preserve reconnect sync progress on unreliable networks

apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts

@@ -68,6 +68,72 @@ const seedEvents = Effect.fn("test.seedSequenceEvents")(function* (
   `;
 });
 
+it.effect("checks creation ranges without restoring or decoding payloads", () =>
+  Effect.gen(function* () {
+    const store = yield* OrchestrationEventStore.OrchestrationEventStore;
+    const threadId = ThreadId.make("creation-existence");
+    const inserted = yield* seedEvents([
+      { ...eventRow(1, "thread", 2, threadId), event_type: "thread.created", payload_json: "{" },
+      eventRow(2, "thread", 2, threadId),
+      { ...eventRow(3, "thread", 2, "other"), event_type: "thread.created", payload_json: "{" },
+      { ...eventRow(4, "thread", 1, threadId), event_type: "thread.created", payload_json: "{" },
+    ]);
+    const range = {
+      threadId,
+      afterSequence: 0,
+      throughSequence: inserted[3]!.sequence,
+      eventType: "thread.created" as const,
+    };
+    assert.isTrue(yield* store.hasAgentEvent(range));
+    assert.isFalse(yield* store.hasAgentEvent({ ...range, afterSequence: inserted[0]!.sequence }));
+    assert.isFalse(yield* store.hasAgentEvent({ ...range, throughSequence: 0 }));
+    assert.isTrue(yield* store.hasAgentEvent({ ...range, throughSequence: inserted[0]!.sequence }));
+  }).pipe(Effect.provide(layerEventStore)),
+);
+
+it.effect("fetches the next byte-bounded replay page after delivering the previous page", () =>
+  Effect.gen(function* () {
+    const store = yield* OrchestrationEventStore.OrchestrationEventStore;
+    const sql = yield* SqlClient.SqlClient;
+    const threadId = ThreadId.make("byte-paged-replay");
+    const rows = [1, 2].map((ordinal) => ({
+      ...eventRow(ordinal, "thread", 2, threadId),
+      payload_json: encodeJson({
+        providerSessionId: `session:${ordinal}:${"x".repeat(180_000)}`,
+        detachedAt: occurredAt,
+      }),
+    }));
+    const inserted = yield* seedEvents(rows);
+    const updatedPayload = {
+      providerSessionId: ProviderSessionId.make("session:updated"),
+      detachedAt: occurredAt,
+    };
+    const items = yield* store
+      .readAgentEvents({
+        threadId,
+        throughSequence: inserted[1]!.sequence,
+        maxPageBytes: 256 * 1024,
+      })
+      .pipe(
+        Stream.tap((item) =>
+          item.sequence === inserted[0]!.sequence
+            ? sql`UPDATE orchestration_events SET payload_json = ${encodeJson(updatedPayload)}
+              WHERE sequence = ${inserted[1]!.sequence}`
+            : Effect.void,
+        ),
+        Stream.runCollect,
+      );
+    assert.deepEqual(
+      items.map((item) => item.sequence),
+      inserted.map((item) => item.sequence),
+    );
+    assert.deepEqual(items[1]!.event.payload, {
+      ...updatedPayload,
+      detachedAt: DateTime.makeUnsafe(occurredAt),
+    });
+  }).pipe(Effect.provide(layerEventStore)),
+);
+
 it.effect("keeps application and scoped agent high-water marks separate from legacy history", () =>
   Effect.gen(function* () {
     const store = yield* OrchestrationEventStore.OrchestrationEventStore;

apps/server/src/persistence/OrchestrationEventStore.test.ts

@@ -113,7 +113,7 @@ layer("OrchestrationEventStore", (it) => {
       const id = NodeId.make("node:subagent-prompt");
       const driver = ProviderDriverKind.make("claudeAgent");
       const providerInstanceId = ProviderInstanceId.make("claudeAgent");
-      const prompt = "Audit the repository. ".repeat(200);
+      const prompt = 'Audit "repo"\n漢字 '.repeat(9_000);
       const subagent = (index: number, progress: string, nextPrompt = prompt) =>
         ({
           id: EventId.make(`event:subagent-prompt:subagent:${index}`),
@@ -185,14 +185,14 @@ layer("OrchestrationEventStore", (it) => {
         subagent(3, "Retrying", "Audit again."),
         item(3, "Retrying", "Audit again."),
       ];
-      yield* store.appendAgentEvents({ events });
+      const appended = yield* store.appendAgentEvents({ events });
 
       const stored = yield* sql<{ readonly payload_json: string }>`
         SELECT payload_json FROM orchestration_events
         WHERE stream_id = ${threadId} ORDER BY sequence ASC
       `;
       assert.deepEqual(
-        stored.map((row) => row.payload_json.includes(prompt)),
+        stored.map((row) => row.payload_json.includes(JSON.stringify(prompt).slice(1, -1))),
         [true, true, false, false, false, false, false, false],
       );
       const replayed = yield* store
@@ -214,6 +214,28 @@ layer("OrchestrationEventStore", (it) => {
         throughSequence: yield* store.latestApplicationSequence,
       });
       assert.isAbove(applicationStats.rawPayloadBytes, prompt.length * 6);
+      const paged = yield* store
+        .readAgentEvents({ threadId, afterSequence, maxPageBytes: 256 * 1024 })
+        .pipe(Stream.runCollect);
+      assert.deepEqual(
+        paged.map((entry) => entry.event),
+        events,
+      );
+      // The first two updates here reference prompts in earlier anchor rows.
+      // Sizing must stop before touching a later page's malformed payload.
+      yield* sql`UPDATE orchestration_events SET payload_json = '{' WHERE sequence = ${appended[4]!.sequence}`;
+      const firstPage = yield* store
+        .readAgentEvents({
+          threadId,
+          afterSequence: appended[1]!.sequence,
+          throughSequence: appended[4]!.sequence,
+          maxPageBytes: 256 * 1024,
+        })
+        .pipe(Stream.take(1), Stream.runCollect);
+      assert.deepEqual(
+        firstPage.map((entry) => entry.event),
+        [events[2]!],
+      );
     }),
   );
 

apps/server/src/persistence/OrchestrationEventStore.ts

@@ -60,8 +60,18 @@ export class OrchestrationEventStore extends Context.Service<
       readonly eventType?: OrchestrationV2DomainEvent["type"];
       readonly limit?: number;
       readonly skipUnknownEventTypes?: boolean;
+      /** Bound fetched JSON per page, always allowing one indivisible event. */
+      readonly maxPageBytes?: number;
     }) => Stream.Stream<OrchestrationV2StoredEvent, OrchestrationEventStoreError>;
 
+    /** Check a scoped replay range without reading or restoring event payloads. */
+    readonly hasAgentEvent: (input: {
+      readonly threadId: ThreadId;
+      readonly afterSequence: number;
+      readonly throughSequence: number;
+      readonly eventType: OrchestrationV2DomainEvent["type"];
+    }) => Effect.Effect<boolean, OrchestrationEventStoreError>;
+
     readonly getAgentReplayStats: (input: {
       readonly threadId: ThreadId;
       readonly afterSequence: number;
@@ -392,9 +402,9 @@ const makeEventStore = Effect.gen(function* () {
     readonly eventType?: OrchestrationV2DomainEvent["type"];
     readonly onlyAgentEvents?: boolean;
     readonly limit: number;
-  }) =>
-    sql<ApplicationEventRow>`
-      SELECT
+    readonly maxPageBytes?: number;
+  }) => {
+    const columns = sql`
         sequence,
         event_id,
         command_id,
@@ -407,6 +417,8 @@ const makeEventStore = Effect.gen(function* () {
         application_event_version,
         causation_event_id,
         correlation_id
+    `;
+    const from = sql`
       FROM orchestration_events
       ${
         // Otherwise the sequence index wins and command lookups scan every V2 event.
@@ -431,9 +443,41 @@ const makeEventStore = Effect.gen(function* () {
           ...(input.commandId === undefined ? [] : [sql`command_id = ${input.commandId}`]),
           ...(input.eventType === undefined ? [] : [sql`event_type = ${input.eventType}`]),
         ])}
-      ORDER BY sequence ASC
-      LIMIT ${input.limit}
+    `;
+    if (input.maxPageBytes === undefined) {
+      return sql<ApplicationEventRow>`
+        SELECT ${columns} ${from} ORDER BY sequence ASC LIMIT ${input.limit}
+      `.pipe(NodeSqliteClient.readOnly);
+    }
+    // Materialize only candidate ids. Walk their restored sizes until the page
+    // fills (at most one overflow row), rather than restoring all 500 bodies.
+    // Prompt references must be sized after restoration, including JSON escaping.
+    return sql<ApplicationEventRow>`
+      WITH RECURSIVE candidates AS MATERIALIZED (
+        SELECT sequence, ROW_NUMBER() OVER (ORDER BY sequence) AS ordinal
+        ${from} ORDER BY sequence ASC LIMIT ${input.limit}
+      ), replay_page(cursor, replay_bytes, row_count) AS (
+        SELECT orchestration_events.sequence,
+          octet_length(${restoredPayloadJson}) + octet_length(metadata_json), 1
+        FROM orchestration_events
+        JOIN candidates ON candidates.sequence = orchestration_events.sequence
+        WHERE candidates.ordinal = 1
+        UNION ALL
+        SELECT orchestration_events.sequence,
+          page.replay_bytes + octet_length(${restoredPayloadJson}) + octet_length(metadata_json),
+          page.row_count + 1
+        FROM replay_page AS page
+        JOIN candidates ON candidates.ordinal = page.row_count + 1
+        JOIN orchestration_events ON orchestration_events.sequence = candidates.sequence
+        WHERE page.replay_bytes < ${input.maxPageBytes}
+      )
+      SELECT ${columns}
+      FROM orchestration_events
+      JOIN replay_page ON replay_page.cursor = orchestration_events.sequence
+      WHERE replay_page.row_count = 1 OR replay_page.replay_bytes <= ${input.maxPageBytes}
+      ORDER BY sequence
     `.pipe(NodeSqliteClient.readOnly);
+  };
 
   const appendAgentEvents: OrchestrationEventStore["Service"]["appendAgentEvents"] = (input) =>
     Effect.forEach(
@@ -556,6 +600,7 @@ const makeEventStore = Effect.gen(function* () {
           ...(input?.eventType === undefined ? {} : { eventType: input.eventType }),
           onlyAgentEvents: true,
           limit: pageLimit,
+          ...(input?.maxPageBytes === undefined ? {} : { maxPageBytes: input.maxPageBytes }),
         }).pipe(
           Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readAgentEvents:query")),
           Effect.map((rows) => {
@@ -564,7 +609,7 @@ const makeEventStore = Effect.gen(function* () {
             return [
               rows,
               last === undefined ||
-              rows.length < pageLimit ||
+              (input?.maxPageBytes === undefined && rows.length < pageLimit) ||
               nextRemaining <= 0 ||
               (input?.throughSequence !== undefined && last.sequence >= input.throughSequence)
                 ? Option.none()
@@ -585,6 +630,23 @@ const makeEventStore = Effect.gen(function* () {
     );
   };
 
+  const hasAgentEvent: OrchestrationEventStore["Service"]["hasAgentEvent"] = (input) =>
+    sql<{ readonly found: number }>`
+      SELECT EXISTS (
+        SELECT 1 FROM orchestration_events
+        WHERE aggregate_kind = 'thread'
+          AND stream_id = ${input.threadId}
+          AND application_event_version = 2
+          AND sequence > ${input.afterSequence}
+          AND sequence <= ${input.throughSequence}
+          AND event_type = ${input.eventType}
+      ) AS found
+    `.pipe(
+      NodeSqliteClient.readOnly,
+      Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.hasAgentEvent:query")),
+      Effect.map((rows) => rows[0]?.found === 1),
+    );
+
   const getAgentReplayStats: OrchestrationEventStore["Service"]["getAgentReplayStats"] = (input) =>
     sql<{
       readonly eventCount: number;
@@ -743,6 +805,7 @@ const makeEventStore = Effect.gen(function* () {
     appendProjectEvent,
     appendAgentEvents,
     readAgentEvents,
+    hasAgentEvent,
     getAgentReplayStats,
     getReplayStats,
     latestAgentSequence,

apps/server/src/ws.test.ts

@@ -303,6 +303,71 @@ it.live.each(["thread", "shell"] as const)(
     }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)),
 );
 
+it.effect.each([
+  { count: 260, padding: 0, unknownPrefix: 0 },
+  { count: 3, padding: 600_000, unknownPrefix: 0 },
+  { count: 3, padding: 0, unknownPrefix: 140 },
+])(
+  "resumes catch-up of $count events ($padding bytes, $unknownPrefix unknown rows) after interruption",
+  ({ count, padding, unknownPrefix }) =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      const eventStore = yield* OrchestrationEventStore.OrchestrationEventStore;
+      const threadId = ThreadId.make("thread:incremental-replay");
+      const now = "2026-10-28T00:00:00.000Z";
+      const rows = Array.from({ length: count + unknownPrefix }, (_, version) => ({
+        event_id: `incremental-replay:${version}`,
+        aggregate_kind: "thread",
+        stream_id: threadId,
+        stream_version: version,
+        event_type: version < unknownPrefix ? "future.event" : "provider-session.detached",
+        occurred_at: now,
+        actor_kind: "server",
+        payload_json: JSON.stringify({
+          providerSessionId: `session:${version}:${"x".repeat(padding)}`,
+          detachedAt: now,
+        }),
+        metadata_json: "{}",
+        application_event_version: 2,
+      }));
+      const inserted = yield* sql<{ sequence: number }>`
+      INSERT INTO orchestration_events ${sql.insert(rows)} RETURNING sequence
+    `;
+      const subscribe = (afterSequence: number) =>
+        subscribeOrchestrationV2Thread({
+          threadId,
+          afterSequence,
+          requestCompletionMarker: true,
+        }).pipe(
+          Effect.provideService(OrchestrationEventStore.OrchestrationEventStore, {
+            ...eventStore,
+            getAgentReplayStats: () =>
+              Effect.die("Ordinary catch-up must not restore payloads for stats"),
+          }),
+          Effect.provideService(ThreadManagement.ThreadManagementService, {
+            ensureLegacyTranscript: () => Effect.void,
+            getThreadSnapshot: () =>
+              Effect.die("Ordinary catch-up must not restart with a snapshot"),
+            streamStoredEventsFrom: () => Stream.empty,
+          } as unknown as ThreadManagement.ThreadManagementService["Service"]),
+        );
+      const first = yield* subscribe(0).pipe(
+        Effect.flatMap((stream) =>
+          stream.pipe(Stream.take(Math.min(10, count - 1)), Stream.runCollect),
+        ),
+      );
+      const last = first.at(-1);
+      assert.equal(first.length, Math.min(10, count - 1));
+      assert.equal(last?.kind, "event");
+      if (last?.kind !== "event") return;
+      const resumed = yield* subscribe(last.sequence).pipe(Effect.flatMap(Stream.runCollect));
+      assert.deepEqual(
+        [...first, ...resumed].map((item) => (item.kind === "event" ? item.sequence : item.kind)),
+        [...inserted.slice(unknownPrefix).map((row) => row.sequence), "synchronized"],
+      );
+    }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)),
+);
+
 it.effect("coalesces streaming replay before synchronization and the live tail", () =>
   Effect.gen(function* () {
     const store = yield* OrchestrationEventStore.OrchestrationEventStore;