From 28f5516a9acfcfc1e203487e78869a406fd23b76 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 7 Oct 2026 22:30:47 -0700 Subject: [PATCH] perf(mobile): omit duplicated turn items from bounded thread snapshots (#15385) Co-authored-by: Claude Opus 5.5 (1M context) --- .../src/orchestration-v2/ThreadStream.test.ts | 103 +++++ .../src/orchestration-v2/ThreadStream.ts | 40 +- .../boundedSnapshotTransport.test.ts | 370 ++++++++++++++++++ apps/server/src/orchestration-v2/http.ts | 10 +- apps/server/src/ws.ts | 6 +- .../state/boundedThreadSnapshotHttp.test.ts | 77 ++++ .../src/state/boundedThreadSnapshotHttp.ts | 16 +- .../src/state/threads-sync.test.ts | 111 ++++++ packages/client-runtime/src/state/threads.ts | 7 +- packages/contracts/src/environmentHttp.ts | 9 + .../contracts/src/orchestrationV2.test.ts | 83 ++++ packages/contracts/src/orchestrationV2.ts | 13 + packages/shared/package.json | 4 + .../orchestrationV2BoundedSnapshot.test.ts | 147 +++++++ .../src/orchestrationV2BoundedSnapshot.ts | 58 +++ 15 files changed, 1038 insertions(+), 16 deletions(-) create mode 100644 apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts create mode 100644 packages/shared/src/orchestrationV2BoundedSnapshot.test.ts create mode 100644 packages/shared/src/orchestrationV2BoundedSnapshot.ts diff --git a/apps/server/src/orchestration-v2/ThreadStream.test.ts b/apps/server/src/orchestration-v2/ThreadStream.test.ts index 5ef09f776da0f45118f6c4cfeecf0506cae06290..d64de56cbf8aa3edec86c42d6914fb1a261dfc32 100644 --- a/apps/server/src/orchestration-v2/ThreadStream.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStream.test.ts @@ -1,5 +1,6 @@ import { EventId, + OrchestrationV2ThreadStreamItem, ThreadId, TurnItemId, type OrchestrationV2ProjectedTurnItem, @@ -8,6 +9,8 @@ import { } from "@t3tools/contracts"; import { describe, expect, it } from "vite-plus/test"; import * as DateTime from "effect/DateTime"; +import * as Schema from "effect/Schema"; +import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; import { buildBoundedThreadStreamSnapshot, @@ -262,3 +265,103 @@ describe("decideThreadResume", () => { ).toEqual({ mode: "replay", afterSequence: 9, throughSequence: 10 }); }); }); + +describe("compact bounded snapshots", () => { + const FORK_PARENT = ThreadId.make("thread-stream-parent"); + const decodeStreamItem = Schema.decodeUnknownSync( + Schema.toCodecJson(OrchestrationV2ThreadStreamItem), + ); + const encodeStreamItem = Schema.encodeSync(Schema.toCodecJson(OrchestrationV2ThreadStreamItem)); + + /** Fork child: inherited parent rows and marker, then local rows, plus an older interrupt request. */ + function forkProjection(): OrchestrationV2ThreadProjection { + const base = timelineProjection(6); + const local = base.visibleTurnItems.slice(3).map((row, index) => ({ ...row, position: index })); + const inherited = base.visibleTurnItems.slice(0, 3).map((row, index) => { + const item = { ...row.item, id: TurnItemId.make(`parent-${index}`), threadId: FORK_PARENT }; + return { + position: index, + visibility: "inherited" as const, + sourceThreadId: FORK_PARENT, + sourceItemId: item.id, + item, + }; + }); + const request = { + ...base.turnItems[0]!, + id: TurnItemId.make("interrupt-request"), + type: "run_interrupt_request", + ordinal: 0, + message: "Stop", + } as unknown as OrchestrationV2TurnItem; + return { + ...base, + turnItems: [request, ...local.map((row) => row.item)], + visibleTurnItems: [...inherited, ...local].map((row, position) => ({ ...row, position })), + }; + } + + it.each([ + ["linear", () => timelineProjection(80)], + ["fork with retained interrupt request", forkProjection], + ])("restores the exact opted-out snapshot for %s threads", (_, makeProjection) => { + const base = makeProjection(); + // The shared fixture omits a model; the wire codec requires one. + const projection = { + ...base, + thread: { ...base.thread, modelSelection: { instanceId: "codex", model: "gpt-5" } }, + } as OrchestrationV2ThreadProjection; + const full = buildBoundedThreadStreamSnapshot({ snapshotSequence: 44, projection }); + const compact = buildBoundedThreadStreamSnapshot({ + snapshotSequence: 44, + projection, + compactTurnItems: true, + }); + + // Opted-out clients get exactly the previous representation. + expect("turnItemsOmitLocalVisible" in full).toBe(false); + expect(compact.turnItemsOmitLocalVisible).toBe(true); + expect(compact.projection.turnItems.length).toBeLessThan(full.projection.turnItems.length); + const { turnItemsOmitLocalVisible: _marker, ...compactFields } = compact; + expect({ ...compactFields, projection: full.projection }).toEqual(full); + + // Decode the wire form like a client, then restore before use. + const decodedFull = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(full)))); + const decodedCompact = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(compact)))); + if (decodedFull.kind !== "snapshot" || decodedCompact.kind !== "snapshot") { + throw new Error("Expected snapshots"); + } + expect(boundedSnapshotProjection(decodedCompact)).toEqual(decodedFull.projection); + }); + + it("keeps the retained interrupt request outside the visible window", () => { + const compact = buildBoundedThreadStreamSnapshot({ + snapshotSequence: 1, + projection: forkProjection(), + compactTurnItems: true, + }); + expect(compact.projection.turnItems.map((item) => item.id)).toEqual(["interrupt-request"]); + expect(boundedSnapshotProjection(compact).turnItems.map((item) => String(item.id))).toEqual([ + "item-3", + "item-4", + "item-5", + "interrupt-request", + ]); + }); + + it("omits the marker when the window has no local rows", () => { + const fork = forkProjection(); + const inheritedOnly = { + ...fork, + turnItems: fork.turnItems.slice(0, 1), + visibleTurnItems: fork.visibleTurnItems.slice(0, 3), + }; + const compact = buildBoundedThreadStreamSnapshot({ + snapshotSequence: 1, + projection: inheritedOnly, + compactTurnItems: true, + }); + expect("turnItemsOmitLocalVisible" in compact).toBe(false); + expect(compact.projection.turnItems.map((item) => item.id)).toEqual(["interrupt-request"]); + }); +}); diff --git a/apps/server/src/orchestration-v2/ThreadStream.ts b/apps/server/src/orchestration-v2/ThreadStream.ts index ca8e2170c06d9524bcf76e8573876335d7154dd8..fad28abf677aba8c3e4d79e9d62338b2baaf125c 100644 --- a/apps/server/src/orchestration-v2/ThreadStream.ts +++ b/apps/server/src/orchestration-v2/ThreadStream.ts @@ -3,7 +3,12 @@ import type { OrchestrationV2ThreadStreamItem, } from "@t3tools/contracts"; -import { buildBoundedThreadProjection } from "./threadHistoryPaging.ts"; +import { omitLocalVisibleTurnItems } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; + +import { + buildBoundedThreadProjection, + type BoundedProjectionResult, +} from "./threadHistoryPaging.ts"; import { projectThreadProjectionForWire } from "./WireProjection.ts"; /** Maximum number of reducer applications allowed during a thread resume. */ @@ -39,10 +44,34 @@ type ThreadSnapshotStreamItem = Extract< { readonly kind: "snapshot" } >; +/** + * Bounded snapshot fields shared by the HTTP route and socket fallbacks. Only + * clients that opted in get compact `turnItems` and the marker; everyone else + * gets the unchanged representation. + */ +export function boundedSnapshotResponseFields(input: { + readonly bounded: BoundedProjectionResult; + readonly compactTurnItems: boolean; +}) { + const { bounded } = input; + const compactProjection = input.compactTurnItems + ? omitLocalVisibleTurnItems(bounded.projection) + : null; + return { + projection: compactProjection ?? bounded.projection, + historyCursor: bounded.historyCursor, + hasMoreHistory: bounded.hasMoreHistory, + latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal, + payloadBudgetExceeded: bounded.payloadBudgetExceeded, + ...(compactProjection === null ? {} : { turnItemsOmitLocalVisible: true as const }), + }; +} + /** Build the same bounded, wire-projected snapshot for every socket fallback path. */ export function buildBoundedThreadStreamSnapshot(input: { readonly snapshotSequence: number; readonly projection: OrchestrationV2ThreadProjection; + readonly compactTurnItems?: boolean; }): ThreadSnapshotStreamItem { const bounded = buildBoundedThreadProjection({ snapshotSequence: input.snapshotSequence, @@ -51,11 +80,10 @@ export function buildBoundedThreadStreamSnapshot(input: { return { kind: "snapshot", snapshotSequence: input.snapshotSequence, - projection: bounded.projection, - historyCursor: bounded.historyCursor, - hasMoreHistory: bounded.hasMoreHistory, - latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal, - payloadBudgetExceeded: bounded.payloadBudgetExceeded, + ...boundedSnapshotResponseFields({ + bounded, + compactTurnItems: input.compactTurnItems === true, + }), }; } diff --git a/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts b/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..a86b19be95f7ef4da0b118f6e99784156075a74c --- /dev/null +++ b/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts @@ -0,0 +1,370 @@ +import { assert, describe, it } from "@effect/vitest"; +import { + AuthOrchestrationReadScope, + AuthSessionId, + EnvironmentAuthenticatedAuth, + EnvironmentAuthenticatedPrincipal, + EnvironmentHttpApi, + EventId, + MessageId, + NodeId, + ORCHESTRATION_PROTOCOL_HEADER, + ORCHESTRATION_PROTOCOL_VERSION, + OrchestrationV2ThreadBoundedSnapshot, + OrchestrationV2ThreadStreamItem, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + RunId, + ThreadId, + TurnItemId, + type OrchestrationV2DomainEvent, + type OrchestrationV2ThreadProjection, + type OrchestrationV2TurnItem, +} from "@t3tools/contracts"; +import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; +import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as HttpApi from "effect/http-api/HttpApi"; +import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder"; +import { Etag, HttpRouter } from "effect/http"; + +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; +import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts"; +import { subscribeOrchestrationV2Thread } from "../ws.ts"; +import * as OrchestrationHttp from "./http.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ProjectStore from "./ProjectStore.ts"; +import * as ThreadManagementService from "./ThreadManagementService.ts"; + +const decodeBounded = Schema.decodeUnknownSync( + Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadBoundedSnapshot)), +); +const decodeStreamItem = Schema.decodeUnknownSync( + Schema.toCodecJson(OrchestrationV2ThreadStreamItem), +); +const encodeStreamItem = Schema.encodeSync(Schema.toCodecJson(OrchestrationV2ThreadStreamItem)); + +const provider = ProviderInstanceId.make("codex"); +const modelSelection = { instanceId: provider, model: "gpt-5.4" }; +const projectId = ProjectId.make("project:compact-transport"); +const PARENT = ThreadId.make("thread:compact-transport:parent"); +const FORK = ThreadId.make("thread:compact-transport:fork"); +const LONG = ThreadId.make("thread:compact-transport:long"); +const parentRun = RunId.make("run:compact-transport:parent"); +const longRun = RunId.make("run:compact-transport:long"); + +const store = Layer.mergeAll( + ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)), + SqlitePersistence.layerMemory, +); +const management = Layer.unwrap( + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + return Layer.mock(ThreadManagementService.ThreadManagementService)({ + ensureLegacyTranscript: () => Effect.void, + getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie), + getThreadSnapshotWindow: (id, options) => + projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie), + streamStoredEventsFrom: () => Stream.empty, + }); + }), +); + +class OrchestrationApi extends HttpApi.make("environment").add( + EnvironmentHttpApi.groups.orchestration, +) {} +const auth = Layer.succeed(EnvironmentAuthenticatedAuth, (effect) => + effect.pipe( + Effect.provideService(EnvironmentAuthenticatedPrincipal, { + sessionId: AuthSessionId.make("compact-transport"), + subject: "compact-transport", + method: "browser-session-cookie", + scopes: new Set([AuthOrchestrationReadScope]), + }), + ), +); + +function threadCreated(id: ThreadId, at: DateTime.Utc, forkOf?: RunId): OrchestrationV2DomainEvent { + return { + id: EventId.make(`event:${id}:created`), + type: "thread.created", + threadId: id, + occurredAt: at, + payload: { + createdBy: "user", + creationSource: "web", + id, + projectId, + title: String(id), + providerInstanceId: provider, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: + forkOf === undefined + ? { parentThreadId: null, relationshipToParent: null, rootThreadId: id } + : { parentThreadId: PARENT, relationshipToParent: "fork", rootThreadId: PARENT }, + forkedFrom: forkOf === undefined ? null : { type: "run", threadId: PARENT, runId: forkOf }, + createdAt: at, + updatedAt: at, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }; +} + +function runCreated( + threadId: ThreadId, + runId: RunId, + at: DateTime.Utc, +): OrchestrationV2DomainEvent { + const rootNodeId = NodeId.make(`node:${runId}`); + return { + id: EventId.make(`event:${runId}:created`), + type: "run.created", + threadId, + runId, + nodeId: rootNodeId, + driver: ProviderDriverKind.make("codex"), + occurredAt: at, + payload: { + id: runId, + threadId, + ordinal: 1, + providerInstanceId: provider, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:${runId}`), + rootNodeId, + activeAttemptId: null, + status: "completed", + requestedAt: at, + startedAt: at, + completedAt: at, + checkpointId: null, + contextHandoffId: null, + }, + }; +} + +function itemUpdated( + threadId: ThreadId, + runId: RunId | null, + key: string, + ordinal: number, + at: DateTime.Utc, + type: + | "command_execution" + | "run_interrupt_request" + | "run_interrupt_result" = "command_execution", +): OrchestrationV2DomainEvent { + const base = { + id: TurnItemId.make(`turn-item:${threadId}:${key}`), + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed" as const, + title: key, + startedAt: at, + completedAt: at, + updatedAt: at, + }; + const payload: OrchestrationV2TurnItem = + type === "command_execution" + ? { ...base, type, input: `echo ${key}`, output: `output ${key}`, exitCode: 0 } + : { ...base, type, message: key }; + return { + id: EventId.make(`event:${threadId}:${key}`), + type: "turn-item.updated", + threadId, + ...(runId === null ? {} : { runId }), + occurredAt: at, + payload, + }; +} + +const seed = Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const at = yield* DateTime.now; + const events: OrchestrationV2DomainEvent[] = [ + threadCreated(PARENT, at), + runCreated(PARENT, parentRun, at), + ...[1, 2, 3].map((n) => itemUpdated(PARENT, parentRun, `parent-${n}`, n, at)), + threadCreated(FORK, at, parentRun), + ...[1, 2, 3, 4].map((n) => itemUpdated(FORK, null, `fork-${n}`, n, at)), + threadCreated(LONG, at), + runCreated(LONG, longRun, at), + // The request sits far outside the recent window; its result is inside it. + itemUpdated(LONG, longRun, "request", 1, at, "run_interrupt_request"), + ...Array.from({ length: 90 }, (_, index) => + itemUpdated(LONG, longRun, `tool-${index}`, index + 2, at), + ), + itemUpdated(LONG, longRun, "result", 92, at, "run_interrupt_result"), + ]; + for (const event of events) yield* projections.apply(event); +}); + +// Bounded subscriptions without afterSequence never read the event store, and +// thread routes never touch projects. +const TestLayer = Layer.mergeAll( + management, + Layer.effectDiscard(seed), + Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}), + Layer.mock(ProjectStore.ProjectStoreV2)({}), + Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({}), +).pipe(Layer.provideMerge(store)); + +const withHttp = (use: (get: (path: string) => Promise) => Promise) => + Effect.gen(function* () { + const context = yield* Effect.context>(); + const appLayer = HttpApiBuilder.layer(OrchestrationApi).pipe( + Layer.provide(OrchestrationHttp.layer), + Layer.provide(auth), + Layer.provide(Layer.succeedContext(context)), + Layer.provide(NodeHttpPlatform.layer), + Layer.provide(Etag.layerWeak), + Layer.provide(NodeServices.layer), + ); + return yield* Effect.acquireUseRelease( + Effect.sync(() => HttpRouter.toWebHandler(appLayer, { disableLogger: true })), + ({ handler }) => + Effect.promise(() => + use((path) => + handler( + new Request(`http://environment.test${path}`, { + headers: { + [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION), + }, + }), + ), + ), + ), + ({ dispose }) => Effect.promise(() => dispose()), + ); + }); + +const firstSnapshot = (input: Parameters[0]) => + subscribeOrchestrationV2Thread(input).pipe( + Effect.flatMap((stream) => Stream.runHead(stream)), + Effect.map((head) => { + if (head._tag !== "Some" || head.value.kind !== "snapshot") throw new Error("No snapshot"); + // Through the wire codec, as a client receives it. + const decoded = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(head.value)))); + if (decoded.kind !== "snapshot") throw new Error("No snapshot"); + return { raw: head.value, decoded }; + }), + ); + +function itemIds(projection: OrchestrationV2ThreadProjection): string[] { + return projection.turnItems.map((item) => String(item.id)); +} + +it.layer(TestLayer)("compact bounded snapshot transport", (it) => { + describe.each([ + ["fork", FORK], + ["long run with a retained interrupt request", LONG], + ])("%s", (_, threadId) => { + it.effect("HTTP: legacy, ignored and opted-in requests restore to the same snapshot", () => + withHttp(async (get) => { + const base = `/api/orchestration/threads/${encodeURIComponent(threadId)}/bounded`; + const legacyBody = await (await get(base)).text(); + const ignoredBody = await (await get(`${base}?compactTurnItems=yes&unknown=1`)).text(); + const compactResponse = await get(`${base}?compactTurnItems=1`); + assert.strictEqual(compactResponse.status, 200); + const compactBody = await compactResponse.text(); + + // Requests without the exact opt-in get the unchanged representation. + assert.notInclude(legacyBody, "turnItemsOmitLocalVisible"); + assert.strictEqual(ignoredBody, legacyBody); + assert.isBelow(compactBody.length, legacyBody.length); + + const legacy = decodeBounded(legacyBody); + const compact = decodeBounded(compactBody); + assert.isTrue(compact.turnItemsOmitLocalVisible); + const { turnItemsOmitLocalVisible: _marker, projection: _p, ...compactRest } = compact; + const { projection: _legacyProjection, ...legacyRest } = legacy; + assert.deepStrictEqual(compactRest, legacyRest); + assert.deepStrictEqual(boundedSnapshotProjection(compact), legacy.projection); + return { legacy, compact }; + }).pipe( + Effect.tap(({ legacy }) => + Effect.sync(() => { + if (threadId === FORK) { + const visibility = legacy.projection.visibleTurnItems.map((row) => row.visibility); + assert.deepStrictEqual(visibility, [ + "inherited", + "inherited", + "inherited", + "synthetic", + "local", + "local", + "local", + "local", + ]); + } else { + const ids = itemIds(legacy.projection); + assert.isTrue(legacy.hasMoreHistory); + // Window rows first, then the retained request outside the window. + assert.strictEqual(ids.at(-1), `turn-item:${LONG}:request`); + assert.isFalse( + legacy.projection.visibleTurnItems.some((row) => row.sourceItemId === ids.at(-1)), + ); + } + }), + ), + ), + ); + + it.effect("WS: bounded snapshots stay unchanged unless the client opts in", () => + Effect.gen(function* () { + const legacy = yield* firstSnapshot({ threadId, acceptBoundedSnapshot: true }); + const compact = yield* firstSnapshot({ + threadId, + acceptBoundedSnapshot: true, + acceptCompactTurnItems: true, + }); + // Full snapshots ignore the compact opt-in entirely. + const full = yield* firstSnapshot({ threadId, acceptCompactTurnItems: true }); + + assert.isUndefined(legacy.decoded.turnItemsOmitLocalVisible); + assert.isUndefined(full.decoded.turnItemsOmitLocalVisible); + assert.isUndefined(full.decoded.historyCursor); + assert.isTrue(compact.decoded.turnItemsOmitLocalVisible); + assert.isBelow( + compact.decoded.projection.turnItems.length, + legacy.decoded.projection.turnItems.length, + ); + assert.deepStrictEqual( + boundedSnapshotProjection(compact.decoded), + legacy.decoded.projection, + ); + assert.deepStrictEqual( + { + ...compact.decoded, + projection: legacy.decoded.projection, + turnItemsOmitLocalVisible: undefined, + }, + { ...legacy.decoded, turnItemsOmitLocalVisible: undefined }, + ); + }), + ); + }); +}); diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index cd433d3621ad9cbeaa83104a0b5aea9a3ef7446b..e6b769d98c4076f1fa377c60839c95c6a0fb1e22 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -32,6 +32,7 @@ import { import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ProjectStore from "./ProjectStore.ts"; import { buildActiveShellSnapshot } from "./ShellStream.ts"; +import { boundedSnapshotResponseFields } from "./ThreadStream.ts"; import { projectThreadProjectionForWire } from "./WireProjection.ts"; function isThreadNotFound(error: unknown): boolean { @@ -211,11 +212,10 @@ export const layer = HttpApiBuilder.group( }); return { snapshotSequence: snapshot.snapshotSequence, - projection: bounded.projection, - historyCursor: bounded.historyCursor, - hasMoreHistory: bounded.hasMoreHistory, - latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal, - payloadBudgetExceeded: bounded.payloadBudgetExceeded, + ...boundedSnapshotResponseFields({ + bounded, + compactTurnItems: args.query.compactTurnItems === "1", + }), }; }), ) diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 55d5ec85b5a54f0a48b711421a28a46da5a70a10..53bf27800583bfd7b7fa1ab5e94c27b20967e452 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -748,6 +748,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs readonly afterSequence?: number; readonly requestCompletionMarker?: boolean; readonly acceptBoundedSnapshot?: boolean; + readonly acceptCompactTurnItems?: boolean; }) { const threadManagement = yield* ThreadManagementService.ThreadManagementService; const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; @@ -841,7 +842,10 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs ); const { snapshotSequence } = snapshot; const snapshotItem = useBoundedSnapshot - ? buildBoundedThreadStreamSnapshot(snapshot) + ? buildBoundedThreadStreamSnapshot({ + ...snapshot, + compactTurnItems: input.acceptCompactTurnItems === true, + }) : { kind: "snapshot" as const, snapshotSequence, diff --git a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.test.ts b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.test.ts index 48d26b31f70caf8474191e4dfa349b455fe4d995..906f06fb95aa2e703aad065b60709b33def7ed9e 100644 --- a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.test.ts +++ b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.test.ts @@ -76,7 +76,84 @@ const FULL_SNAPSHOT_BODY = { }, }; +const AT = "2026-06-20T00:00:00.000Z"; +const command = (id: string, ordinal: number) => ({ + id, + type: "command_execution", + threadId: String(THREAD_ID), + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: null, + input: id, + exitCode: 0, + startedAt: AT, + completedAt: AT, + updatedAt: AT, +}); +const LOCAL_ITEMS = [command("local-1", 1), command("local-2", 2)]; +const DEPENDENCY = command("older-dependency", 0); +const VISIBLE = LOCAL_ITEMS.map((item, position) => ({ + position, + visibility: "local", + sourceThreadId: String(THREAD_ID), + sourceItemId: item.id, + item, +})); +const boundedBody = (compact: boolean) => ({ + snapshotSequence: 12, + projection: { + ...FULL_SNAPSHOT_BODY.projection, + turnItems: compact ? [DEPENDENCY] : [...LOCAL_ITEMS, DEPENDENCY], + visibleTurnItems: VISIBLE, + }, + historyCursor: "bounded-cursor", + hasMoreHistory: true, + latestLocalTurnOrdinal: 2, + payloadBudgetExceeded: false, + ...(compact ? { turnItemsOmitLocalVisible: true } : {}), +}); + describe("boundedThreadSnapshotLoader", () => { + it.effect.each([ + ["an older server that ignores the opt-in", false], + ["a server that sends compact turnItems", true], + ] as const)("requests compact turnItems and loads %s", ([, compact]) => { + const urls: string[] = []; + const fetchFn = ((input: RequestInfo | URL) => { + urls.push(String(input)); + return Promise.resolve(Response.json(boundedBody(compact))); + }) satisfies typeof fetch; + + return Effect.gen(function* () { + const loader = yield* ThreadSnapshotLoader.ThreadSnapshotLoader; + const result = yield* loader.load(PREPARED, THREAD_ID); + expect(new URL(urls[0]!).searchParams.get("compactTurnItems")).toBe("1"); + expect(result._tag).toBe("present"); + if (result._tag !== "present") return; + expect(result.snapshot.projection.turnItems.map((item) => String(item.id))).toEqual([ + "local-1", + "local-2", + "older-dependency", + ]); + expect("turnItemsOmitLocalVisible" in result.snapshot).toBe(false); + expect(result.history).toEqual({ + historyCursor: "bounded-cursor", + hasMoreHistory: true, + latestLocalTurnOrdinal: 2, + }); + }).pipe( + Effect.provide( + Layer.provide(BoundedThreadSnapshotHttp.layer, RpcHttp.layerRemoteHttpClient(fetchFn)), + ), + ); + }); + it.effect("falls back to full HTTP snapshot when bounded returns a plain route 404", () => { const fetchFn = ((input: RequestInfo | URL) => { const url = String(input); diff --git a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts index ffad41874c94028238515ee0f69090ddd766ca56..a049a6b1486345869f0ccb5799ac37d7595c2051 100644 --- a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts @@ -1,4 +1,5 @@ import type { ThreadId } from "@t3tools/contracts"; +import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -18,7 +19,11 @@ import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts"; // Same cold-open budget as the full snapshot path; bounded payloads should fit. const DEFAULT_BOUNDED_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000; -/** Load a bounded recent-window thread snapshot over HTTP. */ +/** + * Load a bounded recent-window thread snapshot over HTTP. Opts into compact + * turnItems and restores them, so callers always see the full bounded shape. + * Older servers ignore the query and send the full shape. + */ export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn( "clientRuntime.state.fetchEnvironmentBoundedThreadSnapshot", )(function* (input: { @@ -40,9 +45,16 @@ export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn( request: ({ client, headers }) => client.threadBoundedSnapshot({ params: { threadId: input.threadId }, + query: { compactTurnItems: "1" }, headers: withOrchestrationProtocolHeader(headers), }), - }); + }).pipe( + // Drop the marker with the restore so nothing can restore twice. + Effect.map(({ turnItemsOmitLocalVisible, ...snapshot }) => ({ + ...snapshot, + projection: boundedSnapshotProjection({ ...snapshot, turnItemsOmitLocalVisible }), + })), + ); }); /** diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index a832d46b673257b17cea14344408976b81520f3c..d6cbbf6a7f4233c2a13a071940a4d8b1090f84c7 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -122,6 +122,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const lastSubscribeAfterSequence = yield* Ref.make(undefined); const lastRequestCompletionMarker = yield* Ref.make(false); const lastAcceptBoundedSnapshot = yield* Ref.make(undefined); + const lastAcceptCompactTurnItems = yield* Ref.make(undefined); const wakeups = yield* Queue.unbounded(); const savedThreads = yield* Ref.make>([]); const removedThreads = yield* Ref.make>([]); @@ -139,6 +140,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly afterSequence?: number; readonly requestCompletionMarker?: true; readonly acceptBoundedSnapshot?: true; + readonly acceptCompactTurnItems?: true; }) => Stream.unwrap( Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe( @@ -147,6 +149,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker === true), ), Effect.andThen(Ref.set(lastAcceptBoundedSnapshot, input.acceptBoundedSnapshot)), + Effect.andThen(Ref.set(lastAcceptCompactTurnItems, input.acceptCompactTurnItems)), Effect.as(streamFrom(inputs)), ), ), @@ -261,6 +264,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o lastSubscribeAfterSequence, lastRequestCompletionMarker, lastAcceptBoundedSnapshot, + lastAcceptCompactTurnItems, supervisorState, supervisorSession, savedThreads, @@ -842,6 +846,113 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("restores compact socket snapshot turnItems before reducing and caching", () => + Effect.gen(function* () { + const at = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"); + const command = (id: string, ordinal: number, threadId = THREAD_ID) => + ({ + id: TurnItemId.make(id), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed" as const, + title: null, + startedAt: at, + completedAt: at, + updatedAt: at, + type: "command_execution" as const, + input: id, + output: id, + exitCode: 0, + }) satisfies OrchestrationV2TurnItem; + const parentThread = ThreadId.make("thread:compact-parent"); + const inherited = command("parent-item", 1, parentThread); + const local = [command("local-1", 1), command("local-2", 2)]; + const retainedRequest = { + ...command("retained-request", 0), + type: "run_interrupt_request" as const, + message: "Stop", + } as unknown as OrchestrationV2TurnItem; + const visibleTurnItems = [ + { + position: 0, + visibility: "inherited" as const, + sourceThreadId: parentThread, + sourceItemId: inherited.id, + item: inherited, + }, + ...local.map((item, index) => ({ + position: index + 1, + visibility: "local" as const, + sourceThreadId: THREAD_ID, + sourceItemId: item.id, + item, + })), + ]; + const harness = yield* makeHarness(); + yield* Queue.offer(harness.inputs, { + kind: "snapshot", + snapshotSequence: 5, + // Only the dependency that is not a visible local row is on the wire. + projection: { ...BASE_PROJECTION, turnItems: [retainedRequest], visibleTurnItems }, + historyCursor: "compact-cursor", + hasMoreHistory: true, + latestLocalTurnOrdinal: 2, + payloadBudgetExceeded: false, + turnItemsOmitLocalVisible: true, + }); + + const seeded = yield* awaitThreadState( + harness.observed, + (value) => value.history.historyCursor === "compact-cursor", + ); + expect(Option.getOrThrow(seeded.data).turnItems).toEqual([...local, retainedRequest]); + expect(yield* Ref.get(harness.lastAcceptBoundedSnapshot)).toBe(true); + expect(yield* Ref.get(harness.lastAcceptCompactTurnItems)).toBe(true); + + // Live updates reduce against the restored list, not the compact one. + const updated = { ...local[1]!, output: "updated", ordinal: 2 }; + yield* Queue.offer(harness.inputs, { + kind: "event", + sequence: 6, + event: { + id: EventId.make("event-compact-update"), + type: "turn-item.updated", + threadId: THREAD_ID, + occurredAt: at, + payload: updated, + }, + }); + const live = yield* awaitThreadState(harness.observed, (value) => + Option.exists(value.data, (data) => + data.turnItems.some( + (item) => item.type === "command_execution" && item.output === "updated", + ), + ), + ); + expect(Option.getOrThrow(live.data).turnItems.map((item) => String(item.id))).toEqual([ + "local-1", + "local-2", + "retained-request", + ]); + + yield* TestClock.adjust("500 millis"); + yield* Effect.yieldNow; + const saved = (yield* Ref.get(harness.savedThreads)).at(-1); + expect(saved?.projection.turnItems.map((item) => String(item.id))).toEqual([ + "local-1", + "local-2", + "retained-request", + ]); + expect(saved && "turnItemsOmitLocalVisible" in saved).toBe(false); + }), + ); + it.effect("seeds the thread from the HTTP snapshot and resumes live events", () => Effect.gen(function* () { const httpProjection: OrchestrationV2ThreadProjection = { diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 62df7cd30dfb7de9d473d1bb2996593d247df0a7..5c64c41b1d01883e68c0758c4b251048ac3b12a6 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -6,6 +6,7 @@ import { type OrchestrationV2ThreadStreamItem, type ThreadId as ThreadIdType, } from "@t3tools/contracts"; +import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -589,7 +590,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // Bounded socket fallbacks carry their cursor. Legacy-compatible full // snapshots omit these fields and still replace progressive state. yield* setThread( - item.projection, + boundedSnapshotProjection(item), hasProgressiveHistory ? { history: { @@ -919,7 +920,9 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make threadId, ...(canResume ? { afterSequence: sequence } : {}), ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), - ...(acceptBoundedSnapshot ? { acceptBoundedSnapshot: true as const } : {}), + ...(acceptBoundedSnapshot + ? { acceptBoundedSnapshot: true as const, acceptCompactTurnItems: true as const } + : {}), }; }), { diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index a722ba064bd3b45c0aeab7057ba074f11ff688b1..b74affb86216805de7390eb8944031ac6477b0bb 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -584,6 +584,14 @@ const EnvironmentOrchestrationThreadSnapshotParams = Schema.Struct({ threadId: ThreadId, }); +/** + * `compactTurnItems=1` opts into `turnItemsOmitLocalVisible`. Other values are + * ignored, and older servers ignore the whole query. + */ +const EnvironmentOrchestrationThreadBoundedSnapshotQuery = Schema.Struct({ + compactTurnItems: Schema.optionalKey(Schema.String), +}); + const EnvironmentOrchestrationThreadHistoryQuery = Schema.Struct({ cursor: TrimmedNonEmptyString, }); @@ -615,6 +623,7 @@ class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestration") HttpApiEndpoint.get("threadBoundedSnapshot", "/api/orchestration/threads/:threadId/bounded", { headers: OrchestrationProtocolHeaders, params: EnvironmentOrchestrationThreadSnapshotParams, + query: EnvironmentOrchestrationThreadBoundedSnapshotQuery, success: OrchestrationV2ThreadBoundedSnapshot, error: EnvironmentOrchestrationThreadSnapshotErrors, }).middleware(EnvironmentAuthenticatedAuth), diff --git a/packages/contracts/src/orchestrationV2.test.ts b/packages/contracts/src/orchestrationV2.test.ts index e8fd216de980322494981b6fcdc911d498133110..cd185f908d5cef56d9bf10f809edacf2a6b4a2c1 100644 --- a/packages/contracts/src/orchestrationV2.test.ts +++ b/packages/contracts/src/orchestrationV2.test.ts @@ -45,6 +45,48 @@ import { } from "./orchestrationV2.ts"; const now = DateTime.makeUnsafe("2026-04-20T00:00:00.000Z"); +function emptyThreadProjection() { + return { + thread: { + createdBy: "user", + creationSource: "web", + id: "thread-1", + projectId: "project-1", + title: "Thread", + providerInstanceId: "codex", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: "thread-1" }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + deletedAt: null, + }, + runs: [], + attempts: [], + nodes: [], + subagents: [], + providerSessions: [], + providerThreads: [], + providerTurns: [], + runtimeRequests: [], + messages: [], + plans: [], + turnItems: [], + checkpointScopes: [], + checkpoints: [], + contextHandoffs: [], + contextTransfers: [], + visibleTurnItems: [], + updatedAt: now, + }; +} + const LegacyShellStreamItem = Schema.Union([ Schema.Struct({ kind: Schema.Literal("synchronized") }), Schema.Struct({ @@ -326,6 +368,47 @@ describe("orchestration V2 contracts", () => { expect("acceptBoundedSnapshot" in legacyDecoded).toBe(false); }); + it("negotiates compact bounded turnItems without disturbing older peers", () => { + // Older servers strip the unknown opt-in, so they keep sending full turnItems. + const legacyDecoded = decodeLegacySubscribeThreadInput({ + threadId: "thread-1", + acceptBoundedSnapshot: true, + acceptCompactTurnItems: true, + }); + expect("acceptCompactTurnItems" in legacyDecoded).toBe(false); + expect( + decodeOrchestrationV2SubscribeThreadInput({ + threadId: "thread-1", + acceptCompactTurnItems: true, + }).acceptCompactTurnItems, + ).toBe(true); + + const decodeStreamItem = Schema.decodeUnknownSync(OrchestrationV2ThreadStreamItem); + const projection = decodeStreamItem({ + kind: "snapshot", + snapshotSequence: 1, + projection: emptyThreadProjection(), + }); + // Older servers never send the marker; newer clients treat absence as full turnItems. + expect(projection.kind === "snapshot" && projection.turnItemsOmitLocalVisible).toBe(undefined); + const marked = decodeStreamItem({ + kind: "snapshot", + snapshotSequence: 1, + projection: emptyThreadProjection(), + turnItemsOmitLocalVisible: true, + }); + expect(marked.kind === "snapshot" && marked.turnItemsOmitLocalVisible).toBe(true); + // The marker only means "omitted"; any other value is a protocol error. + expect(() => + decodeStreamItem({ + kind: "snapshot", + snapshotSequence: 1, + projection: emptyThreadProjection(), + turnItemsOmitLocalVisible: false, + }), + ).toThrow(); + }); + it("decodes persisted capability snapshots that predate runtimePolicy", () => { // Events written before the field existed must replay; absent decodes to // the weaker client-boundary guarantee so history never overclaims. diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index c26d5a9327e4b85fdfb9dfa2d99e6ebe79451ff8..abc226e356df2d613adb2106a97838931971f7fe 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3244,6 +3244,11 @@ export const OrchestrationV2SubscribeThreadInput = Schema.Struct({ requestCompletionMarker: Schema.optionalKey(Schema.Boolean), /** Allows snapshot fallbacks to contain a bounded, pageable history window. */ acceptBoundedSnapshot: Schema.optionalKey(Schema.Boolean), + /** + * Allows bounded snapshot fallbacks to omit `projection.turnItems` entries + * that repeat local visible rows. See `turnItemsOmitLocalVisible`. + */ + acceptCompactTurnItems: Schema.optionalKey(Schema.Boolean), }); export type OrchestrationV2SubscribeThreadInput = typeof OrchestrationV2SubscribeThreadInput.Type; @@ -3283,6 +3288,12 @@ export const OrchestrationV2ThreadBoundedSnapshot = Schema.Struct({ latestLocalTurnOrdinal: Schema.NullOr(NonNegativeInt), /** True when complete turns or required live control state exceed the usual byte budget. */ payloadBudgetExceeded: Schema.optional(Schema.Boolean), + /** + * Set only for clients that opted in: `projection.turnItems` omits the items + * of local visible rows, which lead the full list. Clients must restore them + * with `boundedSnapshotProjection` before using the projection. + */ + turnItemsOmitLocalVisible: Schema.optionalKey(Schema.Literal(true)), }); export type OrchestrationV2ThreadBoundedSnapshot = typeof OrchestrationV2ThreadBoundedSnapshot.Type; @@ -3363,6 +3374,8 @@ export const OrchestrationV2ThreadStreamItem = Schema.Union([ hasMoreHistory: Schema.optionalKey(Schema.Boolean), latestLocalTurnOrdinal: Schema.optionalKey(Schema.NullOr(NonNegativeInt)), payloadBudgetExceeded: Schema.optionalKey(Schema.Boolean), + /** Same meaning as on `OrchestrationV2ThreadBoundedSnapshot`. */ + turnItemsOmitLocalVisible: Schema.optionalKey(Schema.Literal(true)), }), Schema.Struct({ kind: Schema.Literal("event"), diff --git a/packages/shared/package.json b/packages/shared/package.json index 50cc08129d8e9c1d7c37e252159dbef636774dff..4025994af0765545eecb0eb6606a8b71e6b3f65c 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -359,6 +359,10 @@ "types": "./src/nodeSqliteClient.ts", "import": "./src/nodeSqliteClient.ts" }, + "./orchestrationV2BoundedSnapshot": { + "types": "./src/orchestrationV2BoundedSnapshot.ts", + "import": "./src/orchestrationV2BoundedSnapshot.ts" + }, "./orchestrationV2Timeline": { "types": "./src/orchestrationV2Timeline.ts", "import": "./src/orchestrationV2Timeline.ts" diff --git a/packages/shared/src/orchestrationV2BoundedSnapshot.test.ts b/packages/shared/src/orchestrationV2BoundedSnapshot.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..cf231c0617ef834979d952cc76c7b0f9f2c2d4d5 --- /dev/null +++ b/packages/shared/src/orchestrationV2BoundedSnapshot.test.ts @@ -0,0 +1,147 @@ +import { + RunId, + ThreadId, + TurnItemId, + type OrchestrationV2ProjectedTurnItem, + type OrchestrationV2ThreadProjection, + type OrchestrationV2TurnItem, +} from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { + boundedSnapshotProjection, + omitLocalVisibleTurnItems, + restoreLocalVisibleTurnItems, +} from "./orchestrationV2BoundedSnapshot.ts"; + +const PARENT = ThreadId.make("thread:compact-parent"); +const CHILD = ThreadId.make("thread:compact-child"); +const RUN = RunId.make("run:compact"); + +function item( + threadId: ThreadId, + id: string, + ordinal: number, + type: "command_execution" | "run_interrupt_request" | "fork" = "command_execution", +): OrchestrationV2TurnItem { + return { + id: TurnItemId.make(id), + type, + threadId, + runId: type === "command_execution" ? null : RUN, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: id, + input: id, + exitCode: 0, + startedAt: null, + completedAt: null, + updatedAt: "2026-10-01T00:00:00.000Z", + } as unknown as OrchestrationV2TurnItem; +} + +function row( + visibility: OrchestrationV2ProjectedTurnItem["visibility"], + sourceThreadId: ThreadId, + value: OrchestrationV2TurnItem, +): Omit { + return { visibility, sourceThreadId, sourceItemId: value.id, item: value }; +} + +function projection(input: { + readonly visible: ReadonlyArray>; + readonly turnItems: ReadonlyArray; +}): OrchestrationV2ThreadProjection { + return { + thread: { id: CHILD }, + turnItems: [...input.turnItems], + visibleTurnItems: input.visible.map((value, position) => ({ ...value, position })), + } as unknown as OrchestrationV2ThreadProjection; +} + +/** Fork child: inherited parent rows, synthetic marker, local rows, retained interrupt request. */ +function forkWindow() { + const inherited = [item(PARENT, "parent-1", 1), item(PARENT, "parent-2", 2)]; + const marker = item(PARENT, "fork-marker", 3, "fork"); + const local = [item(CHILD, "child-1", 1), item(CHILD, "child-2", 2), item(CHILD, "child-3", 3)]; + const retainedRequest = item(CHILD, "interrupt-request", 0, "run_interrupt_request"); + return { + local, + retainedRequest, + projection: projection({ + visible: [ + ...inherited.map((value) => row("inherited", PARENT, value)), + row("synthetic", PARENT, marker), + ...local.map((value) => row("local", CHILD, value)), + ], + turnItems: [...local, retainedRequest], + }), + }; +} + +describe("compact bounded snapshot turnItems", () => { + it("omits only local visible items and restores the exact list and order", () => { + const { projection: full, local, retainedRequest } = forkWindow(); + const compact = omitLocalVisibleTurnItems(full); + + expect(compact).not.toBeNull(); + // Inherited and synthetic rows never had turnItems; the retained request stays. + expect(compact!.turnItems).toEqual([retainedRequest]); + expect(compact!.visibleTurnItems).toBe(full.visibleTurnItems); + + const restored = restoreLocalVisibleTurnItems(compact!); + expect(restored.turnItems).toEqual([...local, retainedRequest]); + expect(restored).toEqual(full); + }); + + it("round-trips through JSON as a separately decoded client would see it", () => { + const { projection: full } = forkWindow(); + const wire = JSON.parse(JSON.stringify(omitLocalVisibleTurnItems(full))); + expect( + boundedSnapshotProjection({ projection: wire, turnItemsOmitLocalVisible: true }), + ).toEqual(JSON.parse(JSON.stringify(full))); + }); + + it("treats rows sourced from the thread itself as local", () => { + const own = item(CHILD, "own", 1); + const full = projection({ visible: [row("inherited", CHILD, own)], turnItems: [own] }); + const compact = omitLocalVisibleTurnItems(full); + expect(compact?.turnItems).toEqual([]); + expect(restoreLocalVisibleTurnItems(compact!)).toEqual(full); + }); + + it("refuses shapes it cannot restore exactly", () => { + const [first, second] = [item(CHILD, "a", 1), item(CHILD, "b", 2)]; + const visible = [row("local", CHILD, first), row("local", CHILD, second)]; + // Different order than the visible rows. + expect(omitLocalVisibleTurnItems(projection({ visible, turnItems: [second, first] }))).toBe( + null, + ); + // Equal content but a different object: restoring would not be provably exact. + expect( + omitLocalVisibleTurnItems(projection({ visible, turnItems: [{ ...first }, second] })), + ).toBe(null); + // A local row without its turn item. + expect(omitLocalVisibleTurnItems(projection({ visible, turnItems: [first] }))).toBe(null); + }); + + it("sends nothing compact when no local rows are visible", () => { + const inherited = item(PARENT, "parent", 1); + const request = item(CHILD, "interrupt-request", 0, "run_interrupt_request"); + expect( + omitLocalVisibleTurnItems( + projection({ visible: [row("inherited", PARENT, inherited)], turnItems: [request] }), + ), + ).toBe(null); + }); + + it("leaves unmarked snapshots untouched", () => { + const { projection: full } = forkWindow(); + expect(boundedSnapshotProjection({ projection: full })).toBe(full); + }); +}); diff --git a/packages/shared/src/orchestrationV2BoundedSnapshot.ts b/packages/shared/src/orchestrationV2BoundedSnapshot.ts new file mode 100644 index 0000000000000000000000000000000000000000..117306819cd26766ba4e11fb171d7bac61e1a4de --- /dev/null +++ b/packages/shared/src/orchestrationV2BoundedSnapshot.ts @@ -0,0 +1,58 @@ +import type { + OrchestrationV2ProjectedTurnItem, + OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; + +/** Local timeline rows also carry their item in `projection.turnItems`. */ +function isLocalTimelineRow( + projection: Pick, + row: OrchestrationV2ProjectedTurnItem, +): boolean { + return row.visibility === "local" || row.sourceThreadId === projection.thread.id; +} + +/** + * Server half of the opt-in compact bounded snapshot. Bounded `turnItems` start + * with the items of the window's local visible rows, so those entries can be + * dropped and rebuilt by `restoreLocalVisibleTurnItems`. Returns null when the + * projection does not have that exact shape, or when nothing would be omitted; + * callers then send the unchanged projection without a marker. + */ +export function omitLocalVisibleTurnItems( + projection: OrchestrationV2ThreadProjection, +): OrchestrationV2ThreadProjection | null { + let omitted = 0; + for (const row of projection.visibleTurnItems) { + if (!isLocalTimelineRow(projection, row)) continue; + // Same object, not merely the same id: restoring from the row must + // reproduce the original entry exactly. + if (projection.turnItems[omitted] !== row.item) return null; + omitted += 1; + } + if (omitted === 0) return null; + return { ...projection, turnItems: projection.turnItems.slice(omitted) }; +} + +/** + * Client half: call on a decoded snapshot that carries the + * `turnItemsOmitLocalVisible` marker, before reducers or caches see it. + */ +export function restoreLocalVisibleTurnItems( + projection: OrchestrationV2ThreadProjection, +): OrchestrationV2ThreadProjection { + const local = projection.visibleTurnItems + .filter((row) => isLocalTimelineRow(projection, row)) + .map((row) => row.item); + if (local.length === 0) return projection; + return { ...projection, turnItems: [...local, ...projection.turnItems] }; +} + +/** The usable projection of a decoded bounded snapshot, compact or not. */ +export function boundedSnapshotProjection(snapshot: { + readonly projection: OrchestrationV2ThreadProjection; + readonly turnItemsOmitLocalVisible?: true | undefined; +}): OrchestrationV2ThreadProjection { + return snapshot.turnItemsOmitLocalVisible === true + ? restoreLocalVisibleTurnItems(snapshot.projection) + : snapshot.projection; +} -- 2.54.0