feat(server): preserve budgeted history across provider handoffs

apps/server/src/mcp/OrchestratorMcpService.ts

@@ -701,10 +701,13 @@ function turnItemText(item: OrchestrationV2TurnItem): string | null {
 function timelineItem(input: {
   readonly row: OrchestrationV2ThreadProjection["visibleTurnItems"][number];
   readonly maxChars: number;
+  readonly textOffset?: number;
   readonly messagesByThreadId: ReadonlyMap<ThreadId, OrchestrationV2ThreadProjection["messages"]>;
 }): OrchestratorMcpThreadTimelineItem {
   const text = turnItemText(input.row.item);
-  const textTruncated = text !== null && text.length > input.maxChars;
+  const offset = input.textOffset ?? 0;
+  const end = offset + input.maxChars;
+  const textTruncated = text !== null && text.length > end;
   const messageId =
     input.row.item.type === "user_message" || input.row.item.type === "assistant_message"
       ? input.row.item.messageId
@@ -727,8 +730,9 @@ function timelineItem(input: {
     type: input.row.item.type,
     status: input.row.item.status,
     title: input.row.item.title,
-    text: textTruncated ? `${text.slice(0, input.maxChars)}\n…[truncated]` : text,
+    text: text === null ? null : text.slice(offset, end),
     textTruncated,
+    nextTextOffset: textTruncated ? end : null,
     updatedAt: DateTime.formatIso(input.row.item.updatedAt),
   };
 }
@@ -1659,9 +1663,14 @@ const make = Effect.gen(function* () {
         const limit = input.limit ?? DEFAULT_THREAD_READ_LIMIT;
         const maxChars = input.maxCharsPerItem ?? DEFAULT_THREAD_ITEM_MAX_CHARS;
         const matching = target.visibleTurnItems
-          .filter((row) => row.position > afterPosition)
+          .filter((row) =>
+            input.itemId === undefined
+              ? row.position > afterPosition
+              : row.sourceItemId === input.itemId,
+          )
           .filter(
             (row) =>
+              input.itemId !== undefined ||
               view === "activity" ||
               row.item.type === "user_message" ||
               row.item.type === "assistant_message" ||
@@ -1692,6 +1701,7 @@ const make = Effect.gen(function* () {
         const task = directAppOwnedChildTask(parent, target);
         if (
           task !== undefined &&
+          (input.textOffset ?? 0) === 0 &&
           pageIncludesTerminalTaskResult({ parent, page, task, target, maxChars })
         ) {
           yield* readTask(scope, task.id, false, true, "thread-read-acknowledge");
@@ -1702,7 +1712,14 @@ const make = Effect.gen(function* () {
             .toSorted((left, right) => right.ordinal - left.ordinal)
             .slice(0, input.runLimit ?? DEFAULT_THREAD_RUN_LIMIT)
             .map(threadRun),
-          items: page.map((row) => timelineItem({ row, maxChars, messagesByThreadId })),
+          items: page.map((row) =>
+            timelineItem({
+              row,
+              maxChars,
+              messagesByThreadId,
+              ...(input.itemId === undefined ? {} : { textOffset: input.textOffset ?? 0 }),
+            }),
+          ),
           nextPosition: page.at(-1)?.position ?? null,
           hasMore: page.length < matching.length,
         } satisfies OrchestratorMcpThreadReadResult;

apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts

@@ -3,6 +3,7 @@ import { describe, expect, it } from "@effect/vitest";
 import {
   CommandId,
   EnvironmentId,
+  EventId,
   IsoDateTime,
   MessageId,
   type ModelSelection,
@@ -47,6 +48,7 @@ import { McpSchema, McpServer } from "effect/unstable/ai";
 import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts";
 import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts";
 import { CodexOrchestratorReplayHarness } from "../orchestration-v2/Adapters/CodexAdapterV2.testkit.ts";
+import { EventSinkV2 } from "../orchestration-v2/EventSink.ts";
 import { OrchestratorV2, type OrchestratorV2Shape } from "../orchestration-v2/Orchestrator.ts";
 import { layer as threadManagementServiceLayer } from "../orchestration-v2/ThreadManagementService.ts";
 import {
@@ -1009,6 +1011,28 @@ describe("orchestrator MCP toolkit", () => {
               )?.completionDelivery?.state,
             ).toBe("claimed");
 
+            const firstChunk = truncatedResultRead.items[0]!;
+            expect(firstChunk.nextTextOffset).toBe(1);
+            const remainderCall = yield* invoke("t3_thread_read", {
+              threadId: directChildThreadId,
+              itemId: firstChunk.itemId,
+              textOffset: firstChunk.nextTextOffset,
+              maxCharsPerItem: 50_000,
+            });
+            const remainder = yield* decodeThreadReadResult(remainderCall.structuredContent).pipe(
+              Effect.orDie,
+            );
+            expect(remainder.items[0]?.nextTextOffset).toBeNull();
+            expect(firstChunk.text + (remainder.items[0]?.text ?? "")).toBe(
+              "Claude completed: Complete before a parent reads this child result directly.",
+            );
+            // Reading a suffix alone cannot acknowledge a whole child result.
+            expect(
+              (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
+                (task) => task.id === directRead.task.id,
+              )?.completionDelivery?.state,
+            ).toBe("claimed");
+
             const terminalResultReadCall = yield* invoke("t3_thread_read", {
               threadId: directChildThreadId,
               afterPosition: childPromptRead.nextPosition,
@@ -1024,6 +1048,48 @@ describe("orchestrator MCP toolkit", () => {
                 textTruncated: false,
               },
             ]);
+            const longText = "界🧪\n".repeat(20_000) + "FINAL_CONSTRAINT";
+            const childProjection = yield* orchestrator.getThreadProjection(directChildThreadId);
+            const sourceItem = childProjection.turnItems.find(
+              (item) => item.type === "user_message",
+            )!;
+            const oversizedItem = {
+              ...sourceItem,
+              id: TurnItemId.make("item:oversized-retrieval"),
+              type: "assistant_message" as const,
+              text: longText,
+              streaming: false,
+              ordinal: 999,
+            };
+            yield* (yield* EventSinkV2).write({
+              events: [
+                {
+                  id: EventId.make("event:oversized-retrieval"),
+                  type: "turn-item.updated",
+                  threadId: directChildThreadId,
+                  occurredAt: yield* DateTime.now,
+                  payload: oversizedItem,
+                },
+              ],
+            });
+            let recovered = "";
+            let textOffset: number | null = 0;
+            while (textOffset !== null) {
+              const pageCall = yield* invoke("t3_thread_read", {
+                threadId: directChildThreadId,
+                itemId: oversizedItem.id,
+                textOffset,
+                maxCharsPerItem: 50_000,
+              });
+              const page: OrchestratorMcpThreadReadResult = yield* decodeThreadReadResult(
+                pageCall.structuredContent,
+              ).pipe(Effect.orDie);
+              expect(page.items).toHaveLength(1);
+              recovered += page.items[0]!.text;
+              textOffset = page.items[0]!.nextTextOffset ?? null;
+            }
+            expect(recovered).toBe(longText);
+
             const acknowledgedByDirectRead = yield* waitForProjection(
               orchestrator,
               parentThreadId,

apps/server/src/mcp/toolkits/orchestrator/tools.ts

@@ -170,7 +170,7 @@ const ThreadListTool = Tool.make("t3_thread_list", {
 
 const ThreadReadTool = Tool.make("t3_thread_read", {
   description:
-    "Read durable state and a paginated timeline from a T3 thread in the calling project, or from a thread the user attached to this conversation as context. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition.",
+    "Read durable state and a paginated timeline from a T3 thread in the calling project, or from a thread the user attached to this conversation as context. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units.",
   parameters: OrchestratorMcpThreadReadInput,
   success: OrchestratorMcpThreadReadResult,
   failure: OrchestratorMcpFailure,

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts

@@ -89,6 +89,7 @@ import {
 import {
   BUNDLED_CLAUDE_MODEL_CATALOG,
   resolveClaudeCatalogContextWindow,
+  resolveClaudeCatalogContextWindowTokens,
 } from "../../provider/ClaudeModelCatalog.ts";
 import {
   boundProviderEventForLogging,
@@ -5962,6 +5963,8 @@ export function makeClaudeAdapterV2(
           driver: CLAUDE_PROVIDER,
           providerSessionId: input.providerSessionId,
           providerSession: session,
+          getModelContextWindow: (selection) =>
+            resolveClaudeCatalogContextWindowTokens(BUNDLED_CLAUDE_MODEL_CATALOG, selection),
           events: Stream.fromEffectRepeat(Queue.take(events)),
           hasPendingBackgroundWork: Effect.gen(function* () {
             // Session capability: any native thread with pending work pins idle.

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts

@@ -1,3 +1,5 @@
+import { historyResponseItems } from "../ContextHandoffBudget.ts";
+import type { ProviderAdapterV2HistoricalContext } from "../ProviderAdapter.ts";
 import {
   makeProviderTextDeltaCoalescer,
   type ProviderTextDeltaUpdate,
@@ -22,6 +24,7 @@ import {
   RunAttemptId,
   RunId,
   ThreadId,
+  TurnItemId,
 } from "@t3tools/contracts";
 import { assert, describe, it } from "@effect/vitest";
 import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess";
@@ -31,6 +34,7 @@ import * as CodexReplay from "effect-codex-app-server/replay";
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Predicate from "effect/Predicate";
 import * as FileSystem from "effect/FileSystem";
 import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
@@ -57,6 +61,7 @@ import {
 import type { ProviderContinuationRequest } from "../ProviderContinuationRequests.ts";
 import {
   buildCodexTurnStartParams,
+  canReuseCodexContextUsage,
   CODEX_DEFAULT_INSTANCE_ID,
   CODEX_DRIVER_KIND,
   codexBackgroundCommandDetail,
@@ -82,6 +87,33 @@ const encodeReplayTranscriptJson = Schema.encodeEffect(replayTranscriptJson);
 const decodeReplayTranscriptJson = Schema.decodeUnknownEffect(replayTranscriptJson);
 const encodeStringJson = Schema.encodeEffect(Schema.fromJsonString(Schema.String));
 
+describe("Codex context usage compatibility", () => {
+  const previous: ModelSelection = {
+    instanceId: ProviderInstanceId.make("codex"),
+    model: "gpt-6-astra",
+  };
+  it("retains measured usage for reasoning-only changes in either direction", () => {
+    const low: ModelSelection = { ...previous, options: [{ id: "reasoningEffort", value: "low" }] };
+    assert.isTrue(canReuseCodexContextUsage(previous, low));
+    assert.isTrue(canReuseCodexContextUsage(low, previous));
+    assert.isTrue(
+      canReuseCodexContextUsage(low, {
+        ...low,
+        options: [{ id: "reasoningEffort", value: "high" }],
+      }),
+    );
+  });
+  it("invalidates usage for model, instance, context-window and unknown option changes", () => {
+    for (const next of [
+      { ...previous, model: "other-model" },
+      { ...previous, instanceId: ProviderInstanceId.make("other-codex") },
+      { ...previous, options: [{ id: "contextWindow", value: "32k" }] },
+      { ...previous, options: [{ id: "customOption", value: "value" }] },
+    ])
+      assert.isFalse(canReuseCodexContextUsage(previous, next));
+  });
+});
+
 describe("CodexAdapterV2 file change approvals", () => {
   it("uses nonblank reasons before sorted file operations and renamed paths", () => {
     const fileChanges = {
@@ -1540,6 +1572,111 @@ describe("CodexAdapterV2 post-settle continuation", () => {
       };
     });
 
+  for (const response of ["supported", "unsupported", "invalid"] as const) {
+    it.effect(`delivers native history with ${response} app-server protocol`, () =>
+      Effect.gen(function* () {
+        const nativeThreadId = `inject-${response}`;
+        const prompt = "Only the current request";
+        const history: ProviderAdapterV2HistoricalContext = {
+          context: "Historical conversation",
+          messages: (["user", "assistant"] as const).map((role) => ({
+            role,
+            text:
+              role === "user"
+                ? "Original request\n" + "界".repeat(300)
+                : "Partial interrupted work",
+            threadId: ThreadId.make("source"),
+            runId: RunId.make("source-run"),
+            itemId: TurnItemId.make(`source-${role}`),
+            providerThreadId: null,
+            kind: `${role}_message`,
+            status: "interrupted",
+          })),
+        };
+        const items = historyResponseItems(history.messages, history.context);
+        const preamble = codexReplayPreamble({
+          nativeThreadId,
+          nativeTurnId: "current-turn",
+          prompt,
+        });
+        const transcript = makeCodexReplayTranscript({
+          scenario: `inject-${response}`,
+          entries: [
+            ...preamble.slice(0, -3),
+            {
+              type: "expect_outbound",
+              label: "inject",
+              frame: {
+                id: 3,
+                method: "thread/inject_items",
+                params: { threadId: nativeThreadId, items },
+              },
+            },
+            {
+              type: "emit_inbound",
+              label: "inject-result",
+              frame:
+                response === "supported"
+                  ? { id: 3, result: {} }
+                  : {
+                      id: 3,
+                      error: {
+                        code: response === "unsupported" ? -32601 : -32602,
+                        message: "Injection rejected",
+                      },
+                    },
+            },
+            ...(response === "invalid"
+              ? []
+              : preamble
+                  .slice(-3)
+                  .map((entry) =>
+                    "frame" in entry && Predicate.isObject(entry.frame) && "id" in entry.frame
+                      ? { ...entry, frame: { ...entry.frame, id: 4 } }
+                      : entry,
+                  )),
+          ],
+        });
+        const requests: string[] = [];
+        const harness = yield* makeCodexReplayHarness(
+          transcript,
+          () => Effect.void,
+          (method) =>
+            Effect.sync(() => {
+              requests.push(method);
+            }),
+        );
+        const injection = yield* harness.runtime.injectHistory!({
+          providerThread: harness.providerThread,
+          ...history,
+        }).pipe(Effect.result);
+        if (response === "invalid") {
+          assert.equal(injection._tag, "Failure");
+          if (injection._tag === "Failure") {
+            assert.equal(injection.failure._tag, "ProviderAdapterProtocolError");
+            assert.propertyVal(injection.failure.cause, "code", -32602);
+            assert.notProperty(injection.failure, "payload");
+          }
+          assert.notInclude(requests, "turn/start");
+          return;
+        }
+        assert.equal(injection._tag, "Success");
+        if (injection._tag === "Success") assert.equal(injection.success, response === "supported");
+        yield* harness.runtime.startTurn(
+          makeCodexTestTurnInput({
+            threadId: harness.threadId,
+            providerThread: harness.providerThread,
+            now: yield* DateTime.now,
+            attemptId: RunAttemptId.make("inject-attempt"),
+            text: prompt,
+          }),
+        );
+        assert.equal(requests.filter((method) => method === "turn/start").length, 1);
+        assert.isBelow(requests.indexOf("thread/inject_items"), requests.indexOf("turn/start"));
+      }).pipe(Effect.scoped, Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))),
+    );
+  }
+
   it.effect("waits for native start before interrupting an acknowledged queued turn", () =>
     Effect.gen(function* () {
       const nativeThreadId = "early-stop-thread";
@@ -1884,6 +2021,20 @@ describe("CodexAdapterV2 post-settle continuation", () => {
           questionNode?.type === "node.updated" && questionNode.node.countsForRun,
           false,
         );
+        const contextReport = harness.events.find(
+          (event) =>
+            event.type === "provider_turn.updated" && event.providerTurn.tokenUsage !== undefined,
+        );
+        assert.equal(contextReport?.type, "provider_turn.updated");
+        if (contextReport?.type === "provider_turn.updated") {
+          assert.equal(
+            contextReport.providerTurn.runAttemptId,
+            RunAttemptId.make("async-question-attempt"),
+          );
+          assert.equal(contextReport.providerTurn.providerThreadId, harness.providerThread.id);
+          assert.equal(contextReport.providerTurn.tokenUsage?.usedTokens, 15);
+          assert.equal(contextReport.providerTurn.tokenUsage?.maxTokens, 200_000);
+        }
         const completed = harness.events.find(
           (event) =>
             event.type === "provider_turn.updated" && event.providerTurn.status === "completed",