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.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,8 +402,9 @@ const makeEventStore = Effect.gen(function* () {
     readonly eventType?: OrchestrationV2DomainEvent["type"];
     readonly onlyAgentEvents?: boolean;
     readonly limit: number;
-  }) =>
-    sql<ApplicationEventRow>`
+    readonly maxPageBytes?: number;
+  }) => {
+    const query = sql<ApplicationEventRow>`
       SELECT
         sequence,
         event_id,
@@ -433,7 +444,19 @@ const makeEventStore = Effect.gen(function* () {
         ])}
       ORDER BY sequence ASC
       LIMIT ${input.limit}
+    `;
+    if (input.maxPageBytes === undefined) return query.pipe(NodeSqliteClient.readOnly);
+    return sql<ApplicationEventRow>`
+      SELECT * FROM (
+        SELECT *, ROW_NUMBER() OVER (ORDER BY sequence) AS replay_row,
+          SUM(octet_length(payload_json) + octet_length(metadata_json))
+            OVER (ORDER BY sequence ROWS UNBOUNDED PRECEDING) AS replay_bytes
+        FROM (${query})
+      )
+      WHERE replay_row = 1 OR replay_bytes <= ${input.maxPageBytes}
+      ORDER BY sequence
     `.pipe(NodeSqliteClient.readOnly);
+  };
 
   const appendAgentEvents: OrchestrationEventStore["Service"]["appendAgentEvents"] = (input) =>
     Effect.forEach(
@@ -556,6 +579,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 +588,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 +609,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 +784,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;

apps/server/src/ws.ts

@@ -554,6 +554,7 @@ function projectFileFailureContext(
 }
 
 const PROVIDER_STATUS_DEBOUNCE_MS = 200;
+const THREAD_REPLAY_PAGE_BYTES = 256 * 1024;
 
 // Middleware added later wraps middleware added earlier, so instrumentation wraps authorization.
 const ServerWsRpcGroup = WsRpcGroup.middleware(RpcInstrumentation);
@@ -823,13 +824,20 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs
           ),
         );
 
-    const loadReplayThrough = (afterSequence: number, throughSequence: number) =>
-      applicationEvents
+    const loadReplayThrough = (
+      afterSequence: number,
+      throughSequence: number,
+      incremental = false,
+    ) => {
+      let encodedBytes = 0;
+      const replay = applicationEvents
         .readAgentEvents({
           threadId: input.threadId,
           afterSequence,
           throughSequence,
-          limit: THREAD_RESUME_MAX_REPLAY_EVENTS + 1,
+          ...(incremental
+            ? { maxPageBytes: THREAD_REPLAY_PAGE_BYTES }
+            : { limit: THREAD_RESUME_MAX_REPLAY_EVENTS + 1 }),
           // Downgrades can leave unknown rows; skip them and resume live from highWater so later events survive.
           skipUnknownEventTypes: true,
         })
@@ -839,17 +847,32 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs
             sequence: stored.sequence,
             event: projectDomainEventForWire(stored.event),
           })),
-          Stream.runCollect,
-          Effect.map((items) => Array.from(items)),
-          Effect.mapError(
-            (cause) =>
-              new OrchestrationV2GetThreadProjectionError({
-                threadId: input.threadId,
-                message: `Failed while replaying orchestration V2 thread ${input.threadId}`,
-                cause,
-              }),
-          ),
         );
+      return (
+        incremental
+          ? replay.pipe(
+              // Count delivered events rather than stored rows: a page of
+              // unknown future event types must not truncate the catch-up.
+              Stream.take(THREAD_RESUME_MAX_REPLAY_EVENTS),
+              Stream.takeUntil((item) => {
+                encodedBytes += threadReplayEncodedBytes([item]);
+                return encodedBytes >= THREAD_REPLAY_PAGE_BYTES;
+              }),
+            )
+          : replay
+      ).pipe(
+        Stream.runCollect,
+        Effect.map((items) => Array.from(items)),
+        Effect.mapError(
+          (cause) =>
+            new OrchestrationV2GetThreadProjectionError({
+              threadId: input.threadId,
+              message: `Failed while replaying orchestration V2 thread ${input.threadId}`,
+              cause,
+            }),
+        ),
+      );
+    };
 
     const completionMarker =
       input.requestCompletionMarker === true
@@ -914,6 +937,46 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs
       if (input.afterSequence > highWater) {
         return yield* snapshotThenLive();
       }
+      const hasCreateEvent = yield* applicationEvents
+        .hasAgentEvent({
+          threadId: input.threadId,
+          afterSequence: input.afterSequence,
+          throughSequence: highWater,
+          eventType: "thread.created",
+        })
+        .pipe(
+          Effect.mapError(
+            (cause) =>
+              new OrchestrationV2GetThreadProjectionError({
+                threadId: input.threadId,
+                message: `Failed to check orchestration V2 thread ${input.threadId} creation`,
+                cause,
+              }),
+          ),
+        );
+      if (!hasCreateEvent) {
+        // Pull each page only as the transport consumes the previous one. The
+        // client commits each event's cursor before the final completion marker,
+        // so a dropped connection resumes from its partial catch-up progress.
+        const replay = Stream.paginate(input.afterSequence, (cursor) =>
+          loadReplayThrough(cursor, highWater, true).pipe(
+            Effect.map((items) => {
+              const last = items.at(-1);
+              return [
+                coalesceLiveUpdatedEvents(items),
+                last === undefined || last.sequence >= highWater
+                  ? Option.none()
+                  : Option.some(last.sequence),
+              ] as const;
+            }),
+          ),
+        );
+        return replay.pipe(
+          Stream.concat(completionMarker),
+          Stream.concat(eventStreamFrom(highWater)),
+          withLiveDedupe,
+        );
+      }
       const stats = yield* applicationEvents
         .getAgentReplayStats({
           threadId: input.threadId,