test(orchestration): cover bounded V2 socket fallback paging

apps/server/src/orchestration-v2/threadHistoryPaging.test.ts

@@ -24,6 +24,7 @@ import {
   THREAD_HISTORY_CURSOR_MAX_LENGTH,
   THREAD_HISTORY_PAGE_POLICY,
 } from "./threadHistoryPaging.ts";
+import { buildBoundedThreadStreamSnapshot } from "./ThreadStream.ts";
 import { projectThreadProjectionForWire } from "./WireProjection.ts";
 
 const NOW = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z");
@@ -158,6 +159,22 @@ function makeProjection(visibleTurnItems: OrchestrationV2ProjectedTurnItem[]) {
 }
 
 describe("threadHistoryPaging", () => {
+  it("builds a resumable bounded socket snapshot frame", () => {
+    const projection = makeProjection(Array.from({ length: 90 }, (_, index) => makeRow(index)));
+    const item = buildBoundedThreadStreamSnapshot({
+      snapshotSequence: 23,
+      projection,
+    });
+
+    expect(item.kind).toBe("snapshot");
+    expect(item.snapshotSequence).toBe(23);
+    expect(item.projection.visibleTurnItems).toHaveLength(THREAD_HISTORY_PAGE_POLICY.maxItems);
+    expect(item.historyCursor).not.toBeNull();
+    expect(item.hasMoreHistory).toBe(true);
+    expect(item.latestLocalTurnOrdinal).toBe(90);
+    expect(item.payloadBudgetExceeded).toBe(false);
+  });
+
   it("encodes opaque cursors with stable source identity", () => {
     const cursor = encodeThreadHistoryCursor({
       snapshotSequence: 9,

packages/client-runtime/src/state/threads-atoms.test.ts

@@ -2,6 +2,8 @@ import {
   EnvironmentId,
   EventId,
   ORCHESTRATION_V2_WS_METHODS,
+  TurnItemId,
+  type OrchestrationV2ThreadHistoryPage,
   type OrchestrationV2ThreadProjection,
   type OrchestrationV2ThreadDetailSnapshot,
   type OrchestrationV2ThreadStreamItem,
@@ -58,6 +60,7 @@ const SNAPSHOT: OrchestrationV2ThreadDetailSnapshot = { snapshotSequence: 7, pro
 
 const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: {
   readonly snapshot?: OrchestrationV2ThreadDetailSnapshot;
+  readonly snapshotUnavailable?: boolean;
 }) {
   const subscriptions = yield* Queue.unbounded<{
     readonly afterSequence: number | undefined;
@@ -66,7 +69,7 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
   }>();
   const olderLoads = yield* Queue.unbounded<{
     readonly cursor: string | null;
-    readonly response: Deferred.Deferred<void>;
+    readonly response: Deferred.Deferred<OrchestrationV2ThreadHistoryPage>;
     readonly closed: Deferred.Deferred<void>;
   }>();
   const snapshot = options?.snapshot ?? SNAPSHOT;
@@ -150,15 +153,12 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
   );
   const historyHttpClient = HttpClient.make((request, url) =>
     Effect.gen(function* () {
-      const response = yield* Deferred.make<void>();
+      const response = yield* Deferred.make<OrchestrationV2ThreadHistoryPage>();
       const closed = yield* Deferred.make<void>();
       yield* Effect.addFinalizer(() => Deferred.succeed(closed, undefined));
       yield* Queue.offer(olderLoads, { cursor: url.searchParams.get("cursor"), response, closed });
-      yield* Deferred.await(response);
-      return HttpClientResponse.fromWeb(
-        request,
-        new Response('{"items":[],"nextCursor":null,"hasMoreHistory":false}'),
-      );
+      const page = yield* Deferred.await(response);
+      return HttpClientResponse.fromWeb(request, Response.json(page));
     }).pipe(Effect.scoped),
   );
   const runtime = Atom.runtime(
@@ -193,6 +193,9 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
           load: () =>
             Effect.sync(() => {
               httpLoads += 1;
+              if (options?.snapshotUnavailable === true) {
+                return { _tag: "unavailable" as const };
+              }
               return {
                 _tag: "present" as const,
                 snapshot,
@@ -406,4 +409,87 @@ describe("createEnvironmentThreadStateAtoms", () => {
       yield* Fiber.await(retrying);
     }),
   );
+
+  it.effect(
+    "merges older history after an HTTP failure falls back to a bounded socket snapshot",
+    () =>
+      Effect.gen(function* () {
+        const h = yield* makeHarness({ snapshotUnavailable: true });
+        const unmount = h.registry.mount(h.stateAtom);
+        const subscription = yield* Queue.take(h.subscriptions);
+        expect(subscription.afterSequence).toBeUndefined();
+
+        yield* Queue.offer(subscription.events, {
+          kind: "snapshot",
+          snapshotSequence: 12,
+          projection: THREAD,
+          historyCursor: "socket-older-1",
+          hasMoreHistory: true,
+          latestLocalTurnOrdinal: 4,
+          payloadBudgetExceeded: false,
+        });
+        yield* Queue.offer(subscription.events, { kind: "synchronized" });
+        yield* observeState(
+          h.registry,
+          h.stateAtom,
+          (state) => state.status === "live" && state.history.historyCursor === "socket-older-1",
+        );
+
+        const loading = yield* h.loadEarlier().pipe(Effect.forkScoped);
+        const request = yield* Queue.take(h.olderLoads);
+        expect(request.cursor).toBe("socket-older-1");
+        const olderItem = {
+          id: TurnItemId.make("socket-fallback-older-item"),
+          type: "command_execution" as const,
+          threadId: THREAD_ID,
+          runId: null,
+          nodeId: null,
+          providerThreadId: null,
+          providerTurnId: null,
+          nativeItemRef: null,
+          parentItemId: null,
+          ordinal: 1,
+          status: "completed" as const,
+          title: "Earlier command",
+          input: "pwd",
+          output: "/workspace",
+          exitCode: 0,
+          startedAt: THREAD.thread.createdAt,
+          completedAt: THREAD.thread.createdAt,
+          updatedAt: THREAD.thread.createdAt,
+        };
+        yield* Deferred.succeed(request.response, {
+          snapshotSequence: 12,
+          items: [
+            {
+              position: 0,
+              visibility: "local",
+              sourceThreadId: THREAD_ID,
+              sourceItemId: olderItem.id,
+              item: olderItem,
+            },
+          ],
+          nextCursor: null,
+          hasMoreHistory: false,
+        });
+        expect(yield* Fiber.join(loading)).toEqual({ _tag: "loaded" });
+
+        const state = h.registry.get(h.stateAtom);
+        expect(
+          Option.getOrThrow(state.data).visibleTurnItems.map((row) => row.sourceItemId),
+        ).toEqual([olderItem.id]);
+        expect(state.history).toMatchObject({
+          historyCursor: null,
+          hasMoreHistory: false,
+          loading: false,
+          error: null,
+          expanded: true,
+          latestLocalTurnOrdinal: 4,
+        });
+        expect(h.counts().httpLoads).toBe(1);
+
+        unmount();
+        yield* Deferred.await(subscription.closed);
+      }),
+  );
 });

packages/client-runtime/src/state/threads-sync.test.ts

@@ -278,7 +278,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
 const snapshot = (
   projection: OrchestrationV2ThreadProjection,
   snapshotSequence = 1,
-): OrchestrationV2ThreadStreamItem => ({
+): Extract<OrchestrationV2ThreadStreamItem, { readonly kind: "snapshot" }> => ({
   kind: "snapshot",
   snapshotSequence,
   projection,
@@ -1126,6 +1126,54 @@ describe("EnvironmentThreads", () => {
     }),
   );
 
+  it.effect("bounded socket fallback replaces progressive history meta during resume", () =>
+    Effect.gen(function* () {
+      const harness = yield* makeHarness({
+        cached: {
+          ...BASE_PROJECTION,
+          thread: { ...BASE_PROJECTION.thread, title: "Warm progressive" },
+        },
+        cachedHistory: {
+          historyCursor: "stale-cursor",
+          hasMoreHistory: true,
+          latestLocalTurnOrdinal: 3,
+        },
+      });
+
+      yield* Queue.offer(harness.inputs, {
+        ...snapshot(
+          {
+            ...BASE_PROJECTION,
+            thread: { ...BASE_PROJECTION.thread, title: "Bounded resume fallback" },
+          },
+          CACHED_SNAPSHOT_SEQUENCE + 1,
+        ),
+        historyCursor: "replacement-cursor",
+        hasMoreHistory: true,
+        latestLocalTurnOrdinal: 9,
+        payloadBudgetExceeded: false,
+      });
+
+      const state = yield* awaitThreadState(
+        harness.observed,
+        (value) =>
+          value.status === "live" &&
+          Option.getOrNull(value.data)?.thread.title === "Bounded resume fallback" &&
+          value.history.historyCursor === "replacement-cursor",
+      );
+
+      expect(state.history).toMatchObject({
+        historyCursor: "replacement-cursor",
+        hasMoreHistory: true,
+        latestLocalTurnOrdinal: 9,
+        expanded: false,
+        loading: false,
+        error: null,
+      });
+      expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE);
+    }),
+  );
+
   it.effect("live events preserve progressive history meta under atomic setThread", () =>
     Effect.gen(function* () {
       const harness = yield* makeHarness({

packages/contracts/src/orchestrationV2.test.ts

@@ -32,6 +32,7 @@ import {
   OrchestrationV2SubscribeThreadInput,
   OrchestrationV2Subagent,
   OrchestrationV2ThreadProjection,
+  OrchestrationV2ThreadStreamItem,
   OrchestrationV2ThreadShell,
   OrchestrationV2TurnItem,
   OrchestrationV2TurnItemJson,
@@ -66,6 +67,9 @@ const decodeOrchestrationV2Subagent = Schema.decodeUnknownSync(OrchestrationV2Su
 const decodeOrchestrationV2ThreadProjection = Schema.decodeUnknownSync(
   OrchestrationV2ThreadProjection,
 );
+const decodeOrchestrationV2ThreadStreamItem = Schema.decodeUnknownSync(
+  OrchestrationV2ThreadStreamItem,
+);
 const decodeOrchestrationV2ProviderThreadJson = Schema.decodeUnknownSync(
   OrchestrationV2ProviderThreadJson,
 );
@@ -777,6 +781,23 @@ describe("orchestration V2 contracts", () => {
     });
 
     expect(projection.turnItems.map((item) => item.type)).toEqual(["command_execution"]);
+
+    const boundedSnapshot = decodeOrchestrationV2ThreadStreamItem({
+      kind: "snapshot",
+      snapshotSequence: 12,
+      projection,
+      historyCursor: "older-page",
+      hasMoreHistory: true,
+      latestLocalTurnOrdinal: 1,
+      payloadBudgetExceeded: false,
+    });
+    expect(boundedSnapshot).toMatchObject({
+      kind: "snapshot",
+      historyCursor: "older-page",
+      hasMoreHistory: true,
+      latestLocalTurnOrdinal: 1,
+      payloadBudgetExceeded: false,
+    });
   });
 
   it("decodes orchestration lifecycle turn items for compaction, handoff, and fork UI", () => {