From c7744ebf58c3f4a0b9d335879d10994e163b1f70 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 24 Oct 2026 18:21:11 -0700 Subject: [PATCH] perf(server): thread opens and live events send each body once Turn items repeated bodies their records already carry: message text, checkpoint file lists, and subagent prompts and results. Opted-in clients now get those repeats blank and marked, and restore them from the record. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/ThreadStream.ts | 14 +- .../orchestration-v2/WireProjection.test.ts | 75 ++++++ .../src/orchestration-v2/WireProjection.ts | 72 ++++++ .../boundedSnapshotTransport.test.ts | 228 +++++++++++++++++- apps/server/src/orchestration-v2/http.ts | 1 + apps/server/src/ws.ts | 38 ++- .../src/state/boundedThreadSnapshotHttp.ts | 2 +- .../state/orchestrationV2Projection.test.ts | 78 ++++++ .../src/state/orchestrationV2Projection.ts | 43 +++- packages/client-runtime/src/state/threads.ts | 1 + packages/contracts/src/environmentHttp.ts | 1 + packages/contracts/src/orchestrationV2.ts | 23 ++ packages/shared/package.json | 4 + .../src/orchestrationV2BoundedSnapshot.ts | 11 +- .../src/orchestrationV2DedupedBodies.test.ts | 196 +++++++++++++++ .../src/orchestrationV2DedupedBodies.ts | 185 ++++++++++++++ 16 files changed, 943 insertions(+), 29 deletions(-) create mode 100644 packages/shared/src/orchestrationV2DedupedBodies.test.ts create mode 100644 packages/shared/src/orchestrationV2DedupedBodies.ts diff --git a/apps/server/src/orchestration-v2/ThreadStream.ts b/apps/server/src/orchestration-v2/ThreadStream.ts index 7983093a3b9e3..9f902cc35fe53 100644 --- a/apps/server/src/orchestration-v2/ThreadStream.ts +++ b/apps/server/src/orchestration-v2/ThreadStream.ts @@ -4,6 +4,7 @@ import type { } from "@t3tools/contracts"; import { omitLocalVisibleTurnItems } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; +import { dedupeSnapshotBodies } from "@t3tools/shared/orchestrationV2DedupedBodies"; import { buildBoundedThreadProjection, @@ -42,13 +43,16 @@ type ThreadSnapshotStreamItem = Extract< export function boundedSnapshotResponseFields(input: { readonly bounded: BoundedProjectionResult; readonly compactTurnItems: boolean; + readonly dedupeBodies: boolean; }) { const { bounded } = input; - const compactProjection = input.compactTurnItems - ? omitLocalVisibleTurnItems(bounded.projection) - : null; + // Dedupe keeps shared items one object, which the compaction's identity check needs. + const projection = input.dedupeBodies + ? dedupeSnapshotBodies(bounded.projection) + : bounded.projection; + const compactProjection = input.compactTurnItems ? omitLocalVisibleTurnItems(projection) : null; return { - projection: compactProjection ?? bounded.projection, + projection: compactProjection ?? projection, historyCursor: bounded.historyCursor, hasMoreHistory: bounded.hasMoreHistory, latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal, @@ -60,6 +64,7 @@ export function buildBoundedThreadStreamSnapshot(input: { readonly snapshotSequence: number; readonly projection: OrchestrationV2ThreadProjection; readonly compactTurnItems?: boolean; + readonly dedupeBodies?: boolean; }): ThreadSnapshotStreamItem { const bounded = buildBoundedThreadProjection({ snapshotSequence: input.snapshotSequence, @@ -71,6 +76,7 @@ export function buildBoundedThreadStreamSnapshot(input: { ...boundedSnapshotResponseFields({ bounded, compactTurnItems: input.compactTurnItems === true, + dedupeBodies: input.dedupeBodies === true, }), }; } diff --git a/apps/server/src/orchestration-v2/WireProjection.test.ts b/apps/server/src/orchestration-v2/WireProjection.test.ts index 17213d37810a1..baf1ea0a3cb77 100644 --- a/apps/server/src/orchestration-v2/WireProjection.test.ts +++ b/apps/server/src/orchestration-v2/WireProjection.test.ts @@ -9,7 +9,9 @@ import { ProviderInstanceId, ThreadId, TurnItemId, + type OrchestrationV2DomainEvent, type OrchestrationV2ThreadProjection, + type OrchestrationV2ThreadStreamItem, OrchestrationV2TurnItem, OrchestrationV2TurnItemJson, } from "@t3tools/contracts"; @@ -17,7 +19,9 @@ import { describe, expect, it } from "@effect/vitest"; import * as DateTime from "effect/DateTime"; import * as Schema from "effect/Schema"; +import { coalesceLiveUpdatedEvents } from "./ThreadLiveEventCoalescer.ts"; import { + makeLiveBodyDeduper, projectTurnItemForWire, projectTurnItemForDetail, projectDomainEventForWire, @@ -302,3 +306,74 @@ describe("orchestration V2 wire projection", () => { expect(projectDomainEventForWire(projected)).toEqual(projected); }); }); + +describe("live body deduper", () => { + const at = base.updatedAt; + const messageId = MessageId.make("message-live"); + const event = ( + sequence: number, + rest: Pick, + ): Extract => ({ + kind: "event", + sequence, + event: { + id: EventId.make(`event-${sequence}`), + threadId: base.threadId, + occurredAt: at, + ...rest, + } as OrchestrationV2DomainEvent, + }); + const message = (sequence: number, text: string) => + event(sequence, { + type: "message.updated", + payload: { + id: messageId, + threadId: base.threadId, + runId: null, + nodeId: null, + role: "assistant", + text, + attachments: [], + streaming: true, + createdBy: "agent", + creationSource: "provider", + createdAt: at, + updatedAt: at, + }, + }); + const reply = (sequence: number, text: string) => + event(sequence, { + type: "turn-item.updated", + payload: { + ...base, + status: "running", + type: "assistant_message", + messageId, + text, + streaming: true, + }, + }); + const payloadOf = (item: OrchestrationV2ThreadStreamItem) => + item.kind === "event" ? item.event.payload : undefined; + + it("blanks a reply that repeats the message delivered before it, after coalescing", () => { + const delivered = coalesceLiveUpdatedEvents([ + message(1, "a"), + reply(2, "a"), + message(3, "ab"), + reply(4, "ab"), + ]).map(makeLiveBodyDeduper()); + + expect( + delivered.map((item) => [item.kind === "event" && item.sequence, payloadOf(item)]), + ).toEqual([ + [3, message(3, "ab").event.payload], + [4, { ...reply(4, "ab").event.payload, text: "", textOmitted: true }], + ]); + }); + + it("ships a reply whole when its message arrives after it", () => { + const first = reply(1, "a"); + expect([first, message(2, "a")].map(makeLiveBodyDeduper())[0]).toBe(first); + }); +}); diff --git a/apps/server/src/orchestration-v2/WireProjection.ts b/apps/server/src/orchestration-v2/WireProjection.ts index 043fbd3435acb..f6f2bfc7800b5 100644 --- a/apps/server/src/orchestration-v2/WireProjection.ts +++ b/apps/server/src/orchestration-v2/WireProjection.ts @@ -1,9 +1,19 @@ import type { + CheckpointId, + MessageId, + NodeId, + OrchestrationV2Checkpoint, OrchestrationV2DomainEvent, OrchestrationV2ContextHandoff, + OrchestrationV2Subagent, OrchestrationV2ThreadProjection, + OrchestrationV2ThreadStreamItem, OrchestrationV2TurnItem, } from "@t3tools/contracts"; +import { + dedupeSubagentBodies, + dedupeTurnItemBodies, +} from "@t3tools/shared/orchestrationV2DedupedBodies"; import { compactDynamicToolOutput, omitToolOutputImageData, @@ -245,3 +255,65 @@ export function projectDomainEventForWire( ? { ...event, payload: projectContextHandoffForWire(event.payload) } : event; } + +const LIVE_DEDUPE_MEMORY = 32; + +function remember(memory: Map, key: K, value: V): void { + memory.delete(key); + memory.set(key, value); + if (memory.size > LIVE_DEDUPE_MEMORY) memory.delete(memory.keys().next().value as K); +} + +/** + * One per subscription, applied after coalescing so it only remembers records the client + * received, in the snapshot or in events. A record event comes before the turn item that repeats + * its body (adapters and the orchestrator emit them in that order), so the item can ship blank + * and marked; so can a repeated subagent record. Emitting in the other order, or forgetting a + * record, only costs bytes: the repeat ships whole. + */ +export function makeLiveBodyDeduper(): ( + item: OrchestrationV2ThreadStreamItem, +) => OrchestrationV2ThreadStreamItem { + const messageText = new Map(); + const checkpointFiles = new Map(); + const subagents = new Map>(); + const source = { + messageText: (id: MessageId) => messageText.get(id), + checkpointFiles: (id: CheckpointId) => checkpointFiles.get(id), + subagent: (id: NodeId) => subagents.get(id), + }; + const rememberSubagent = ({ id, prompt, result }: OrchestrationV2Subagent) => + remember(subagents, id, { prompt, result }); + return (item) => { + if (item.kind === "snapshot") { + // The client now holds exactly these records; their bodies are never blanked. + const { messages, checkpoints, subagents: records } = item.projection; + for (const message of messages) remember(messageText, message.id, message.text); + for (const checkpoint of checkpoints) + remember(checkpointFiles, checkpoint.id, checkpoint.files); + for (const subagent of records) rememberSubagent(subagent); + return item; + } + if (item.kind !== "event") return item; + const { event } = item; + switch (event.type) { + case "message.updated": + remember(messageText, event.payload.id, event.payload.text); + return item; + case "checkpoint.captured": + remember(checkpointFiles, event.payload.id, event.payload.files); + return item; + case "subagent.updated": { + const payload = dedupeSubagentBodies(event.payload, subagents.get(event.payload.id)); + rememberSubagent(event.payload); + return payload === event.payload ? item : { ...item, event: { ...event, payload } }; + } + case "turn-item.updated": { + const payload = dedupeTurnItemBodies(event.payload, source); + return payload === event.payload ? item : { ...item, event: { ...event, payload } }; + } + default: + return item; + } + }; +} diff --git a/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts b/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts index 9519b661cbf3f..ec22a06a19355 100644 --- a/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts +++ b/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts @@ -2,6 +2,9 @@ import { assert, describe, it } from "@effect/vitest"; import { AuthOrchestrationReadScope, AuthSessionId, + CheckpointId, + CheckpointRef, + CheckpointScopeId, EnvironmentAuthenticatedAuth, EnvironmentAuthenticatedPrincipal, EnvironmentHttpApi, @@ -20,6 +23,8 @@ import { ThreadId, TurnItemId, type OrchestrationV2DomainEvent, + type OrchestrationV2StoredEvent, + type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, } from "@t3tools/contracts"; @@ -59,8 +64,10 @@ 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 BODIES = ThreadId.make("thread:compact-transport:bodies"); const parentRun = RunId.make("run:compact-transport:parent"); const longRun = RunId.make("run:compact-transport:long"); +const bodiesRun = RunId.make("run:compact-transport:bodies"); const store = Layer.mergeAll( ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)), @@ -74,7 +81,8 @@ const management = Layer.unwrap( getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie), getThreadSnapshotWindow: (id, options) => projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie), - streamStoredEventsFrom: () => Stream.empty, + streamStoredEventsFrom: (input) => + input?.threadId === BODIES ? Stream.fromIterable(bodiesLiveTail()) : Stream.empty, }); }), ); @@ -203,6 +211,157 @@ function itemUpdated( }; } +// A conversation turn whose records (message, checkpoint, subagent) repeat in its turn items. +function bodiesEvents(at: DateTime.Utc): OrchestrationV2DomainEvent[] { + const event = (key: string, rest: Pick) => + ({ + id: EventId.make(`event:${BODIES}:${key}`), + threadId: BODIES, + runId: bodiesRun, + occurredAt: at, + ...rest, + }) as OrchestrationV2DomainEvent; + const item = (key: string, ordinal: number, fields: object) => + event(`${key}-item`, { + type: "turn-item.updated", + payload: { + id: TurnItemId.make(`turn-item:${BODIES}:${key}`), + threadId: BODIES, + runId: bodiesRun, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: null, + startedAt: at, + completedAt: at, + updatedAt: at, + ...fields, + } as OrchestrationV2TurnItem, + }); + const message = (key: string, role: "user" | "assistant", text: string) => + event(`${key}-message`, { + type: "message.updated", + payload: { + id: MessageId.make(`message:${key === "prompt" ? bodiesRun : `${BODIES}:${key}`}`), + threadId: BODIES, + runId: bodiesRun, + nodeId: null, + role, + text, + attachments: [], + streaming: false, + createdBy: role === "user" ? "user" : "agent", + creationSource: role === "user" ? "web" : "provider", + createdAt: at, + updatedAt: at, + }, + }); + const files = [{ path: "src/index.ts", kind: "modified", additions: 4, deletions: 2 }]; + return [ + threadCreated(BODIES, at), + runCreated(BODIES, bodiesRun, at), + message("prompt", "user", "Please fix the build."), + item("prompt", 1, { + type: "user_message", + messageId: MessageId.make(`message:${bodiesRun}`), + inputIntent: "turn_start", + text: "Please fix the build.", + attachments: [], + createdBy: "user", + creationSource: "web", + }), + event("task-record", { type: "subagent.updated", payload: bodiesSubagent(at) }), + item("task", 2, { + type: "subagent", + subagentId: bodiesSubagent(at).id, + origin: "app_owned", + driver: "codex", + providerInstanceId: provider, + childThreadId: null, + prompt: bodiesSubagent(at).prompt, + result: bodiesSubagent(at).result, + }), + message("reply", "assistant", "Fixed the import."), + item("reply", 3, { + type: "assistant_message", + messageId: MessageId.make(`message:${BODIES}:reply`), + text: "Fixed the import.", + streaming: false, + }), + event("checkpoint-record", { + type: "checkpoint.captured", + payload: { + id: CheckpointId.make(`checkpoint:${BODIES}`), + threadId: BODIES, + scopeId: CheckpointScopeId.make(`scope:${BODIES}`), + runId: bodiesRun, + nodeId: NodeId.make(`node:${bodiesRun}`), + parentCheckpointId: null, + ordinalWithinScope: 1, + appRunOrdinal: 1, + ref: CheckpointRef.make(`refs/t3/${BODIES}`), + status: "ready", + files, + capturedAt: at, + }, + }), + item("checkpoint", 4, { + type: "checkpoint", + checkpointId: CheckpointId.make(`checkpoint:${BODIES}`), + scopeId: CheckpointScopeId.make(`scope:${BODIES}`), + files, + }), + ]; +} + +function bodiesSubagent(at: DateTime.Utc): OrchestrationV2Subagent { + return { + id: NodeId.make(`node:${BODIES}:task`), + threadId: BODIES, + runId: bodiesRun, + parentNodeId: NodeId.make(`node:${bodiesRun}`), + origin: "app_owned", + createdBy: "agent", + driver: ProviderDriverKind.make("codex"), + providerInstanceId: provider, + providerThreadId: null, + childThreadId: null, + nativeTaskRef: null, + prompt: "Find the failing import.", + title: null, + model: null, + status: "completed", + result: "It was src/index.ts.", + startedAt: at, + completedAt: at, + updatedAt: at, + }; +} + +// Live after the snapshot: a reply pair, then a subagent update that repeats its prompt. +function bodiesLiveTail(): OrchestrationV2StoredEvent[] { + const at = DateTime.makeUnsafe("2026-10-01T00:00:00.000Z"); + const [reply, replyItem] = bodiesEvents(at).filter((event) => + String(event.id).includes(":reply-"), + ); + return [ + reply!, + replyItem!, + { + id: EventId.make(`event:${BODIES}:task-progress`), + type: "subagent.updated", + threadId: BODIES, + runId: bodiesRun, + occurredAt: at, + payload: { ...bodiesSubagent(at), progress: "Checked the build." }, + } as OrchestrationV2DomainEvent, + ].map((event, index) => ({ sequence: 1_000 + index, commandId: null, event })); +} + const seed = Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; const at = yield* DateTime.now; @@ -219,6 +378,7 @@ const seed = Effect.gen(function* () { itemUpdated(LONG, longRun, `tool-${index}`, index + 2, at), ), itemUpdated(LONG, longRun, "result", 92, at, "run_interrupt_result"), + ...bodiesEvents(at), ]; for (const event of events) yield* projections.apply(event); }); @@ -365,4 +525,70 @@ it.layer(TestLayer)("compact bounded snapshot transport", (it) => { }), ); }); + + describe("deduped bodies", () => { + const markers = ["textOmitted", "filesOmitted", "promptOmitted", "resultOmitted"]; + + it.effect( + "HTTP: opted-in snapshots blank repeated bodies and restore to the same snapshot", + () => + withHttp(async (get) => { + const base = `/api/orchestration/threads/${encodeURIComponent(BODIES)}/bounded`; + const compactBody = await (await get(`${base}?compactTurnItems=1`)).text(); + const dedupedBody = await (await get(`${base}?compactTurnItems=1&dedupeBodies=1`)).text(); + + for (const marker of markers) { + assert.notInclude(compactBody, marker); + assert.include(dedupedBody, marker); + } + assert.isBelow(dedupedBody.length, compactBody.length); + assert.deepStrictEqual( + boundedSnapshotProjection(decodeBounded(dedupedBody)), + boundedSnapshotProjection(decodeBounded(compactBody)), + ); + }), + ); + + it.effect( + "WS: bounded and full snapshots restore, and the live tail sends each body once", + () => + Effect.gen(function* () { + const subscribe = ( + input: Partial[0]>, + ) => + subscribeOrchestrationV2Thread({ threadId: BODIES, ...input }).pipe( + Effect.flatMap(Stream.runCollect), + Effect.map((items) => + Array.from(items, (item) => + decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(item)))), + ), + ), + ); + const bounded = { acceptBoundedSnapshot: true, acceptCompactTurnItems: true }; + for (const mode of [bounded, {}]) { + const [plainSnapshot, ...plainTail] = yield* subscribe(mode); + const [dedupedSnapshot, ...dedupedTail] = yield* subscribe({ + ...mode, + acceptDedupedBodies: true, + }); + if (plainSnapshot?.kind !== "snapshot" || dedupedSnapshot?.kind !== "snapshot") { + throw new Error("No snapshot"); + } + assert.deepStrictEqual( + boundedSnapshotProjection(dedupedSnapshot), + boundedSnapshotProjection(plainSnapshot), + ); + const payloads = (tail: typeof plainTail) => + tail.map((item) => (item.kind === "event" ? item.event.payload : null)); + const [reply, replyItem, progress] = payloads(plainTail); + assert.deepStrictEqual(payloads(dedupedTail), [ + reply, + { ...replyItem, text: "", textOmitted: true }, + // The snapshot delivered this subagent's record, so its unchanged bodies repeat. + { ...progress, prompt: "", promptOmitted: true, result: null, resultOmitted: true }, + ]); + } + }), + ); + }); }); diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index cc8e987880df8..cfdf38d304772 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -219,6 +219,7 @@ export const layer = HttpApiBuilder.group( ...boundedSnapshotResponseFields({ bounded, compactTurnItems: args.query.compactTurnItems === "1", + dedupeBodies: args.query.dedupeBodies === "1", }), }; }), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index f9f289ea30a86..8501fdb8eaf26 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -63,6 +63,7 @@ import { OrchestrationV2ThreadLaunchError, type OrchestrationProjectShell, type OrchestrationV2ShellSnapshot, + type OrchestrationV2ThreadStreamItem, type ApplicationStoredEvent, type ProjectEntriesFailure, type ProjectFileFailure, @@ -155,9 +156,11 @@ import { THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, } from "./orchestration-v2/threadHistoryPaging.ts"; import { + makeLiveBodyDeduper, projectDomainEventForWire, projectThreadProjectionForWire, } from "./orchestration-v2/WireProjection.ts"; +import { dedupeSnapshotBodies } from "@t3tools/shared/orchestrationV2DedupedBodies"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; @@ -786,8 +789,13 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs readonly requestCompletionMarker?: boolean; readonly acceptBoundedSnapshot?: boolean; readonly acceptCompactTurnItems?: boolean; + readonly acceptDedupedBodies?: boolean; }) { const threadManagement = yield* ThreadManagementService.ThreadManagementService; + const dedupeBodies = input.acceptDedupedBodies === true; + // Replay, snapshot, and live tail share one deduper: it must see every delivered record. + const withLiveDedupe = (stream: Stream.Stream) => + dedupeBodies ? Stream.suspend(() => Stream.map(stream, makeLiveBodyDeduper())) : stream; const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; yield* Effect.annotateCurrentSpan({ @@ -886,15 +894,21 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs ? buildBoundedThreadStreamSnapshot({ ...snapshot, compactTurnItems: input.acceptCompactTurnItems === true, + dedupeBodies, }) - : { - kind: "snapshot" as const, - snapshotSequence, - projection: projectThreadProjectionForWire(snapshot.projection), - }; - return Stream.concat( - Stream.concat(rpcInitialItems([snapshotItem]), completionMarker), - eventStreamFrom(snapshotSequence), + : (() => { + const projection = projectThreadProjectionForWire(snapshot.projection); + return { + kind: "snapshot" as const, + snapshotSequence, + projection: dedupeBodies ? dedupeSnapshotBodies(projection) : projection, + }; + })(); + return withLiveDedupe( + Stream.concat( + Stream.concat(rpcInitialItems([snapshotItem]), completionMarker), + eventStreamFrom(snapshotSequence), + ), ); }); @@ -965,9 +979,11 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs if (plan.mode === "snapshot") { return yield* snapshotThenLive(); } - return Stream.concat( - Stream.concat(rpcInitialItems(replay), completionMarker), - eventStreamFrom(highWater), + return withLiveDedupe( + Stream.concat( + Stream.concat(rpcInitialItems(replay), completionMarker), + eventStreamFrom(highWater), + ), ); } diff --git a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts index 5a903226a3197..5c57eec1fafc0 100644 --- a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts @@ -30,7 +30,7 @@ export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn( }) { const endpoint = { params: { threadId: input.threadId }, - query: { compactTurnItems: "1" }, + query: { compactTurnItems: "1", dedupeBodies: "1" }, }; return yield* executeAuthenticatedEnvironmentHttpRequest({ ...input, diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts index e1274a36770cb..e0680c17c46f8 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts @@ -638,3 +638,81 @@ it("rechecks visibility when an existing item changes runs without changing ordi ]); expect(projection.visibleTurnItems[0]?.item).toBe(first); }); + +it("restores deduped bodies from the records it already holds, also in a partial timeline", () => { + const messageId = MessageId.make("message-deduped"); + const subagentId = NodeId.make("node-deduped"); + const reply = { + ...commandItem("item-reply", "", 400), + type: "assistant_message", + messageId, + text: "the full reply", + streaming: false, + } as unknown as OrchestrationV2TurnItem; + const subagent = { + id: subagentId, + threadId, + runId, + parentNodeId: NodeId.make("node-root"), + origin: "app_owned", + createdBy: "agent", + driver: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + providerThreadId: null, + childThreadId: null, + nativeTaskRef: null, + prompt: "the task brief", + title: null, + model: null, + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + } as OrchestrationV2ThreadProjection["subagents"][number]; + const projection = { + ...emptyProjection, + runs: [run], + messages: [ + { + id: messageId, + threadId, + runId, + nodeId: null, + role: "assistant", + text: "the full reply", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "provider", + createdAt: now, + updatedAt: now, + }, + ], + subagents: [subagent], + } as OrchestrationV2ThreadProjection; + const event = (type: string, payload: unknown) => + ({ + id: `event-${type}`, + type, + threadId, + runId, + occurredAt: now, + payload, + }) as OrchestrationV2DomainEvent; + + for (const options of [undefined, { partialTimeline: true, latestLocalTurnOrdinal: 300 }]) { + const withReply = applyOrchestrationV2ProjectionEvent( + projection, + event("turn-item.updated", { ...reply, text: "", textOmitted: true }), + options, + ); + expect(withReply?.turnItems).toEqual([reply]); + expect(withReply?.visibleTurnItems.map((row) => row.item)).toEqual([reply]); + } + const progressed = applyOrchestrationV2ProjectionEvent( + projection, + event("subagent.updated", { ...subagent, prompt: "", promptOmitted: true, progress: "step 2" }), + ); + expect(progressed?.subagents).toEqual([{ ...subagent, progress: "step 2" }]); +}); diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index 506afe1394ddb..f501ec2c68bda 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -5,6 +5,11 @@ import type { OrchestrationV2ThreadProjection, OrchestrationV2TurnItem, } from "@t3tools/contracts"; +import { + restoreSubagentBodies, + restoreTurnItemBodies, + type DedupedBodySource, +} from "@t3tools/shared/orchestrationV2DedupedBodies"; import { createOrchestrationV2TurnItemVisibility, isOrchestrationV2TurnItemVisible, @@ -56,6 +61,16 @@ function upsertEntity( return next; } +// A deduped item repeats a record the stream just sent, so search newest first. +function latestBodies(projection: OrchestrationV2ThreadProjection): DedupedBodySource { + return { + messageText: (id) => projection.messages.findLast((message) => message.id === id)?.text, + checkpointFiles: (id) => + projection.checkpoints.findLast((checkpoint) => checkpoint.id === id)?.files, + subagent: (id) => projection.subagents.findLast((subagent) => subagent.id === id), + }; +} + function removeVisibleItem( rows: OrchestrationV2ThreadProjection["visibleTurnItems"], sourceItemId: OrchestrationV2TurnItem["id"], @@ -306,7 +321,16 @@ export function applyOrchestrationV2ProjectionEvent( case "node.updated": return { ...base, nodes: upsertEntity(base.nodes, event.payload) }; case "subagent.updated": - return { ...base, subagents: upsertEntity(base.subagents, event.payload) }; + return { + ...base, + subagents: upsertEntity( + base.subagents, + restoreSubagentBodies( + event.payload, + base.subagents.find((subagent) => subagent.id === event.payload.id), + ), + ), + }; case "provider-session.attached": case "provider-session.updated": return { @@ -345,35 +369,36 @@ export function applyOrchestrationV2ProjectionEvent( case "plan.updated": return { ...base, plans: upsertEntity(base.plans, event.payload) }; case "turn-item.updated": { + const payload = restoreTurnItemBodies(event.payload, latestBodies(projection)); // Streamed updates land on the newest items, so search from the end. let itemIndex = projection.turnItems.length - 1; - while (itemIndex >= 0 && projection.turnItems[itemIndex]!.id !== event.payload.id) { + while (itemIndex >= 0 && projection.turnItems[itemIndex]!.id !== payload.id) { itemIndex -= 1; } if ( partialTimeline && itemIndex === -1 && - shouldDropMissingPartialTurnItem(projection, event.payload, latestLocalTurnOrdinal) + shouldDropMissingPartialTurnItem(projection, payload, latestLocalTurnOrdinal) ) { return projection; } const turnItems = itemIndex === -1 - ? [...projection.turnItems, event.payload] - : replaceProjectionArrayItem(projection.turnItems, itemIndex, event.payload); + ? [...projection.turnItems, payload] + : replaceProjectionArrayItem(projection.turnItems, itemIndex, payload); const next = { ...base, turnItems }; // Only interrupt requests can change another item's visibility. Streaming // text/tool updates must not recheck every row against every run. const previous = itemIndex === -1 ? undefined : projection.turnItems[itemIndex]; const visible = - event.payload.type === "run_interrupt_request" || previous?.type === "run_interrupt_request" + payload.type === "run_interrupt_request" || previous?.type === "run_interrupt_request" ? { ...next, visibleTurnItems: activeVisibleTurnItems(next) } : next; return { ...next, - visibleTurnItems: shouldShowLocalTurnItem(next, event.payload) - ? upsertVisibleTurnItem(visible, event.payload, partialTimeline, latestLocalTurnOrdinal) - : removeVisibleItem(visible.visibleTurnItems, event.payload.id), + visibleTurnItems: shouldShowLocalTurnItem(next, payload) + ? upsertVisibleTurnItem(visible, payload, partialTimeline, latestLocalTurnOrdinal) + : removeVisibleItem(visible.visibleTurnItems, payload.id), }; } case "checkpoint-scope.created": diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 31985507aac1a..9500980377faa 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -963,6 +963,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ...(acceptBoundedSnapshot ? { acceptBoundedSnapshot: true as const, acceptCompactTurnItems: true as const } : {}), + acceptDedupedBodies: true as const, }; }), { diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index e7c057ea0cfa8..24cb3c50b698c 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -669,6 +669,7 @@ const EnvironmentOrchestrationThreadSnapshotParams = Schema.Struct({ const EnvironmentOrchestrationThreadBoundedSnapshotQuery = Schema.Struct({ compactTurnItems: Schema.optionalKey(Schema.String), + dedupeBodies: Schema.optionalKey(Schema.String), }); const EnvironmentOrchestrationThreadHistoryQuery = Schema.Struct({ diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 61b21a97ddc3e..3d6396172d44e 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -753,6 +753,16 @@ export const OrchestrationV2SubagentWorkflow = Schema.Struct({ }); export type OrchestrationV2SubagentWorkflow = typeof OrchestrationV2SubagentWorkflow.Type; +/** + * Wire-only. Set for clients that opted in with `acceptDedupedBodies` or `dedupeBodies=1` when the + * marked field was blanked because the client already holds the same value: in the record it + * received earlier on the same response or stream. Restore with + * `@t3tools/shared/orchestrationV2DedupedBodies` before use; never store or forward a marked + * value. Clients that opted in restore only these fields, so blanking another field needs a new + * opt-in. + */ +const DedupedBodyMarker = Schema.optionalKey(Schema.Literal(true)); + export const OrchestrationV2Subagent = Schema.Struct({ id: NodeId, threadId: ThreadId, @@ -787,6 +797,9 @@ export const OrchestrationV2Subagent = Schema.Struct({ startedAt: Schema.NullOr(Schema.DateTimeUtc), completedAt: Schema.NullOr(Schema.DateTimeUtc), updatedAt: Schema.DateTimeUtc, + /** Same as this subagent's previous record on this stream: its snapshot or `subagent.updated`. */ + promptOmitted: DedupedBodyMarker, + resultOmitted: DedupedBodyMarker, }); export type OrchestrationV2Subagent = typeof OrchestrationV2Subagent.Type; @@ -1425,6 +1438,8 @@ export const OrchestrationV2TurnItem = Schema.Union([ workflowScript: OrchestrationV2ConversationMessage.fields.workflowScript, inputIntent: OrchestrationV2UserMessageInputIntent, text: Schema.String, + /** Same as the text of message `messageId`. */ + textOmitted: DedupedBodyMarker, context: Schema.optional(OrchestrationMessageContext), attachments: Schema.Array(ChatAttachment), }), @@ -1433,6 +1448,7 @@ export const OrchestrationV2TurnItem = Schema.Union([ type: Schema.Literal("assistant_message"), messageId: MessageId, text: Schema.String, + textOmitted: DedupedBodyMarker, attachments: Schema.optional(Schema.Array(ChatAttachment)), streaming: Schema.Boolean, }), @@ -1516,6 +1532,8 @@ export const OrchestrationV2TurnItem = Schema.Union([ checkpointId: CheckpointId, scopeId: CheckpointScopeId, files: Schema.Array(OrchestrationV2CheckpointFileSummary), + /** Same as the files of checkpoint `checkpointId`. */ + filesOmitted: DedupedBodyMarker, }), Schema.Struct({ ...OrchestrationV2TurnItemBaseFields, @@ -1603,6 +1621,9 @@ export const OrchestrationV2TurnItem = Schema.Union([ prompt: Schema.String, progress: Schema.optional(Schema.String), result: Schema.NullOr(Schema.String), + /** Same as the prompt and result of subagent `subagentId`. */ + promptOmitted: DedupedBodyMarker, + resultOmitted: DedupedBodyMarker, }), Schema.Struct({ ...OrchestrationV2TurnItemBaseFields, @@ -3316,6 +3337,8 @@ export const OrchestrationV2SubscribeThreadInput = Schema.Struct({ requestCompletionMarker: Schema.optionalKey(Schema.Boolean), acceptBoundedSnapshot: Schema.optionalKey(Schema.Boolean), acceptCompactTurnItems: Schema.optionalKey(Schema.Boolean), + /** The client restores `*Omitted` bodies from records it already holds. */ + acceptDedupedBodies: Schema.optionalKey(Schema.Boolean), }); export type OrchestrationV2SubscribeThreadInput = typeof OrchestrationV2SubscribeThreadInput.Type; diff --git a/packages/shared/package.json b/packages/shared/package.json index 81f26b66384b2..5d0c5b76d3525 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -379,6 +379,10 @@ "types": "./src/orchestrationV2BoundedSnapshot.ts", "import": "./src/orchestrationV2BoundedSnapshot.ts" }, + "./orchestrationV2DedupedBodies": { + "types": "./src/orchestrationV2DedupedBodies.ts", + "import": "./src/orchestrationV2DedupedBodies.ts" + }, "./orchestrationV2Timeline": { "types": "./src/orchestrationV2Timeline.ts", "import": "./src/orchestrationV2Timeline.ts" diff --git a/packages/shared/src/orchestrationV2BoundedSnapshot.ts b/packages/shared/src/orchestrationV2BoundedSnapshot.ts index 599ccd5117e9c..90b8e8b0932b0 100644 --- a/packages/shared/src/orchestrationV2BoundedSnapshot.ts +++ b/packages/shared/src/orchestrationV2BoundedSnapshot.ts @@ -3,6 +3,8 @@ import type { OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; +import { restoreSnapshotBodies } from "./orchestrationV2DedupedBodies.ts"; + function isLocalTimelineRow( projection: Pick, row: OrchestrationV2ProjectedTurnItem, @@ -39,11 +41,14 @@ export function restoreLocalVisibleTurnItems( return { ...projection, turnItems: [...local, ...projection.turnItems] }; } +/** Undoes both opt-in compactions; bodies carry their own markers, so they need no flag. */ export function boundedSnapshotProjection(snapshot: { readonly projection: OrchestrationV2ThreadProjection; readonly turnItemsOmitLocalVisible?: true | undefined; }): OrchestrationV2ThreadProjection { - return snapshot.turnItemsOmitLocalVisible === true - ? restoreLocalVisibleTurnItems(snapshot.projection) - : snapshot.projection; + return restoreSnapshotBodies( + snapshot.turnItemsOmitLocalVisible === true + ? restoreLocalVisibleTurnItems(snapshot.projection) + : snapshot.projection, + ); } diff --git a/packages/shared/src/orchestrationV2DedupedBodies.test.ts b/packages/shared/src/orchestrationV2DedupedBodies.test.ts new file mode 100644 index 0000000000000..a0f81bbd4d4d9 --- /dev/null +++ b/packages/shared/src/orchestrationV2DedupedBodies.test.ts @@ -0,0 +1,196 @@ +import { + CheckpointId, + CheckpointScopeId, + MessageId, + NodeId, + ThreadId, + type OrchestrationV2ThreadProjection, + TurnItemId, + type OrchestrationV2Subagent, + type OrchestrationV2TurnItem, +} from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { + boundedSnapshotProjection, + omitLocalVisibleTurnItems, +} from "./orchestrationV2BoundedSnapshot.ts"; +import { + dedupeSnapshotBodies, + restoreSnapshotBodies, + restoreSubagentBodies, +} from "./orchestrationV2DedupedBodies.ts"; + +const THREAD = ThreadId.make("thread:bodies"); +const PARENT = ThreadId.make("thread:bodies-parent"); +const at = "2026-10-01T00:00:00.000Z"; + +function item(id: string, fields: Record, threadId = THREAD) { + return { + id: TurnItemId.make(id), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: at, + completedAt: at, + updatedAt: at, + ...fields, + } as unknown as OrchestrationV2TurnItem; +} + +const message = (id: string, role: "user" | "assistant", text: string) => ({ + id: MessageId.make(id), + threadId: THREAD, + runId: null, + nodeId: null, + role, + text, + attachments: [], + streaming: false, + createdBy: role === "user" ? "user" : "agent", + creationSource: role === "user" ? "web" : "provider", + createdAt: at, + updatedAt: at, +}); +const files = [{ path: "src/a.ts", kind: "modified", additions: 3, deletions: 1 }]; +const subagent = { + id: NodeId.make("node:task"), + prompt: "do the task", + result: "final result", +} as Partial; + +const userMessage = (id: string, messageId: string, text: string) => + item(id, { + type: "user_message", + messageId: MessageId.make(messageId), + inputIntent: "turn_start", + text, + attachments: [], + createdBy: "user", + creationSource: "web", + }); +const assistantMessage = (id: string, messageId: string, text: string, threadId = THREAD) => + item( + id, + { type: "assistant_message", messageId: MessageId.make(messageId), text, streaming: false }, + threadId, + ); +const checkpoint = (id: string, checkpointId: string, itemFiles: typeof files) => + item(id, { + type: "checkpoint", + checkpointId: CheckpointId.make(checkpointId), + scopeId: CheckpointScopeId.make("scope"), + files: itemFiles, + }); + +const local: OrchestrationV2TurnItem[] = [ + userMessage("user", "m-user", "hello"), + assistantMessage("assistant", "m-assistant", "hi there"), + assistantMessage("stale", "m-stale", "older text"), + userMessage("empty", "m-empty", ""), + checkpoint("checkpoint", "c-1", files), + checkpoint("orphan-checkpoint", "c-missing", files), + checkpoint("no-files", "c-empty", []), + item("task", { + type: "subagent", + subagentId: subagent.id, + origin: "app_owned", + driver: "codex", + providerInstanceId: "codex", + childThreadId: null, + prompt: "do the task", + result: "partial result", + }), +]; +// A fork's inherited row may name a message id this thread also has; its record lives elsewhere. +const inherited = assistantMessage("inherited", "m-assistant", "hi there", PARENT); + +const projection = { + thread: { id: THREAD }, + messages: [ + message("m-user", "user", "hello"), + message("m-assistant", "assistant", "hi there"), + message("m-stale", "assistant", "newer text"), + message("m-empty", "user", ""), + ], + checkpoints: [ + { id: CheckpointId.make("c-1"), files }, + { id: CheckpointId.make("c-empty"), files: [] }, + ], + subagents: [subagent], + turnItems: local, + visibleTurnItems: [ + { + position: 0, + visibility: "inherited", + sourceThreadId: PARENT, + sourceItemId: inherited.id, + item: inherited, + }, + ...local.map((value, index) => ({ + position: index + 1, + visibility: "local", + sourceThreadId: THREAD, + sourceItemId: value.id, + item: value, + })), + ], +} as unknown as OrchestrationV2ThreadProjection; + +function markers(value: OrchestrationV2ThreadProjection) { + return Object.fromEntries( + value.visibleTurnItems.flatMap(({ item }) => { + const marked = ["textOmitted", "filesOmitted", "promptOmitted", "resultOmitted"].filter( + (key) => key in item, + ); + return marked.length === 0 ? [] : [[String(item.id), marked]]; + }), + ); +} + +describe("deduped snapshot bodies", () => { + it("blanks only bodies equal to a record and restores them exactly", () => { + const deduped = dedupeSnapshotBodies(projection); + + expect(markers(deduped)).toEqual({ + user: ["textOmitted"], + assistant: ["textOmitted"], + checkpoint: ["filesOmitted"], + task: ["promptOmitted"], + }); + expect(restoreSnapshotBodies(deduped)).toEqual(projection); + expect(restoreSnapshotBodies(projection)).toBe(projection); + }); + + it("keeps items shared with turnItems as one object, so the compact snapshot still applies", () => { + const compact = omitLocalVisibleTurnItems(dedupeSnapshotBodies(projection)); + + expect(compact?.turnItems).toEqual([]); + expect( + boundedSnapshotProjection({ projection: compact!, turnItemsOmitLocalVisible: true }), + ).toEqual(projection); + }); + + it("does not restore a field from a subagent record whose field is itself still marked", () => { + const marked = { + prompt: "", + result: null, + promptOmitted: true as const, + resultOmitted: true as const, + }; + const previous = { prompt: "", result: "done", promptOmitted: true as const }; + + expect(restoreSubagentBodies(marked, previous)).toEqual({ + prompt: "", + result: "done", + promptOmitted: true, + }); + }); +}); diff --git a/packages/shared/src/orchestrationV2DedupedBodies.ts b/packages/shared/src/orchestrationV2DedupedBodies.ts new file mode 100644 index 0000000000000..bf219a066e672 --- /dev/null +++ b/packages/shared/src/orchestrationV2DedupedBodies.ts @@ -0,0 +1,185 @@ +import { + OrchestrationV2CheckpointFileSummary, + type CheckpointId, + type MessageId, + type NodeId, + type OrchestrationV2Checkpoint, + type OrchestrationV2Subagent, + type OrchestrationV2ThreadProjection, + type OrchestrationV2TurnItem, +} from "@t3tools/contracts"; +import * as Schema from "effect/Schema"; + +/** + * A turn item repeats a body its record already carries: message text, checkpoint files, a + * subagent's prompt and result. For clients that opt in, the server blanks the repeat and marks + * the field (`textOmitted`, `filesOmitted`, `promptOmitted`, `resultOmitted`), and the client + * copies the value back from the record it already holds. Only a body equal to the record's is + * blanked, so the restore is exact. + */ +export interface DedupedBodySource { + readonly messageText: (id: MessageId) => string | undefined; + readonly checkpointFiles: (id: CheckpointId) => OrchestrationV2Checkpoint["files"] | undefined; + readonly subagent: (id: NodeId) => WithSubagentBodies | undefined; +} + +type SubagentBodies = Pick; +type WithSubagentBodies = SubagentBodies & { + readonly promptOmitted?: true; + readonly resultOmitted?: true; +}; + +const sameCheckpointFiles = Schema.toEquivalence( + Schema.Array(OrchestrationV2CheckpointFileSummary), +); + +/** For subagent items and for a repeated `subagent.updated`, whose `previous` was sent earlier. */ +export function dedupeSubagentBodies( + value: T, + previous: SubagentBodies | undefined, +): T { + const prompt = value.prompt !== "" && previous?.prompt === value.prompt; + const result = value.result !== null && previous?.result === value.result; + if (!prompt && !result) return value; + return { + ...value, + ...(prompt ? { prompt: "", promptOmitted: true as const } : {}), + ...(result ? { result: null, resultOmitted: true as const } : {}), + }; +} + +/** + * A marked body whose record is missing, or whose record field is itself still marked, stays + * blank and marked; the server only blanks bodies it sent, so either means a server bug. + */ +export function restoreSubagentBodies( + value: T, + previous: WithSubagentBodies | undefined, +): T { + if (previous === undefined) return value; + const prompt = value.promptOmitted === true && previous.promptOmitted !== true; + const result = value.resultOmitted === true && previous.resultOmitted !== true; + if (!prompt && !result) return value; + const { promptOmitted, resultOmitted, ...rest } = value; + // Dropping both optional markers from `T` and keeping the unrestored one leaves a `T`. + return { + ...rest, + ...(prompt ? { prompt: previous.prompt } : promptOmitted === true ? { promptOmitted } : {}), + ...(result ? { result: previous.result } : resultOmitted === true ? { resultOmitted } : {}), + } as T; +} + +export function dedupeTurnItemBodies( + item: OrchestrationV2TurnItem, + source: DedupedBodySource, +): OrchestrationV2TurnItem { + switch (item.type) { + case "user_message": + case "assistant_message": + return item.text !== "" && source.messageText(item.messageId) === item.text + ? { ...item, text: "", textOmitted: true } + : item; + case "checkpoint": { + const files = source.checkpointFiles(item.checkpointId); + return item.files.length > 0 && files !== undefined && sameCheckpointFiles(files, item.files) + ? { ...item, files: [], filesOmitted: true } + : item; + } + case "subagent": + return dedupeSubagentBodies(item, source.subagent(item.subagentId)); + default: + return item; + } +} + +export function restoreTurnItemBodies( + item: OrchestrationV2TurnItem, + source: DedupedBodySource, +): OrchestrationV2TurnItem { + switch (item.type) { + case "user_message": + case "assistant_message": { + const text = item.textOmitted === true ? source.messageText(item.messageId) : undefined; + if (text === undefined) return item; + const { textOmitted: _textOmitted, ...rest } = item; + return { ...rest, text }; + } + case "checkpoint": { + const files = + item.filesOmitted === true ? source.checkpointFiles(item.checkpointId) : undefined; + if (files === undefined) return item; + const { filesOmitted: _filesOmitted, ...rest } = item; + return { ...rest, files }; + } + case "subagent": + return item.promptOmitted === true || item.resultOmitted === true + ? restoreSubagentBodies(item, source.subagent(item.subagentId)) + : item; + default: + return item; + } +} + +/** Builds each index on first lookup, so a snapshot without marked items builds none. */ +function snapshotBodySource(projection: OrchestrationV2ThreadProjection): DedupedBodySource { + let messages: Map | undefined; + let checkpoints: Map | undefined; + let subagents: Map | undefined; + return { + messageText: (id) => + (messages ??= new Map(projection.messages.map((message) => [message.id, message.text]))).get( + id, + ), + checkpointFiles: (id) => + (checkpoints ??= new Map( + projection.checkpoints.map((checkpoint) => [checkpoint.id, checkpoint.files]), + )).get(id), + subagent: (id) => + (subagents ??= new Map(projection.subagents.map((subagent) => [subagent.id, subagent]))).get( + id, + ), + }; +} + +/** + * Inherited rows belong to another thread, whose records this projection lacks. Maps each item + * once, so an item shared by `visibleTurnItems` and `turnItems` stays one object. + */ +function mapLocalTurnItems( + projection: OrchestrationV2ThreadProjection, + map: (item: OrchestrationV2TurnItem) => OrchestrationV2TurnItem, +): OrchestrationV2ThreadProjection { + const mapped = new Map(); + let changed = false; + const once = (item: OrchestrationV2TurnItem) => { + let next = mapped.get(item); + if (next === undefined) { + next = map(item); + mapped.set(item, next); + changed ||= next !== item; + } + return next; + }; + const visibleTurnItems = projection.visibleTurnItems.map((row) => { + if (row.visibility !== "local" && row.sourceThreadId !== projection.thread.id) return row; + const item = once(row.item); + return item === row.item ? row : { ...row, item }; + }); + const turnItems = projection.turnItems.map(once); + return changed ? { ...projection, visibleTurnItems, turnItems } : projection; +} + +export function dedupeSnapshotBodies( + projection: OrchestrationV2ThreadProjection, +): OrchestrationV2ThreadProjection { + const source = snapshotBodySource(projection); + return mapLocalTurnItems(projection, (item) => dedupeTurnItemBodies(item, source)); +} + +/** Call on every decoded snapshot before reducers or caches see it. */ +export function restoreSnapshotBodies( + projection: OrchestrationV2ThreadProjection, +): OrchestrationV2ThreadProjection { + const source = snapshotBodySource(projection); + return mapLocalTurnItems(projection, (item) => restoreTurnItemBodies(item, source)); +}