fix(server): long agent turns no longer load whole when a thread opens

apps/server/integration/TransferBudgetReport.integration.ts

@@ -11,11 +11,17 @@ import {
   TRANSFER_MEASURED_MCP_RESULT_BYTES,
   TRANSFER_MEASURED_TOOLS,
 } from "./fixtures/transferBudget.ts";
+import { LONG_TURN_WORK_ROWS } from "./TransferBudgetV2Fixture.integration.ts";
 
 export interface WebSocketCatchUpMeasurement extends WebSocketTransferTotals {
   readonly mode: "replay" | "snapshot";
 }
 
+export interface TimelineHttpMeasurement {
+  readonly http: HttpTransferMeasurement;
+  readonly rows: number;
+}
+
 export interface TransferBudgetRun {
   readonly provider: ProviderDriverKind;
   readonly threadSnapshot: HttpTransferMeasurement;
@@ -27,6 +33,9 @@ export interface TransferBudgetRun {
   readonly reconnectShell: WebSocketCatchUpMeasurement;
   readonly measuredTurnSqlStatements: number;
   readonly reconnectSqlStatements: number;
+  /** The compact bounded snapshot web and mobile load when the long-turn thread opens. */
+  readonly longTurnColdOpen: TimelineHttpMeasurement;
+  readonly longTurnHistoryPage: TimelineHttpMeasurement;
 }
 
 interface ProviderTransferBudget {
@@ -50,6 +59,15 @@ const TRANSFER_BUDGETS: Readonly<Record<string, ProviderTransferBudget>> = {
   claudeAgent: TRANSFER_BUDGET,
 };
 
+// Decoded bytes, not wire bytes: phones parse and keep the decoded JSON. About
+// 30% above the observed 92 KB; without row budgets inside user turns the cold
+// open carried all 920 rows (1.07 MB). Kept apart from TRANSFER_BUDGET because
+// the PR comment validates that object's keys.
+const LONG_TURN_BUDGET = {
+  decodedBytes: 120_000,
+  rows: 75,
+};
+
 function totalWireBytes(run: TransferBudgetRun): number {
   return run.threadSnapshot.wireBytes + run.measuredTurnWebSocket.wireBytes;
 }
@@ -144,6 +162,24 @@ function webSocketRows(
   ];
 }
 
+function longTurnRows(
+  provider: ProviderDriverKind,
+  phase: string,
+  measured: TimelineHttpMeasurement,
+): string[] {
+  return [
+    infoRow(provider, phase, "HTTP wire", measured.http.wireBytes),
+    row(
+      provider,
+      phase,
+      "HTTP decoded",
+      measured.http.decodedBodyBytes,
+      LONG_TURN_BUDGET.decodedBytes,
+    ),
+    row(provider, phase, "rows", measured.rows, LONG_TURN_BUDGET.rows, String),
+  ];
+}
+
 export function transferBudgetViolations(runs: ReadonlyArray<TransferBudgetRun>): string[] {
   const violations: string[] = [];
   for (const run of runs) {
@@ -170,6 +206,18 @@ export function transferBudgetViolations(runs: ReadonlyArray<TransferBudgetRun>)
         run.measuredTurnWebSocket.messages,
         budget.measuredTurnWebSocketMessages,
       ],
+      [
+        "long-turn cold open decoded bytes",
+        run.longTurnColdOpen.http.decodedBodyBytes,
+        LONG_TURN_BUDGET.decodedBytes,
+      ],
+      ["long-turn cold open rows", run.longTurnColdOpen.rows, LONG_TURN_BUDGET.rows],
+      [
+        "long-turn history page decoded bytes",
+        run.longTurnHistoryPage.http.decodedBodyBytes,
+        LONG_TURN_BUDGET.decodedBytes,
+      ],
+      ["long-turn history page rows", run.longTurnHistoryPage.rows, LONG_TURN_BUDGET.rows],
     ] as const;
     for (const [metric, observed, maximum] of checks) {
       if (observed > maximum) {
@@ -187,6 +235,7 @@ export function formatTransferBudgetReport(runs: ReadonlyArray<TransferBudgetRun
     "Wire values are thread data bytes read from local HTTP and WebSocket sockets. HTTP includes response headers; WebSocket measurement starts after the resumed thread subscription synchronizes. TCP/IP, TLS framing, and the WebSocket upgrade are excluded. WebSocket permessage-deflate is negotiated.",
     "The measured turn is observed on three sockets at once: one with only the thread subscription (the capped rows), one with only the shell subscription, and a second client holding both. Server egress is the sum of the three. After the turn the second client disconnects and new sockets resubscribe from the cursor it held before the turn, which is the cursor a backgrounded phone would hold. SQL statements are the SQL client's per-statement spans, counted across V2 persistence and the HTTP/WebSocket handlers.",
     `Scenario: ${TRANSFER_HISTORY_TURN_COUNT} historical turns with ${TRANSFER_HISTORY_TOOLS_PER_TURN} command tools and one retained ${formatBytes(TRANSFER_HISTORY_MCP_RESULT_BYTES)} MCP result each, followed by one measured turn with ${TRANSFER_MEASURED_TOOLS} command tools and a retained ${formatBytes(TRANSFER_MEASURED_MCP_RESULT_BYTES)} MCP result. Synthetic V2 domain events exercise persistence, wire projection, and production HTTP/subscription handlers; this does not measure provider adapter ingestion. Payloads contain no user data.`,
+    `The long-turn rows open a second thread through the compact bounded snapshot web and mobile use, then load one older page. Its last user turn holds ${LONG_TURN_WORK_ROWS} rows of reasoning, commands, file changes, MCP calls, progress messages, checkpoints, and subagents.`,
     "",
     "| Provider | Total thread wire | Budget | Result |",
     "| --- | ---: | ---: | --- |",
@@ -264,6 +313,8 @@ export function formatTransferBudgetReport(runs: ReadonlyArray<TransferBudgetRun
         run.reconnectShell,
       ),
       infoRow(run.provider, "reconnect", "SQL statements", run.reconnectSqlStatements, String),
+      ...longTurnRows(run.provider, "long-turn cold open", run.longTurnColdOpen),
+      ...longTurnRows(run.provider, "long-turn history page", run.longTurnHistoryPage),
     );
   }
 

apps/server/integration/TransferBudgetV2Fixture.integration.ts

@@ -1,4 +1,6 @@
 import {
+  CheckpointId,
+  CheckpointScopeId,
   EventId,
   MessageId,
   NodeId,
@@ -9,6 +11,7 @@ import {
   TurnItemId,
   type ProviderDriverKind,
   type OrchestrationV2DomainEvent,
+  type OrchestrationV2TurnItem,
 } from "@t3tools/contracts";
 import * as DateTime from "effect/DateTime";
 import {
@@ -20,15 +23,18 @@ import {
 } from "./fixtures/transferBudget.ts";
 
 export const THREAD_ID = ThreadId.make("transfer-budget-thread");
-export function threadCreated(provider: ProviderDriverKind): OrchestrationV2DomainEvent {
+export function threadCreated(
+  provider: ProviderDriverKind,
+  threadId: ThreadId = THREAD_ID,
+): OrchestrationV2DomainEvent {
   const now = DateTime.makeUnsafe("2026-06-01T00:00:00Z");
   return {
-    id: EventId.make("thread-created"),
+    id: EventId.make(`thread-created:${threadId}`),
     type: "thread.created",
-    threadId: THREAD_ID,
+    threadId,
     occurredAt: now,
     payload: {
-      id: THREAD_ID,
+      id: threadId,
       projectId: ProjectId.make("transfer-project"),
       title: "Transfer budget",
       providerInstanceId: ProviderInstanceId.make(provider),
@@ -41,7 +47,7 @@ export function threadCreated(provider: ProviderDriverKind): OrchestrationV2Doma
       branch: null,
       worktreePath: null,
       activeProviderThreadId: null,
-      lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: THREAD_ID },
+      lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
       forkedFrom: null,
       createdBy: "user",
       creationSource: "web",
@@ -179,3 +185,224 @@ export function turnEvents(
   });
   return events;
 }
+
+export const LONG_TURN_THREAD_ID = ThreadId.make("transfer-budget-long-turn");
+export const LONG_TURN_WORK_ROWS = 900;
+
+/**
+ * Three short user turns, then one agent turn far longer than a cold-open
+ * window, with the item kinds a long agent run produces. One batch per turn.
+ */
+export function longTurnThreadBatches(
+  provider: ProviderDriverKind,
+): ReadonlyArray<ReadonlyArray<OrchestrationV2DomainEvent>> {
+  const threadId = LONG_TURN_THREAD_ID;
+  const providerInstanceId = ProviderInstanceId.make(provider);
+  const modelSelection = {
+    instanceId: providerInstanceId,
+    model: provider === "codex" ? "gpt-5.4" : "claude-sonnet-4-6",
+  };
+  let ordinal = 0;
+  const text = (turn: number, row: number, targetBytes: number) =>
+    diagnosticOutput({ provider, turnIndex: 100 + turn, toolIndex: row, targetBytes });
+
+  const turn = (index: number, workRows: number): OrchestrationV2DomainEvent[] => {
+    const now = DateTime.makeUnsafe(`2026-06-02T00:${String(index).padStart(2, "0")}:00Z`);
+    const runId = RunId.make(`long-turn-run-${index}`);
+    const nodeId = NodeId.make(`long-turn-node-${index}`);
+    const common = {
+      threadId,
+      runId,
+      nodeId,
+      driver: provider,
+      providerInstanceId,
+      occurredAt: now,
+    };
+    const run = {
+      id: runId,
+      threadId,
+      ordinal: index + 1,
+      providerInstanceId,
+      modelSelection,
+      providerThreadId: null,
+      userMessageId: MessageId.make(`long-turn-user-${index}`),
+      rootNodeId: nodeId,
+      activeAttemptId: null,
+      status: "running" as const,
+      requestedAt: now,
+      startedAt: now,
+      completedAt: null,
+      checkpointId: null,
+      contextHandoffId: null,
+    };
+    const base = (row: string, title: string | null) => {
+      ordinal += 1;
+      return {
+        id: TurnItemId.make(`long-turn-item-${index}-${row}`),
+        threadId,
+        runId,
+        nodeId,
+        providerThreadId: null,
+        providerTurnId: null,
+        nativeItemRef: null,
+        parentItemId: null,
+        ordinal,
+        status: "completed" as const,
+        title,
+        startedAt: now,
+        completedAt: now,
+        updatedAt: now,
+      };
+    };
+    const events: OrchestrationV2DomainEvent[] = [
+      { ...common, id: EventId.make(`long-turn-run-${index}`), type: "run.created", payload: run },
+    ];
+    const message = (role: "user" | "assistant", row: string, messageText: string) => {
+      const messageId = MessageId.make(`long-turn-${role}-${index}-${row}`);
+      const createdBy = role === "user" ? ("user" as const) : ("agent" as const);
+      events.push(
+        {
+          ...common,
+          id: EventId.make(`long-turn-message-${index}-${row}`),
+          type: "message.updated",
+          payload: {
+            id: messageId,
+            threadId,
+            runId,
+            nodeId,
+            role,
+            text: messageText,
+            attachments: [],
+            streaming: false,
+            createdAt: now,
+            updatedAt: now,
+            createdBy,
+            creationSource: "web",
+          },
+        },
+        {
+          ...common,
+          id: EventId.make(`long-turn-item-event-${index}-${row}`),
+          type: "turn-item.updated",
+          payload:
+            role === "user"
+              ? {
+                  ...base(row, null),
+                  type: "user_message",
+                  createdBy,
+                  creationSource: "web",
+                  messageId,
+                  inputIntent: "turn_start",
+                  text: messageText,
+                  attachments: [],
+                }
+              : {
+                  ...base(row, null),
+                  type: "assistant_message",
+                  messageId,
+                  text: messageText,
+                  streaming: false,
+                },
+        },
+      );
+    };
+    const workItem = (row: number): OrchestrationV2TurnItem => {
+      if (row % 100 === 50) {
+        return {
+          ...base(String(row), "Checkpoint"),
+          type: "checkpoint",
+          checkpointId: CheckpointId.make(`long-turn-checkpoint-${index}-${row}`),
+          scopeId: CheckpointScopeId.make(`long-turn-scope-${index}`),
+          files: Array.from({ length: 40 }, (_, file) => ({
+            path: `src/feature/module-${row}-${file}.ts`,
+            kind: "modified",
+            additions: file + 1,
+            deletions: file,
+          })),
+        };
+      }
+      if (row % 100 === 99) {
+        return {
+          ...base(String(row), "Subagent"),
+          type: "subagent",
+          subagentId: NodeId.make(`long-turn-subagent-${index}-${row}`),
+          origin: "app_owned",
+          driver: provider,
+          providerInstanceId,
+          childThreadId: null,
+          prompt: text(index, row, 2_048),
+          result: text(index, row + 1, 3_072),
+        };
+      }
+      switch (row % 8) {
+        case 0:
+        case 7:
+          return {
+            ...base(String(row), "Reasoning"),
+            type: "reasoning",
+            text: text(index, row, 1_536),
+            streaming: false,
+          };
+        case 3:
+          return {
+            ...base(String(row), `Edit module ${row}`),
+            type: "file_change",
+            fileName: `src/feature/module-${row}.ts`,
+            additions: 12,
+            deletions: 4,
+            diffStr: text(index, row, 2_048),
+          };
+        case 4:
+          return {
+            ...base(String(row), "MCP search"),
+            type: "dynamic_tool",
+            toolName: "mcp__test__search",
+            input: { query: `module ${row}` },
+            output: { text: text(index, row, 6_144) },
+          };
+        case 5:
+          return {
+            ...base(String(row), null),
+            type: "assistant_message",
+            messageId: MessageId.make(`long-turn-progress-${index}-${row}`),
+            text: text(index, row, 600),
+            streaming: false,
+          };
+        default:
+          return {
+            ...base(String(row), `Run check ${row}`),
+            type: "command_execution",
+            input: `vp test run src/feature/module-${row}.test.ts`,
+            output: text(index, row, 4_096),
+            exitCode: 0,
+          };
+      }
+    };
+
+    message("user", "prompt", `Continue the feature work, step ${index}.`);
+    for (let row = 0; row < workRows; row++) {
+      events.push({
+        ...common,
+        id: EventId.make(`long-turn-work-${index}-${row}`),
+        type: "turn-item.updated",
+        payload: workItem(row),
+      });
+    }
+    message("assistant", "answer", text(index, workRows, 1_200));
+    events.push({
+      ...common,
+      id: EventId.make(`long-turn-complete-${index}`),
+      type: "run.updated",
+      payload: { ...run, status: "completed", completedAt: now },
+    });
+    return events;
+  };
+
+  return [
+    [threadCreated(provider, threadId)],
+    turn(0, 4),
+    turn(1, 4),
+    turn(2, 4),
+    turn(3, LONG_TURN_WORK_ROWS),
+  ];
+}