diff --git a/apps/server/src/orchestration-v2/ThreadStream.test.ts b/apps/server/src/orchestration-v2/ThreadStream.test.ts index a5059eada443b..15f35c64b1609 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, + NodeId, OrchestrationV2ThreadStreamItem, ThreadId, TurnItemId, @@ -227,6 +228,66 @@ describe("compact bounded snapshots", () => { ); const encodeStreamItem = Schema.encodeSync(Schema.toCodecJson(OrchestrationV2ThreadStreamItem)); + it.each([false, true])( + "bounds execution-node bytes in encoded snapshots with compactTurnItems=%s", + (compactTurnItems) => { + const base = timelineProjection(2_000); + const turnItems = base.turnItems.map((item, index) => ({ + ...item, + nodeId: NodeId.make(`node-${index}`), + })); + const projection: OrchestrationV2ThreadProjection = { + ...base, + thread: { + ...base.thread, + modelSelection: { instanceId: base.thread.providerInstanceId, model: "gpt-5" }, + }, + turnItems, + visibleTurnItems: base.visibleTurnItems.map((row, index) => ({ + ...row, + item: turnItems[index]!, + })), + nodes: turnItems.map( + (item) => + ({ + id: item.nodeId!, + threadId: THREAD_ID, + runId: null, + parentNodeId: null, + rootNodeId: item.nodeId!, + kind: "tool_call", + status: "completed", + countsForRun: false, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: NOW, + completedAt: NOW, + }) satisfies OrchestrationV2ThreadProjection["nodes"][number], + ), + }; + + const snapshot = buildBoundedThreadStreamSnapshot({ + snapshotSequence: 44, + projection, + compactTurnItems, + }); + const encoded = encodeStreamItem(snapshot); + const rpcChunk = { _tag: "Chunk", requestId: "0", values: [encoded] }; + expect(Buffer.byteLength(JSON.stringify(rpcChunk), "utf8")).toBeLessThan(160_000); + const decoded = decodeStreamItem(JSON.parse(JSON.stringify(encoded))); + if (decoded.kind !== "snapshot") throw new Error("Expected snapshot"); + const restored = boundedSnapshotProjection(decoded); + expect(restored.nodes.map((node) => node.id)).toEqual( + projection.nodes.slice(-75).map((node) => node.id), + ); + expect(restored.visibleTurnItems).toHaveLength(75); + expect(decoded.hasMoreHistory).toBe(true); + }, + ); + function forkProjection(): OrchestrationV2ThreadProjection { const base = timelineProjection(6); const local = base.visibleTurnItems.slice(3).map((row, index) => ({ ...row, position: index })); diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts index 38ad3b5c7de13..4b79b569401ab 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.test.ts @@ -2,8 +2,10 @@ import { MessageId, NodeId, RunId, + RuntimeRequestId, ThreadId, TurnItemId, + type OrchestrationV2ExecutionNode, type OrchestrationV2ProjectedTurnItem, type OrchestrationV2TurnItem, } from "@t3tools/contracts"; @@ -154,6 +156,30 @@ function makeProjection(visibleTurnItems: OrchestrationV2ProjectedTurnItem[]) { } as unknown as Parameters[0]["projection"]; } +function makeNode( + id: string, + options?: Partial>, +): OrchestrationV2ExecutionNode { + return { + id: NodeId.make(id), + threadId: THREAD, + runId: RUN, + parentNodeId: NodeId.make("root-node"), + rootNodeId: NodeId.make("root-node"), + kind: "tool_call", + status: "completed", + countsForRun: false, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: NOW, + completedAt: NOW, + ...options, + }; +} + describe("threadHistoryPaging", () => { it("keeps a human turn whole up to the turn row budget and pages a longer one from inside", () => { const humanTurn = (length: number) => { @@ -440,6 +466,78 @@ describe("threadHistoryPaging", () => { expect(computeLatestLocalTurnOrdinal(full.turnItems)).toBe(40); }); + it("does not resend settled execution leaves outside the shipped history window", () => { + const kinds = ["tool_call", "assistant_message", "reasoning"] as const; + const nodes = Array.from({ length: 2_000 }, (_, index) => + makeNode(`node-${index}`, { kind: kinds[index % kinds.length]! }), + ); + const rows = nodes.map((node, index) => { + const row = makeRow(index); + return { ...row, item: { ...row.item, nodeId: node.id, runId: RUN } }; + }); + const base = makeProjection(rows); + const full = { + ...base, + nodes: [makeNode("root-node", { kind: "root_turn", parentNodeId: null }), ...nodes], + runs: [{ ...base.runs[0]!, id: RUN, status: "running" }], + } satisfies typeof base; + + const bounded = buildBoundedThreadProjection({ + projection: full, + snapshotSequence: 7, + policy: { maxItems: 5, maxEncodedBytes: 1_048_576 }, + }); + + expect(bounded.projection.nodes.map((node) => node.id)).toEqual([ + NodeId.make("root-node"), + ...nodes.slice(-5).map((node) => node.id), + ]); + expect(Buffer.byteLength(JSON.stringify(bounded.projection), "utf8")).toBeLessThan(20_000); + expect(bounded.hasMoreHistory).toBe(true); + expect(full.nodes).toHaveLength(2_001); + }); + + it("keeps live controls and their execution ancestry when their history rows are paged out", () => { + const full = { + ...makeProjection([makeRow(0), makeRow(1)]), + nodes: [ + makeNode("root-node", { kind: "root_turn", parentNodeId: null }), + makeNode("waiting-parent"), + makeNode("waiting-tool", { + status: "waiting", + parentNodeId: NodeId.make("waiting-parent"), + }), + makeNode("request-parent"), + makeNode("requested-tool", { parentNodeId: NodeId.make("request-parent") }), + makeNode("unreferenced-tool", { status: "failed" }), + ], + runtimeRequests: [ + { + id: RuntimeRequestId.make("pending-input"), + nodeId: NodeId.make("requested-tool"), + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { type: "message" }, + createdAt: NOW, + resolvedAt: null, + }, + ], + } satisfies ReturnType; + + const bounded = buildBoundedThreadProjection({ + projection: full, + snapshotSequence: 7, + policy: { maxItems: 1, maxEncodedBytes: 1_048_576 }, + }); + + expect(bounded.projection.nodes.map((node) => node.id)).toEqual( + full.nodes.slice(0, -1).map((node) => node.id), + ); + expect(bounded.projection.runtimeRequests).toEqual(full.runtimeRequests); + }); + it("carries a full-projection watermark when the bounded window is inherited-only", () => { const localOlder = makeRow(0); const localMid = makeRow(1); diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index 579639ca83ab5..530d8a7f262cb 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -461,6 +461,58 @@ function messagesForBoundedProjection( ); } +const settledLeafNodeKinds = new Set(["tool_call", "assistant_message", "reasoning"]); +const settledNodeStatuses = new Set([ + "completed", + "interrupted", + "failed", + "cancelled", + "rolled_back", +]); + +function nodesForBoundedProjection( + projection: OrchestrationV2ThreadProjection, + turnItems: ReadonlyArray, + visibleTurnItems: ReadonlyArray, +): OrchestrationV2ThreadProjection["nodes"] { + const nodesById = new Map(projection.nodes.map((node) => [String(node.id), node])); + const retained = new Set(); + const pending: string[] = []; + const retain = (id: string | null | undefined) => { + if (id == null || retained.has(id)) return; + retained.add(id); + pending.push(id); + }; + + for (const id of [ + ...turnItems.map((item) => item.nodeId), + ...visibleTurnItems.map((row) => row.item.nodeId), + ...projection.runs.map((run) => run.rootNodeId), + ...projection.attempts.map((attempt) => attempt.rootNodeId), + ...projection.subagents.flatMap((agent) => [agent.id, agent.parentNodeId]), + ...projection.providerThreads.map((thread) => thread.ownerNodeId), + ...projection.providerTurns.map((turn) => turn.nodeId), + ...projection.runtimeRequests.map((request) => request.nodeId), + ...projection.checkpointScopes.map((scope) => scope.nodeId), + ...projection.checkpoints.map((checkpoint) => checkpoint.nodeId), + ]) { + retain(id); + } + for (const node of projection.nodes) { + if (!settledLeafNodeKinds.has(node.kind) || !settledNodeStatuses.has(node.status)) { + retain(node.id); + } + } + for (let index = 0; index < pending.length; index += 1) { + const node = nodesById.get(pending[index]!); + if (node === undefined) continue; + retain(node.parentNodeId); + retain(node.rootNodeId); + } + + return projection.nodes.filter((node) => retained.has(node.id)); +} + /** Watermark prevents old missing updates from appending newest when bounded windows lack local rows. */ export function computeLatestLocalTurnOrdinal( turnItems: ReadonlyArray<{ readonly ordinal: number }>, @@ -602,6 +654,7 @@ export function buildBoundedThreadProjection(input: { const projection = { ...controlProjection, + nodes: nodesForBoundedProjection(controlProjection, turnItems, visibleTurnItems), plans, contextHandoffs, messages: messagesForBoundedProjection(controlProjection, turnItems),