diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 2b588d45ec51..587c831ea039 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/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, diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index 43a52a06cc59..d052eaddeacf 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/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; + readonly response: Deferred.Deferred; readonly closed: Deferred.Deferred; }>(); 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(); + const response = yield* Deferred.make(); const closed = yield* Deferred.make(); 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); + }), + ); }); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 3185f227f672..56ee1184e73d 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/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 => ({ 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({ diff --git a/packages/contracts/src/orchestrationV2.test.ts b/packages/contracts/src/orchestrationV2.test.ts index a141929bad5e..313a0b24df54 100644 --- a/packages/contracts/src/orchestrationV2.test.ts +++ b/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", () => {