fix(server): restart oversized provider handoffs safely

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

@@ -399,6 +399,15 @@ describe("handoff budget", () => {
 });
 
 describe("handoff delivery", () => {
+  it.effect("does not evaluate the budget when no handoff needs delivery", () =>
+    deliverContextHandoffs({
+      handoffs: [],
+      providerThread,
+      budget: Effect.die("settled handoffs need no budget"),
+      alreadyDeliveredItemIds: new Set(),
+      persist: () => Effect.void,
+    }).pipe(Effect.asVoid),
+  );
   it.effect.each([
     { native: true, label: "injected" },
     { native: false, label: "inline" },

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

@@ -1169,16 +1169,46 @@ effectIt.effect("reads current startup state after waiting for the workspace lea
 effectIt.effect.each([
   { boundary: "policy resolution", loss: "replaced" },
   { boundary: "session opening", loss: "interrupted" },
+  { boundary: "native replacement", loss: "interrupted" },
 ])("does not continue startup when $loss during $boundary", ({ boundary, loss }) =>
   Effect.gen(function* () {
     const entered = yield* Deferred.make<void>();
     const release = yield* Deferred.make<void>();
     const gate = Deferred.succeed(entered, undefined).pipe(Effect.andThen(Deferred.await(release)));
     const harness = makeLocalCommandHarness({ text: "Continue" });
     const projection = harness.projection();
-    const providerThread = projection.providerThreads[0]!;
-    const ensureThread = vi.fn(() => Effect.succeed(providerThread));
-    const resumeThread = vi.fn(() => Effect.succeed(providerThread));
+    const replacing = boundary === "native replacement";
+    const providerThread = replacing
+      ? {
+          ...projection.providerThreads[0]!,
+          nativeThreadRef: {
+            driver: projection.providerThreads[0]!.driver,
+            nativeId: "old",
+            strength: "strong" as const,
+          },
+        }
+      : projection.providerThreads[0]!;
+    if (replacing) harness.setProjection({ ...projection, providerThreads: [providerThread] });
+    const ensureThread = vi.fn(() =>
+      replacing
+        ? gate.pipe(
+            Effect.as({
+              ...providerThread,
+              nativeThreadRef: { ...providerThread.nativeThreadRef!, nativeId: "fresh" },
+            }),
+          )
+        : Effect.succeed(providerThread),
+    );
+    const resumeThread = vi.fn(() =>
+      replacing
+        ? Effect.fail(
+            new ProviderAdapter.ProviderAdapterProtocolError({
+              driver: providerThread.driver,
+              detail: "resume failed",
+            }),
+          )
+        : Effect.succeed(providerThread),
+    );
     const forkThread = vi.fn(() => Effect.succeed(providerThread));
     harness.open.mockReturnValue(
       (boundary === "session opening" ? gate : Effect.void).pipe(
@@ -1216,9 +1246,9 @@ effectIt.effect.each([
     yield* Deferred.succeed(release, undefined);
     yield* Fiber.join(startup);
 
-    expect(harness.open).toHaveBeenCalledTimes(boundary === "session opening" ? 1 : 0);
-    expect(ensureThread).not.toHaveBeenCalled();
-    expect(resumeThread).not.toHaveBeenCalled();
+    expect(harness.open).toHaveBeenCalledTimes(boundary === "session opening" || replacing ? 1 : 0);
+    expect(ensureThread).toHaveBeenCalledTimes(replacing ? 1 : 0);
+    expect(resumeThread).toHaveBeenCalledTimes(replacing ? 1 : 0);
     expect(forkThread).not.toHaveBeenCalled();
     expect(harness.startRootRun).not.toHaveBeenCalled();
     expect(harness.writeIfRunCurrent).not.toHaveBeenCalled();

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

@@ -1194,10 +1194,27 @@ export const layer: Layer.Layer<
           }),
         );
         if (replacement === undefined) return undefined;
+        if (!(yield* isCurrentAttemptInStatus("starting"))) return undefined;
+        if (
+          replacement.nativeThreadRef?.nativeId == null ||
+          replacement.nativeThreadRef.nativeId === providerThread.nativeThreadRef?.nativeId
+        ) {
+          yield* settleStartFailure({
+            signal: "provider-thread-load-failure",
+            title: "Provider turn failed to start",
+            error: new ProviderAdapter.ProviderAdapterProtocolError({
+              driver: session.driver,
+              detail: "Provider did not create a fresh native thread for the context handoff",
+            }),
+          });
+          return undefined;
+        }
         const { handoff, transferId, createdAt } = yield* recoveryHandoff;
-        effectiveHandoffs = withRecoveryHandoff(handoff);
-        resumeFellBack = true;
-        yield* eventSink.write({
+        const replacementWrite = yield* eventSink.writeIfRunCurrent({
+          threadId: projection.thread.id,
+          runId: run.id,
+          activeAttemptId: attempt.id,
+          expectedStatus: "starting",
           events: [
             {
               id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }),
@@ -1236,6 +1253,9 @@ export const layer: Layer.Layer<
             },
           ],
         });
+        if (!replacementWrite.committed) return undefined;
+        effectiveHandoffs = withRecoveryHandoff(handoff);
+        resumeFellBack = true;
         return replacement;
       });
       if (loadedProviderThread === undefined) return;

apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts

@@ -430,6 +430,7 @@ describe("orchestration v2 provider switching", () => {
     "screenshot-reported-capacity-native",
     "screenshot-option-change-retry-native",
     "screenshot-exhausted-target-native",
+    "screenshot-exhausted-target-reused-native",
     "screenshot-exhausted-target-fallback",
     "screenshot-exhausted-target-estimate-native",
     "screenshot-exhausted-target-oversized-native",
@@ -441,6 +442,7 @@ describe("orchestration v2 provider switching", () => {
     "prior-images-telemetry-native",
     "prior-images-unsent-native",
     "prior-images-legacy-replacement-native",
+    "prior-images-replacement-reused-native",
     "delivery-write-failure",
   ] as const)("preserves handoffs through %s", (scenario) =>
     Effect.scoped(
@@ -459,6 +461,7 @@ describe("orchestration v2 provider switching", () => {
         const priorImages = scenario.includes("prior-images");
         const exhaustedTarget = scenario.includes("exhausted-target");
         const replaceNative = scenario.includes("replacement");
+        const reuseNative = scenario.includes("-reused-");
         const capacityScenario = modelScenario || scenario.includes("reported-capacity");
         const returning =
           scenario === "large-current-input" ||
@@ -494,8 +497,10 @@ describe("orchestration v2 provider switching", () => {
               : {},
             capturedTurns,
             ...(scenario.includes("retry") || scenario.includes("unsent") ? { failStartOnce } : {}),
-            ...(replaceNative ? { failResumeOnce, nativeThreadGeneration: generation } : {}),
-            ...(exhaustedTarget ? { nativeThreadGeneration: generation } : {}),
+            ...(replaceNative ? { failResumeOnce } : {}),
+            ...((replaceNative || exhaustedTarget) && !reuseNative
+              ? { nativeThreadGeneration: generation }
+              : {}),
             ...(scenario.includes("target-background") || scenario.includes("target-subagent")
               ? {
                   pendingBackgroundWork,
@@ -765,6 +770,27 @@ describe("orchestration v2 provider switching", () => {
             yield* Ref.set(pendingBackgroundWork, true);
           yield* dispatch(targetOrdinal, current, targetSelection);
           yield* wait(targetOrdinal);
+          if (reuseNative) {
+            const projection = yield* orchestrator.getThreadProjection(threadId);
+            assert.equal(projection.runs.at(-1)?.status, "failed");
+            const turns = (yield* Ref.get(capturedTurns)).filter(
+              (turn) => turn.driver === CLAUDE_DRIVER,
+            );
+            assert.lengthOf(turns, 1);
+            const target = projection.providerThreads.find(
+              (thread) => thread.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId,
+            )!;
+            assert.equal(target.nativeThreadRef?.nativeId, turns[0]!.nativeThreadId);
+            assert.isFalse(
+              projection.contextHandoffs.some(
+                (handoff) =>
+                  handoff.targetRunId === projection.runs.at(-1)?.id &&
+                  handoff.strategy === "full_thread_summary" &&
+                  handoff.fromProviderThreadIds.includes(target.id),
+              ),
+            );
+            return;
+          }
           if (scenario.includes("retry")) {
             const failed = yield* orchestrator.getThreadProjection(threadId);
             assert.equal(failed.runs.at(-1)?.status, "failed");