From 0f4163b9c98b5597c03038f56e72a43f4e6773b5 Mon Sep 17 00:00:00 2001 From: TinBane <363069+TinBane@users.noreply.github.com> Date: Fri, 30 Oct 2026 17:05:24 +1100 Subject: [PATCH 1/4] fix(server): native session replacement fails honestly when the provider reuses its session When a resumed native session cannot take an incoming handoff (resume failure, uncertain history delivery, or an exhausted context window), turn start forces a fresh native session and records a replacement handoff. Two safeguards were missing from that path: - A provider can answer forceNewNativeThread with its existing session (Muse does). The replacement was recorded anyway, and the turn then failed with a misleading "shorten the request" error. The replacement now fails before any bookkeeping with a protocol error saying the provider returned its existing session. - The replacement events were written unconditionally, so a run that stopped while the replacement session was being created still got a recovery handoff and transfer. They are now written with writeIfRunCurrent against the starting attempt, and start stops if the write does not commit. Co-Authored-By: Claude Opus 5.5 --- .../ProviderTurnStartService.ts | 46 +++- .../ProviderSwitch.integration.test.ts | 245 ++++++++++++++++++ 2 files changed, 278 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 2de9fcbc70d9..2b59da10cde8 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1444,21 +1444,34 @@ export const layer: Layer.Layer< errorTag: resumed._tag === "Failure" ? resumed.failure._tag : "ContextHandoffBudgetError", }); const replacement = yield* loadFromProvider( - session.ensureThread({ - threadId: projection.thread.id, - title: projection.thread.title, - modelSelection: run.modelSelection, - runtimePolicy: resolvedRuntimePolicy, - providerSessionId, - // Clear the native ref to force a fresh session while preserving row identity. - existingProviderThread: { ...providerThread, nativeThreadRef: null }, - forceNewNativeThread: true, - }), + session + .ensureThread({ + threadId: projection.thread.id, + title: projection.thread.title, + modelSelection: run.modelSelection, + runtimePolicy: resolvedRuntimePolicy, + providerSessionId, + // Clear the native ref to force a fresh session while preserving row identity. + existingProviderThread: { ...providerThread, nativeThreadRef: null }, + forceNewNativeThread: true, + }) + .pipe( + // A provider that hands back its existing session cannot take the recovery handoff. + Effect.filterOrFail( + (replacement) => + replacement.nativeThreadRef?.nativeId == null || + replacement.nativeThreadRef.nativeId !== providerThread.nativeThreadRef?.nativeId, + () => + new ProviderAdapter.ProviderAdapterProtocolError({ + driver: session.driver, + detail: + "the provider returned its existing session when asked for a fresh one, so the conversation context could not be moved to a new session", + }), + ), + ), ); if (replacement === undefined) return undefined; const { handoff, transferId, createdAt } = yield* recoveryHandoff; - effectiveHandoffs = withRecoveryHandoff(handoff); - resumeFellBack = true; const previousRun = projection.runs.findLast( (candidate) => candidate.providerThreadId === providerThread.id && @@ -1503,7 +1516,11 @@ export const layer: Layer.Layer< strategy: handoff.strategy, summary: handoff.summaryText, }; - 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 }), @@ -1542,6 +1559,9 @@ export const layer: Layer.Layer< }, ], }); + if (!replacementWrite.committed) return undefined; + effectiveHandoffs = withRecoveryHandoff(handoff); + resumeFellBack = true; recoveryTurnItem = handoffTurnItem; return replacement; }); 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 4ca516284a3b..79959992a006 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -123,6 +123,7 @@ function makeTestAdapter(input: { readonly releaseFirstTurn?: Deferred.Deferred; readonly backgroundWork?: Ref.Ref; readonly closedSessions?: Ref.Ref>; + readonly beforeForcedNativeThread?: Effect.Effect; }): ProviderAdapter.ProviderAdapterV2["Service"] { return { instanceId: input.instanceId, @@ -182,6 +183,8 @@ function makeTestAdapter(input: { }), ensureThread: (threadInput) => Effect.gen(function* () { + if (threadInput.forceNewNativeThread && input.beforeForcedNativeThread !== undefined) + yield* input.beforeForcedNativeThread; const createdAt = yield* DateTime.now; const generation = input.nativeThreadGeneration === undefined @@ -1469,6 +1472,248 @@ describe("orchestration v2 provider switching", () => { ), ); + const unverifiedReplacementHarness = ( + name: string, + adapters: ReadonlyArray, + body: (helpers: { + readonly orchestrator: Orchestrator.OrchestratorV2["Service"]; + readonly worker: EffectWorker.OrchestrationEffectWorkerV2["Service"]; + readonly dispatchRequest: ( + ordinal: number, + selection: ModelSelection, + ) => ReturnType; + readonly awaitSettled: ( + ordinal: number, + ) => Effect.Effect< + number, + Orchestrator.OrchestratorV2Error | EffectWorker.OrchestrationEffectWorkerError + >; + }) => Effect.Effect, + ) => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace(name); + return yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("unverified-replacement:create"), + threadId, + projectId, + createdBy: "user", + creationSource: "web", + title: "Unverified replacement", + modelSelection: CLAUDE_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + return yield* body({ + orchestrator, + worker, + dispatchRequest: (ordinal, selection) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`unverified-replacement:${ordinal}`), + threadId, + messageId: MessageId.make(`unverified-replacement:${ordinal}`), + createdBy: "user", + creationSource: "web", + text: `Request ${ordinal}`, + attachments: [], + modelSelection: selection, + dispatchMode: { type: "start_immediately" }, + }), + awaitSettled: (ordinal) => + orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.ordinal === ordinal && + (event.payload.status === "completed" || + event.payload.status === "failed" || + event.payload.status === "interrupted"), + ), + Stream.runHead, + Effect.andThen(worker.drain()), + ), + }); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { + name, + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { type: "readOnly" }, + }, + }, + ProviderAdapterRegistry.layerFromAdapters(adapters), + ), + ), + ); + }), + ); + // Replacement bookkeeping is a full-thread summary handoff plus a resume-fallback transfer. + const replacementRecordsFor = (projection: OrchestrationV2ThreadProjection, runId: string) => [ + ...projection.contextHandoffs.filter( + (handoff) => handoff.targetRunId === runId && handoff.strategy === "full_thread_summary", + ), + ...projection.contextTransfers.filter( + (transfer) => + transfer.targetRunId === runId && transfer.id.includes("provider_resume_fallback"), + ), + ]; + + it.live("fails honestly when the provider reuses its session for a context handoff", () => + Effect.gen(function* () { + const capturedTurns = yield* Ref.make>([]); + yield* unverifiedReplacementHarness( + "unverified-replacement-reused", + [ + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: { 2: `Source work ${"s".repeat(12_000)}` }, + capturedTurns, + }), + // Without a generation every ensureThread returns the same native id, like Muse. + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + getModelContextWindow: () => 32_000, + }), + ], + ({ orchestrator, dispatchRequest, awaitSettled }) => + Effect.gen(function* () { + yield* dispatchRequest(1, CLAUDE_MODEL_SELECTION); + yield* awaitSettled(1); + const target = (yield* orchestrator.getThreadProjection(threadId)).providerThreads[0]!; + // The native session has no room left for the returning handoff. + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("unverified-replacement-usage"), + type: "provider-thread.updated", + threadId, + occurredAt: yield* DateTime.now, + payload: { ...target, contextUsage: { usedTokens: 31_000, maxTokens: 32_000 } }, + }, + ], + }); + yield* dispatchRequest(2, CODEX_MODEL_SELECTION); + yield* awaitSettled(2); + yield* dispatchRequest(3, CLAUDE_MODEL_SELECTION); + yield* awaitSettled(3); + + const projection = yield* orchestrator.getThreadProjection(threadId); + const run3 = projection.runs.at(-1)!; + assert.equal(run3.status, "failed"); + assert.deepEqual(replacementRecordsFor(projection, run3.id), []); + const failure = projection.turnItems.findLast((item) => item.type === "error"); + assert.equal( + failure?.type === "error" ? failure.failure.message : null, + "claudeAgent provider protocol error: the provider returned its existing session when asked for a fresh one, so the conversation context could not be moved to a new session.", + ); + assert.deepEqual( + (yield* Ref.get(capturedTurns)).map((turn) => turn.driver), + [CLAUDE_DRIVER, CODEX_DRIVER], + ); + }), + ); + }), + ); + + it.live("drops replacement bookkeeping when the run stops during replacement", () => + Effect.gen(function* () { + const capturedTurns = yield* Ref.make>([]); + const failResumeOnce = yield* Ref.make(false); + const replacementRequested = yield* Deferred.make(); + const releaseReplacement = yield* Deferred.make(); + yield* unverifiedReplacementHarness( + "unverified-replacement-interrupted", + [ + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + failResumeOnce, + nativeThreadGeneration: yield* Ref.make(0), + beforeForcedNativeThread: Deferred.succeed(replacementRequested, undefined).pipe( + Effect.andThen(Deferred.await(releaseReplacement)), + ), + }), + ], + ({ orchestrator, worker, dispatchRequest, awaitSettled }) => + Effect.gen(function* () { + yield* dispatchRequest(1, CLAUDE_MODEL_SELECTION); + yield* awaitSettled(1); + const original = (yield* orchestrator.getThreadProjection(threadId)) + .providerThreads[0]!; + yield* orchestrator.dispatch({ + type: "provider-session.detach", + commandId: CommandId.make("unverified-replacement:detach"), + threadId, + providerSessionId: original.providerSessionId!, + }); + yield* worker.drain(); + yield* Ref.set(failResumeOnce, true); + yield* dispatchRequest(2, CLAUDE_MODEL_SELECTION); + yield* Deferred.await(replacementRequested); + const run2 = (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)!; + assert.equal(run2.status, "starting"); + // run.interrupt would also cancel the start effect, so move the run on directly to + // reach the window where the replacement session exists but is not yet recorded. + const settledAt = yield* DateTime.now; + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("unverified-replacement-interrupted"), + type: "run.updated", + threadId, + runId: run2.id, + occurredAt: settledAt, + payload: { ...run2, status: "interrupted", completedAt: settledAt }, + }, + ], + }); + yield* Deferred.succeed(releaseReplacement, undefined); + + // Effects run one at a time per thread, so run 3 settles after run 2's start finishes. + // The next turn resumes the original native session with no abandoned replacement. + yield* dispatchRequest(3, CLAUDE_MODEL_SELECTION); + yield* awaitSettled(3); + const final = yield* orchestrator.getThreadProjection(threadId); + const run3 = final.runs.at(-1)!; + assert.equal(final.runs[1]?.status, "interrupted"); + assert.deepEqual(replacementRecordsFor(final, run2.id), []); + assert.equal(run3.status, "completed"); + assert.deepEqual(replacementRecordsFor(final, run3.id), []); + assert.equal( + final.providerThreads[0]?.nativeThreadRef?.nativeId, + original.nativeThreadRef?.nativeId, + ); + const turns = yield* Ref.get(capturedTurns); + assert.equal(turns.length, 2); + assert.equal(turns[1]?.nativeThreadId, original.nativeThreadRef?.nativeId); + assert.include(turns[1]!.text, "Request 3"); + }), + ); + }), + ); + it.live.each(["an older handoff", "a missed request"] as const)( "keeps the latest turn when a resume fallback also delivers %s", (older) => From 855ba86384c0b39ceba4cee9f456b067aa6e91d4 Mon Sep 17 00:00:00 2001 From: TinBane <363069+TinBane@users.noreply.github.com> Date: Fri, 30 Oct 2026 21:21:37 +1100 Subject: [PATCH 2/4] fix(server): reused native session after a failed resume fails without retry When a resume fails and the forced replacement comes back with the original native id, the protocol error was propagated for an automatic retry. The session manager has already marked that session loaded, so the retry skipped the resume, found the budget fitting, and started the turn with neither the error nor the recovery summary. The identity check now runs after the thread load and settles the run as failed straight away, whatever the retry budget, so the user sees the honest error and no provider turn starts. Co-Authored-By: Claude Opus 5.5 --- .../ProviderTurnStartService.ts | 52 ++++++++++--------- .../ProviderSwitch.integration.test.ts | 51 ++++++++++++++++++ 2 files changed, 78 insertions(+), 25 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 2b59da10cde8..4a3df85da0ec 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1444,33 +1444,35 @@ export const layer: Layer.Layer< errorTag: resumed._tag === "Failure" ? resumed.failure._tag : "ContextHandoffBudgetError", }); const replacement = yield* loadFromProvider( - session - .ensureThread({ - threadId: projection.thread.id, - title: projection.thread.title, - modelSelection: run.modelSelection, - runtimePolicy: resolvedRuntimePolicy, - providerSessionId, - // Clear the native ref to force a fresh session while preserving row identity. - existingProviderThread: { ...providerThread, nativeThreadRef: null }, - forceNewNativeThread: true, - }) - .pipe( - // A provider that hands back its existing session cannot take the recovery handoff. - Effect.filterOrFail( - (replacement) => - replacement.nativeThreadRef?.nativeId == null || - replacement.nativeThreadRef.nativeId !== providerThread.nativeThreadRef?.nativeId, - () => - new ProviderAdapter.ProviderAdapterProtocolError({ - driver: session.driver, - detail: - "the provider returned its existing session when asked for a fresh one, so the conversation context could not be moved to a new session", - }), - ), - ), + session.ensureThread({ + threadId: projection.thread.id, + title: projection.thread.title, + modelSelection: run.modelSelection, + runtimePolicy: resolvedRuntimePolicy, + providerSessionId, + // Clear the native ref to force a fresh session while preserving row identity. + existingProviderThread: { ...providerThread, nativeThreadRef: null }, + forceNewNativeThread: true, + }), ); if (replacement === undefined) return undefined; + if ( + replacement.nativeThreadRef?.nativeId != null && + replacement.nativeThreadRef.nativeId === providerThread.nativeThreadRef?.nativeId + ) { + // Never retried: the reused session is now marked loaded, so a retry would skip resume + // and start the turn without this error or the recovery handoff. + yield* settleStartFailure({ + signal: "provider-thread-load-failure", + title: "Provider turn failed to start", + error: new ProviderAdapter.ProviderAdapterProtocolError({ + driver: session.driver, + detail: + "the provider returned its existing session when asked for a fresh one, so the conversation context could not be moved to a new session", + }), + }); + return undefined; + } const { handoff, transferId, createdAt } = yield* recoveryHandoff; const previousRun = projection.runs.findLast( (candidate) => 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 79959992a006..2fdc62b11632 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1633,6 +1633,57 @@ describe("orchestration v2 provider switching", () => { }), ); + it.live("fails without retrying when a failed resume's replacement reuses the session", () => + Effect.gen(function* () { + const capturedTurns = yield* Ref.make>([]); + const failResumeOnce = yield* Ref.make(false); + yield* unverifiedReplacementHarness( + "unverified-replacement-reused-resume", + [ + // Without a generation every ensureThread returns the same native id, like Muse. + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + failResumeOnce, + }), + ], + ({ orchestrator, worker, dispatchRequest, awaitSettled }) => + Effect.gen(function* () { + yield* dispatchRequest(1, CLAUDE_MODEL_SELECTION); + yield* awaitSettled(1); + const original = (yield* orchestrator.getThreadProjection(threadId)) + .providerThreads[0]!; + yield* orchestrator.dispatch({ + type: "provider-session.detach", + commandId: CommandId.make("unverified-replacement:detach"), + threadId, + providerSessionId: original.providerSessionId!, + }); + yield* worker.drain(); + // Only the first attempt fails to resume; a retry would find the session loaded. + yield* Ref.set(failResumeOnce, true); + yield* dispatchRequest(2, CLAUDE_MODEL_SELECTION); + yield* awaitSettled(2); + + const projection = yield* orchestrator.getThreadProjection(threadId); + const run2 = projection.runs.at(-1)!; + assert.equal(run2.status, "failed"); + assert.deepEqual(replacementRecordsFor(projection, run2.id), []); + const failure = projection.turnItems.findLast((item) => item.type === "error"); + assert.equal( + failure?.type === "error" ? failure.failure.message : null, + "claudeAgent provider protocol error: the provider returned its existing session when asked for a fresh one, so the conversation context could not be moved to a new session.", + ); + assert.equal((yield* Ref.get(capturedTurns)).length, 1); + }), + ); + }), + ); + it.live("drops replacement bookkeeping when the run stops during replacement", () => Effect.gen(function* () { const capturedTurns = yield* Ref.make>([]); From f0b5f6453c408de6abbf9db0e082262218d54016 Mon Sep 17 00:00:00 2001 From: TinBane <363069+TinBane@users.noreply.github.com> Date: Tue, 3 Nov 2026 20:16:53 +1100 Subject: [PATCH 3/4] fix(server): reused-session check compares the native driver too Native thread identity is driver plus native id, so a replacement on a different driver whose id string happens to match is a fresh session, not a reused one. Compare both before failing the start. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/ProviderTurnStartService.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 4a3df85da0ec..3545e1f163ae 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1458,7 +1458,8 @@ export const layer: Layer.Layer< if (replacement === undefined) return undefined; if ( replacement.nativeThreadRef?.nativeId != null && - replacement.nativeThreadRef.nativeId === providerThread.nativeThreadRef?.nativeId + replacement.nativeThreadRef.driver === providerThread.nativeThreadRef?.driver && + replacement.nativeThreadRef.nativeId === providerThread.nativeThreadRef.nativeId ) { // Never retried: the reused session is now marked loaded, so a retry would skip resume // and start the turn without this error or the recovery handoff. From 2feee4468f1465c50cd4f8e997be5c7fb187cd93 Mon Sep 17 00:00:00 2001 From: TinBane <363069+TinBane@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:21:55 +1100 Subject: [PATCH 4/4] fix(server): a rejected reused session no longer counts as loaded Rejecting a replacement that reused the native session left the thread marked loaded, so the next turn skipped resume and ran on the reused session without the recovery history or an error. Forget that load, and only that load, before settling the failure. Detach would also drop the attachment, idle unload and MCP credential. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/EffectWorker.test.ts | 1 + .../ProviderSessionManager.ts | 27 ++++++++++ .../ProviderTurnControlService.test.ts | 1 + .../ProviderTurnStartService.ts | 11 ++++- .../ProviderSwitch.integration.test.ts | 49 +++++++++++++++++++ 5 files changed, 87 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index cc43601543ee..77f88493d9a5 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -115,6 +115,7 @@ function layerExecutorFor(input: { detach: () => record("detach"), otherWorkingThreads: () => Effect.succeed([]), releaseUnlessBusy: () => Effect.succeed(true), + forgetLoadedProviderThread: () => Effect.void, subscribePromptSuggestion: () => Stream.empty, }), ), diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index eccbb8157ac4..0881dd8bd019 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -250,6 +250,17 @@ export interface ProviderSessionManagerV2Shape { readonly providerSessionId: ProviderSessionId; readonly threadId: ThreadId; }) => Effect.Effect>; + /** + * Forgets that the thread has this provider thread loaded, so its next resume reaches the + * adapter. Leaves the attachment, idle unload and MCP credential alone, unlike detach. + */ + readonly forgetLoadedProviderThread: (input: { + readonly providerSessionId: ProviderSessionId; + readonly threadId: ThreadId; + readonly providerThread: OrchestrationV2ProviderThread; + readonly modelSelection: ModelSelection; + readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy; + }) => Effect.Effect; /** The thread's latest predicted next prompt, then each replacement. */ readonly subscribePromptSuggestion: ( threadId: ThreadId, @@ -3166,6 +3177,22 @@ export const layerWithOptions = ( ), ), ), + forgetLoadedProviderThread: (input) => + Ref.update(sessions, (current) => { + const key = sessionKey(input.providerSessionId); + const entry = current.get(key); + // Only the matching load: a different key belongs to a later load of this thread. + if ( + entry === undefined || + entry.loadedProviderThreadKeyByThread.get(input.threadId) !== + providerThreadLoadKey({ ...input, runtime: entry.runtime }) + ) { + return current; + } + const loadedProviderThreadKeyByThread = new Map(entry.loadedProviderThreadKeyByThread); + loadedProviderThreadKeyByThread.delete(input.threadId); + return new Map(current).set(key, { ...entry, loadedProviderThreadKeyByThread }); + }), // Stored suggestions can outlive a deleted thread, so check the thread before serving one. subscribePromptSuggestion: (threadId) => SubscriptionRef.changes(promptSuggestionStore).pipe( diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index a8bc76977a8a..ca6261bbf876 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -296,6 +296,7 @@ it.effect( detach: () => Effect.void, otherWorkingThreads: () => Effect.succeed([]), releaseUnlessBusy: () => Effect.succeed(true), + forgetLoadedProviderThread: () => Effect.void, subscribePromptSuggestion: () => Stream.empty, }), ); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 3545e1f163ae..7b162706b5e9 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1461,8 +1461,15 @@ export const layer: Layer.Layer< replacement.nativeThreadRef.driver === providerThread.nativeThreadRef?.driver && replacement.nativeThreadRef.nativeId === providerThread.nativeThreadRef.nativeId ) { - // Never retried: the reused session is now marked loaded, so a retry would skip resume - // and start the turn without this error or the recovery handoff. + // Never retried. Forget the load first, or the next turn would skip resume and start on + // the reused session without this error or the recovery handoff. + yield* providerSessions.forgetLoadedProviderThread({ + providerSessionId: session.providerSessionId, + threadId: projection.thread.id, + providerThread: replacement, + modelSelection: run.modelSelection, + runtimePolicy: resolvedRuntimePolicy, + }); yield* settleStartFailure({ signal: "provider-thread-load-failure", title: "Provider turn failed to start", 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 2fdc62b11632..d2395ca018ef 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1684,6 +1684,55 @@ describe("orchestration v2 provider switching", () => { }), ); + it.live("keeps failing honestly on the next turn after rejecting a reused session", () => + Effect.gen(function* () { + const capturedTurns = yield* Ref.make>([]); + yield* unverifiedReplacementHarness( + "unverified-replacement-reused-next-turn", + [ + // Resume always fails and every ensureThread returns the same native id. + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + failResume: true, + }), + ], + ({ orchestrator, worker, dispatchRequest, awaitSettled }) => + Effect.gen(function* () { + yield* dispatchRequest(1, CLAUDE_MODEL_SELECTION); + yield* awaitSettled(1); + const original = (yield* orchestrator.getThreadProjection(threadId)) + .providerThreads[0]!; + yield* orchestrator.dispatch({ + type: "provider-session.detach", + commandId: CommandId.make("unverified-replacement:detach"), + threadId, + providerSessionId: original.providerSessionId!, + }); + yield* worker.drain(); + yield* dispatchRequest(2, CLAUDE_MODEL_SELECTION); + yield* awaitSettled(2); + // The rejected session must not count as loaded, or this turn would skip resume and + // run on it without the recovery history. + yield* dispatchRequest(3, CLAUDE_MODEL_SELECTION); + yield* awaitSettled(3); + + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + projection.runs.map((run) => run.status), + ["completed", "failed", "failed"], + ); + assert.deepEqual(replacementRecordsFor(projection, projection.runs[2]!.id), []); + assert.equal((yield* Ref.get(capturedTurns)).length, 1); + }), + ); + }), + ); + it.live("drops replacement bookkeeping when the run stops during replacement", () => Effect.gen(function* () { const capturedTurns = yield* Ref.make>([]);