diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 035bd9c73db66..3dc0a65ae03a3 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6241,24 +6241,69 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio activeProviderThread.providerInstanceId !== modelSelection.instanceId; // Account overlays share native history. Selection commands may already // have updated the app thread, so classify against the native thread's owner. + const providerSwitchPlan = isProviderSwitch + ? yield* providerSwitchService + .plan({ + projection: { + ...projection, + thread: { + ...projection.thread, + modelSelection: { + ...projection.thread.modelSelection, + instanceId: activeProviderThread.providerInstanceId, + }, + }, + }, + targetModelSelection: modelSelection, + }) + .pipe(mapDispatchError(command)) + : null; const canResumeAcrossInstances = isProviderSwitch && activeProviderThread.nativeThreadRef !== null && - (yield* providerSwitchService - .plan({ - projection: { - ...projection, - thread: { - ...projection.thread, - modelSelection: { - ...projection.thread.modelSelection, - instanceId: activeProviderThread.providerInstanceId, - }, - }, - }, - targetModelSelection: modelSelection, - }) - .pipe(mapDispatchError(command))).transition.type === "restart_and_resume"; + providerSwitchPlan?.transition.type === "restart_and_resume"; + const sessionsToDetach = projection.providerSessions.filter( + (session) => + providerSwitchPlan?.releaseProviderSessionIds.includes(session.id) && + session.status !== "stopped" && + session.status !== "error", + ); + // Detaching kills background work running inside the provider process. + // Probe even with an empty roster: Claude's native subagents never reach + // pendingBackgroundTasks. Sessions shared across threads probe per thread. + for (const detaching of sessionsToDetach) { + const session = yield* providerSessions + .get(detaching.id) + .pipe(Effect.orElseSucceed(() => Option.none())); + if (Option.isNone(session)) { + continue; + } + const runtime = session.value; + const probes = + !detaching.capabilities.sessions.supportsMultipleProviderThreadsPerSession && + runtime.hasPendingBackgroundWork !== undefined + ? [runtime.hasPendingBackgroundWork] + : projection.providerThreads.flatMap((providerThread) => + providerThread.providerSessionId === detaching.id && + runtime.hasPendingBackgroundWorkForThread !== undefined + ? [runtime.hasPendingBackgroundWorkForThread(providerThread)] + : [], + ); + for (const probe of probes) { + const stillRunning = yield* probe.pipe( + // A failed probe cannot confirm that the work ended. + Effect.catchCause(() => Effect.succeed(true)), + ); + if (stillRunning) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: + "The previous provider is still running background agents or commands, and switching would end them. Wait for them to finish, or press Stop, then send the message again.", + }); + } + } + } if ( pendingForkTransfer === undefined && @@ -7565,30 +7610,32 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } - if (canResumeAcrossInstances && activeProviderThread.providerSessionId !== null) { - const previousProviderSessionId = activeProviderThread.providerSessionId; + const detachDetail = canResumeAcrossInstances + ? "Provider account changed; continuing the native thread." + : "Provider or model selection changed."; + for (const session of sessionsToDetach) { yield* emitEvent({ type: "provider-session.detached", threadId: command.threadId, - driver: activeProviderThread.driver, - providerInstanceId: activeProviderThread.providerInstanceId, + driver: session.driver, + providerInstanceId: session.providerInstanceId, occurredAt: now, payload: { - providerSessionId: previousProviderSessionId, + providerSessionId: session.id, detachedAt: now, - reason: "Provider account changed; continuing the native thread.", + reason: detachDetail, }, }); yield* Ref.update(effects, (existing) => [ ...existing, { - id: `effect:${command.commandId}:provider-session.detach:${previousProviderSessionId}`, + id: `effect:${command.commandId}:provider-session.detach:${session.id}`, commandId: command.commandId, threadId: command.threadId, request: { type: "provider-session.detach", - providerSessionId: previousProviderSessionId, - detail: "Provider account changed; continuing the native thread.", + providerSessionId: session.id, + detail: detachDetail, }, } satisfies PendingOrchestrationEffectV2, ]); 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..f36e463079f99 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -119,6 +119,8 @@ function makeTestAdapter(input: { readonly subagentOnlyBackgroundWork?: boolean; readonly holdFirstTurn?: Deferred.Deferred; readonly releaseFirstTurn?: Deferred.Deferred; + readonly backgroundWork?: Ref.Ref; + readonly closedSessions?: Ref.Ref>; }): ProviderAdapter.ProviderAdapterV2["Service"] { return { instanceId: input.instanceId, @@ -128,6 +130,12 @@ function makeTestAdapter(input: { openSession: (sessionInput) => Effect.gen(function* () { const events = yield* PubSub.unbounded(); + if (input.closedSessions !== undefined) { + const closedSessions = input.closedSessions; + yield* Effect.addFinalizer(() => + Ref.update(closedSessions, (ids) => [...ids, sessionInput.providerSessionId]), + ); + } const now = yield* DateTime.now; const providerSession: OrchestrationV2ProviderSession = { id: sessionInput.providerSessionId, @@ -162,6 +170,14 @@ function makeTestAdapter(input: { ...(input.canReuseContextUsage === undefined ? {} : { canReuseContextUsage: input.canReuseContextUsage }), + ...(input.backgroundWork === undefined + ? {} + : { + // Like Claude: native subagents are visible only to the session-wide + // probe, never to the per-thread roster probe. + hasPendingBackgroundWork: Ref.get(input.backgroundWork), + hasPendingBackgroundWorkForThread: () => Effect.succeed(false), + }), ensureThread: (threadInput) => Effect.gen(function* () { const createdAt = yield* DateTime.now; @@ -748,16 +764,7 @@ describe("orchestration v2 provider switching", () => { yield* wait(3); } if (replaceNative) { - const target = (yield* orchestrator.getThreadProjection(threadId)).providerThreads.find( - (thread) => thread.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, - )!; - yield* orchestrator.dispatch({ - type: "provider-session.detach", - commandId: CommandId.make("detach-for-native-replacement"), - threadId, - providerSessionId: target.providerSessionId!, - }); - yield* worker.drain(); + // Switching back to Codex already released Claude's session. yield* Ref.set(failResumeOnce, true); } if (scenario.includes("retry")) yield* Ref.set(failStartOnce, true); @@ -2520,6 +2527,141 @@ describe("orchestration v2 provider switching", () => { ), ); + // regression: #20343 sending to an instance with separate history must release the old session + it.live.each([ + ["detaches the previous instance's session when a send hands off to another instance", false], + ["refuses the hand-off while the previous instance still runs background work", true], + ] as const)("%s", ([, backgroundRunning]) => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("handoff-releases-session"); + const capturedTurns = yield* Ref.make>([]); + const backgroundWork = yield* Ref.make(backgroundRunning); + const closedSessions = yield* Ref.make>([]); + const secondSelection: ModelSelection = { + ...CLAUDE_MODEL_SELECTION, + instanceId: ProviderInstanceId.make("claude-second-account"), + }; + // No shared continuation key: each Claude instance owns its own native history. + const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: { 1: "First account complete" }, + capturedTurns, + backgroundWork, + closedSessions, + }), + makeTestAdapter({ + instanceId: secondSelection.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: secondSelection, + responseByRunOrdinal: { 2: "Second account complete" }, + capturedTurns, + }), + ]); + const handoffThreadId = ThreadId.make("thread:handoff-releases-session"); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const send = (ordinal: number, modelSelection: ModelSelection, attempt = "") => + orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:handoff-releases-session:${ordinal}${attempt}`), + threadId: handoffThreadId, + messageId: MessageId.make(`message:handoff-releases-session:${ordinal}${attempt}`), + text: `Prompt ${ordinal}`, + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:handoff-releases-session:create"), + threadId: handoffThreadId, + projectId: ProjectId.make("project:handoff-releases-session"), + title: "Handoff releases session", + modelSelection: CLAUDE_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* send(1, CLAUDE_MODEL_SELECTION); + yield* worker.drain(); + const afterFirst = yield* waitForIdle(handoffThreadId); + const firstSession = afterFirst.providerSessions.find( + (session) => session.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, + ); + assert.equal(firstSession?.status, "ready"); + + if (backgroundRunning) { + // Native subagents live only in the adapter's roster, so the projection's + // pendingBackgroundTasks stays empty while the probe still reports work. + assert.deepEqual(afterFirst.providerThreads[0]?.pendingBackgroundTasks ?? [], []); + const refused = yield* send(2, secondSelection, ":refused").pipe(Effect.flip); + assert.include(String(refused.cause), "still running background agents or commands"); + const unchanged = yield* orchestrator.getThreadProjection(handoffThreadId); + assert.lengthOf(unchanged.runs, 1); + assert.deepEqual(yield* Ref.get(closedSessions), []); + assert.equal(unchanged.thread.providerInstanceId, CLAUDE_MODEL_SELECTION.instanceId); + assert.deepEqual( + unchanged.providerSessions.map((session) => [session.id, session.status]), + [[firstSession!.id, "ready"]], + ); + yield* Ref.set(backgroundWork, false); + } + + yield* send(2, secondSelection); + yield* worker.drain(); + const afterSecond = yield* waitForIdle(handoffThreadId); + yield* worker.drain(); + const projection = yield* orchestrator.getThreadProjection(handoffThreadId); + + assert.deepEqual( + projection.runs.map((run) => [run.providerInstanceId, run.status]), + [ + [CLAUDE_MODEL_SELECTION.instanceId, "completed"], + [secondSelection.instanceId, "completed"], + ], + ); + assert.lengthOf(afterSecond.contextHandoffs, 1); + assert.deepEqual( + projection.providerSessions.map((session) => session.providerInstanceId), + [secondSelection.instanceId], + ); + // The detached event alone hides the session; the runtime must also close. + assert.deepEqual(yield* Ref.get(closedSessions), [firstSession!.id]); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { + name: "handoff-releases-session", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + layerRegistry, + ), + ), + ); + }), + ), + ); + // regression: #21314 a queued PR update must follow the provider that took over the thread it.live.each([ ["automatic", "Claude"],