fix(server): resume long native threads without handoff budget

apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts

@@ -640,4 +640,32 @@ describe("handoff delivery", () => {
       assert.equal(calls, 0);
     }),
   );
+
+  it.effect("leaves a verified native-thread handoff pending when its wrapper cannot fit", () =>
+    Effect.gen(function* () {
+      let injections = 0;
+      let persists = 0;
+      const result = yield* deliverContextHandoffs({
+        handoffs: [handoff],
+        providerThread,
+        budget: 0,
+        deferIfOverBudget: true,
+        alreadyDeliveredItemIds: new Set(),
+        inject: () =>
+          Effect.sync(() => {
+            injections++;
+            return true;
+          }),
+        persist: () =>
+          Effect.sync(() => {
+            persists++;
+          }),
+      });
+
+      yield* result.delivered;
+      assert.equal(result.context, "");
+      assert.equal(injections, 0);
+      assert.equal(persists, 0);
+    }),
+  );
 });

apps/server/src/orchestration-v2/ContextHandoffDelivery.ts

@@ -19,6 +19,7 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH
     readonly providerThread: OrchestrationV2ProviderThread;
     readonly budget: number | Effect.Effect<number, BudgetError>;
     readonly deferInline?: boolean;
+    readonly deferIfOverBudget?: boolean;
     readonly alreadyDeliveredItemIds: ReadonlySet<string>;
     readonly inject?: (
       history: ProviderAdapter.ProviderAdapterV2HistoricalContext,
@@ -72,6 +73,10 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH
     });
     if (historyCost(selected.messages, selected.context) > budget) {
       if (input.deferInline) return { context: "", delivered: Effect.void, unsent: Effect.void };
+      // The target transcript already has this history, so an oversized wrapper
+      // must not block the turn. Keep the handoff pending for later delivery.
+      if (input.deferIfOverBudget)
+        return { context: "", delivered: Effect.void, unsent: Effect.void };
       return yield* new ContextHandoffBudgetError();
     }
     const omittedItemIds = new Set(selected.omittedItemIds);

apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts

@@ -3,6 +3,7 @@ import { expect, it, vi } from "vite-plus/test";
 import { it as effectIt } from "@effect/vitest";
 import {
   CheckpointScopeId,
+  ContextHandoffId,
   GitCommandError,
   MessageId,
   NodeId,
@@ -15,6 +16,7 @@ import {
   RunAttemptId,
   RunId,
   ThreadId,
+  TurnItemId,
   ProjectId,
   type OrchestrationV2ThreadProjection,
   OrchestrationV2DomainEvent,
@@ -52,6 +54,54 @@ import * as RuntimePolicy from "./RuntimePolicy.ts";
 
 const isDomainEvent = Schema.is(OrchestrationV2DomainEvent);
 
+it("excludes rolled-back runs from native-thread history", () => {
+  const providerThreadId = ProviderThreadId.make("provider_thread_native_history");
+  const keptRunId = RunId.make("run_native_history_kept");
+  const rolledBackRunId = RunId.make("run_native_history_rolled_back");
+  const acceptedAttempts = [keptRunId, rolledBackRunId].map((runId, index) => ({
+    id: RunAttemptId.make(`attempt_native_history_${index}`),
+    runId,
+    providerThreadId,
+    nativeThreadId: "native-thread",
+  }));
+  const nativeInputRunIds = ProviderTurnStart.nativeInputRunIdsForThread({
+    acceptedAttempts,
+    runs: [
+      { id: keptRunId, status: "completed" },
+      { id: rolledBackRunId, status: "rolled_back" },
+    ] as const,
+    providerThreadId,
+    nativeThreadId: "native-thread",
+  });
+
+  expect(nativeInputRunIds).toEqual(new Set([keptRunId]));
+});
+
+it("does not treat runs accepted on multiple native threads as native history", () => {
+  const providerThreadId = ProviderThreadId.make("provider_thread_mixed_native_history");
+  const mixedRunId = RunId.make("run_mixed_native_history");
+  const unknownRunId = RunId.make("run_unknown_native_history");
+  const singleThreadRunId = RunId.make("run_single_native_history");
+  const attempts = [
+    { runId: mixedRunId, providerThreadId, nativeThreadId: "current-native-thread" },
+    { runId: mixedRunId, providerThreadId, nativeThreadId: "older-native-thread" },
+    { runId: unknownRunId, providerThreadId, nativeThreadId: undefined },
+    { runId: unknownRunId, providerThreadId, nativeThreadId: "current-native-thread" },
+    { runId: singleThreadRunId, providerThreadId, nativeThreadId: "current-native-thread" },
+  ];
+
+  expect(ProviderTurnStart.mixedNativeThreadRunIdsForThread(attempts, providerThreadId)).toEqual(
+    new Set([mixedRunId, unknownRunId]),
+  );
+});
+
+it("only treats provider transcript items as native history", () => {
+  expect(ProviderTurnStart.isNativeTranscriptHistoryKind("user_message")).toBe(true);
+  expect(ProviderTurnStart.isNativeTranscriptHistoryKind("command_execution")).toBe(true);
+  expect(ProviderTurnStart.isNativeTranscriptHistoryKind("run_interrupt_result")).toBe(false);
+  expect(ProviderTurnStart.isNativeTranscriptHistoryKind("file_change")).toBe(false);
+});
+
 const managedWorktreesDir = "/tmp/provider-turn-start-worktrees";
 // fs.realpath applies `..` and `.` before following symlinks.
 const normalize = (target: string) => {
@@ -274,6 +324,7 @@ function makeLocalCommandHarness(input: {
   readonly openFailure?: unknown;
   readonly ensureThreadFailure?: unknown;
   readonly historyReadFailureAfterFallback?: unknown;
+  readonly allowRecoveryHandoff?: boolean;
   readonly interruptOpen?: boolean;
   readonly interruptRunBeforeOpenFailure?: boolean;
   readonly writeFailure?: unknown;
@@ -458,6 +509,7 @@ function makeLocalCommandHarness(input: {
     checkpoints: [],
     updatedAt: now,
   };
+  let turnStartHistory: OrchestrationV2ThreadProjection["turnItems"] | undefined;
   if ("historyReadFailureAfterFallback" in input) {
     const nativeThreadRef = {
       driver: providerThread.driver,
@@ -565,6 +617,11 @@ function makeLocalCommandHarness(input: {
       ),
     ),
   );
+  const startedTurns: Array<ProviderAdapter.ProviderAdapterV2TurnInput> = [];
+  const startTurn = vi.fn((turnInput: ProviderAdapter.ProviderAdapterV2TurnInput) => {
+    startedTurns.push(turnInput);
+    return Effect.void;
+  });
   const resumeFallbackSession = {
     driver: providerThread.driver,
     resumeThread: () =>
@@ -615,6 +672,7 @@ function makeLocalCommandHarness(input: {
                   },
                   ensureThread: () => Effect.succeed(providerThread),
                   resumeThread: () => Effect.succeed(projection.providerThreads[0]),
+                  startTurn,
                 } as never)
               : Effect.die("A local command must not open a native session."),
   );
@@ -694,9 +752,31 @@ function makeLocalCommandHarness(input: {
     Layer.provide(
       Layer.mergeAll(
         Layer.mock(ContextHandoffService.ContextHandoffServiceV2)({
-          prepareProviderHandoff: () => Effect.die("history read must fail first"),
+          prepareProviderHandoff: (handoff) =>
+            input.allowRecoveryHandoff
+              ? Effect.succeed({
+                  ...handoff,
+                  id: ContextHandoffId.make("provider-turn-start-recovery-handoff"),
+                  status: "ready",
+                  summaryMessageId: null,
+                  summaryText: "",
+                  history: {
+                    messages: [],
+                    coverage: "",
+                    omittedItems: 0,
+                    omittedItemIds: [],
+                  },
+                  createdByProviderInstanceId: null,
+                  updatedAt: handoff.createdAt,
+                } as never)
+              : Effect.die("history read must fail first"),
+        }),
+        Layer.mock(EventSink.EventSinkV2)({
+          writeIfRunCurrent,
+          ...(input.allowRecoveryHandoff
+            ? { write: () => Effect.succeed({ storedEvents: [] } as never) }
+            : {}),
         }),
-        Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent }),
         IdAllocator.layer,
         layerWorktreeLocation(input.worktreesDir),
         input.fileSystem === undefined
@@ -728,12 +808,14 @@ function makeLocalCommandHarness(input: {
               ),
             }),
           getTurnStartHistory: () =>
-            Effect.fail(
-              new ProjectionStore.ProjectionStoreReadError({
-                threadId,
-                cause: input.historyReadFailureAfterFallback,
-              }),
-            ),
+            turnStartHistory === undefined
+              ? Effect.fail(
+                  new ProjectionStore.ProjectionStoreReadError({
+                    threadId,
+                    cause: input.historyReadFailureAfterFallback,
+                  }),
+                )
+              : Effect.succeed(turnStartHistory),
         }),
         Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
           open,
@@ -765,6 +847,8 @@ function makeLocalCommandHarness(input: {
     resolveRuntimePolicy,
     createWorktree,
     open,
+    startTurn,
+    startedTurns,
     writeIfRunCurrent,
     startRootRun,
     interruptTurn,
@@ -774,6 +858,9 @@ function makeLocalCommandHarness(input: {
     newInstanceId,
     attemptId,
     projection: () => projection,
+    setTurnStartHistory: (items: OrchestrationV2ThreadProjection["turnItems"]) => {
+      turnStartHistory = items;
+    },
     start: Effect.gen(function* () {
       yield* (yield* ProviderTurnStart.ProviderTurnStartServiceV2).start({ threadId, runId });
     }).pipe(Effect.provide(layer)),
@@ -970,6 +1057,131 @@ effectIt.effect("recovers an abandoned failed turn before continuing its native
   }),
 );
 
+effectIt.effect(
+  "recovers an over-budget handoff from mixed native threads before starting the turn",
+  () =>
+    Effect.gen(function* () {
+      const harness = makeLocalCommandHarness({
+        text: "Continue",
+        previousTurn: { runStatus: "failed", turnStatus: "completed" },
+        allowRecoveryHandoff: true,
+      });
+      harness.setTurnStartHistory([]);
+      const projection = harness.projection();
+      const providerThreadId = projection.providerThreads[0]!.id;
+      const currentNativeThreadId = "existing-native-conversation";
+      const previousRunId = RunId.make("previous-failed-run");
+      const currentRunId = projection.runs.at(-1)!.id;
+      const olderAttemptId = RunAttemptId.make("older-native-thread-attempt");
+      const currentAttemptId = RunAttemptId.make("current-native-thread-attempt");
+      const historicalItemId = TurnItemId.make("mixed-native-thread-history-item");
+      const previousTurn = projection.providerTurns[0]!;
+      harness.setProjection({
+        ...projection,
+        providerThreads: projection.providerThreads.map((thread) =>
+          thread.id === providerThreadId
+            ? {
+                ...thread,
+                contextUsage: {
+                  usedTokens: 39_999,
+                  maxTokens: 40_000,
+                  autoCompactThreshold: 40_000,
+                },
+              }
+            : thread,
+        ),
+        attempts: [
+          ...projection.attempts,
+          {
+            ...projection.attempts[0]!,
+            id: olderAttemptId,
+            runId: previousRunId,
+            providerThreadId,
+            nativeThreadId: "older-native-conversation",
+          },
+          {
+            ...projection.attempts[0]!,
+            id: currentAttemptId,
+            runId: previousRunId,
+            providerThreadId,
+            nativeThreadId: currentNativeThreadId,
+          },
+        ],
+        providerTurns: [
+          previousTurn,
+          {
+            ...previousTurn,
+            id: ProviderTurnId.make("older-native-turn"),
+            runAttemptId: olderAttemptId,
+          },
+          {
+            ...previousTurn,
+            id: ProviderTurnId.make("current-native-turn"),
+            runAttemptId: currentAttemptId,
+          },
+        ],
+        contextHandoffs: [
+          {
+            id: ContextHandoffId.make("mixed-native-thread-handoff"),
+            threadId: projection.thread.id,
+            targetRunId: currentRunId,
+            fromProviderThreadIds: [providerThreadId],
+            toProviderThreadId: providerThreadId,
+            coveredRunOrdinals: { from: 1, to: 1 },
+            strategy: "delta_since_target_last_seen",
+            status: "ready",
+            summaryMessageId: null,
+            summaryText: "",
+            history: {
+              messages: [
+                {
+                  role: "user",
+                  text: "Earlier request",
+                  threadId: projection.thread.id,
+                  runId: previousRunId,
+                  itemId: historicalItemId,
+                  providerThreadId,
+                  status: "completed",
+                  kind: "user_message",
+                },
+              ],
+              coverage: "Earlier request",
+              omittedItems: 0,
+            },
+            createdByProviderInstanceId: null,
+            createdAt: DateTime.makeUnsafe("2026-09-04T12:00:00Z"),
+            updatedAt: DateTime.makeUnsafe("2026-09-04T12:00:00Z"),
+          },
+        ],
+      });
+
+      yield* harness.start;
+
+      const startInput = harness.startRootRun.mock.calls[0]?.[0];
+      expect(startInput).toBeDefined();
+      const delivery = yield* Effect.result(
+        startInput!.session.startTurn({
+          appThread: startInput!.appThread,
+          threadId: startInput!.run.threadId,
+          runId: startInput!.run.id,
+          runOrdinal: startInput!.run.ordinal,
+          providerTurnOrdinal: startInput!.providerTurnOrdinal,
+          nativeThreadHasTurns: startInput!.nativeThreadHasTurns ?? false,
+          attemptId: startInput!.attemptId,
+          rootNodeId: startInput!.rootNode.id,
+          providerThread: startInput!.providerThread,
+          message: startInput!.message,
+          modelSelection: startInput!.modelSelection,
+          runtimePolicy: startInput!.runtimePolicy,
+        }),
+      );
+
+      expect(delivery._tag).toBe("Success");
+      expect(harness.startTurn).toHaveBeenCalledOnce();
+      expect(harness.startedTurns[0]?.message.text).toContain("Earlier request");
+    }),
+);
+
 effectIt.effect.each([
   { runStatus: "failed", turnStatus: "completed" },
   { runStatus: "completed", turnStatus: "completed" },