diff --git a/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts b/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts index c4068702e3a57..de5019657d6f0 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts +++ b/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" }, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index beaf7961e3782..4f620a2a19168 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -1169,6 +1169,7 @@ 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(); @@ -1176,9 +1177,38 @@ effectIt.effect.each([ 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(); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index fad06732fb48e..ec09c6b2255a4 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/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; diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index bb21d8ca9929a..1fb8078320152 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/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");