fix(server): trim completed nodes from paged snapshots

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 }));

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<typeof buildBoundedThreadProjection>[0]["projection"];
 }
 
+function makeNode(
+  id: string,
+  options?: Partial<Pick<OrchestrationV2ExecutionNode, "kind" | "status" | "parentNodeId">>,
+): 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<typeof makeProjection>;
+
+    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);

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<OrchestrationV2TurnItem>,
+  visibleTurnItems: ReadonlyArray<OrchestrationV2ProjectedTurnItem>,
+): OrchestrationV2ThreadProjection["nodes"] {
+  const nodesById = new Map(projection.nodes.map((node) => [String(node.id), node]));
+  const retained = new Set<string>();
+  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),