From ac8932f6e15cb676d24bf7c77ba5f3349975898c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 19:45:18 +0000 Subject: [PATCH 1/7] fix(server): a send no longer bounces while the previous session runs background work A direct send now follows the queued-start rules: a hand-off or a fresh session starts beside a busy session and leaves it running, and an account switch that resumes the busy session's native thread waits in the queue. A waiting account switch is no longer held; it starts by itself when the session reports that its work settled. The Claude probe counts a wake the live process still owes (a task whose roster emptied before task_notification, or a notification buffered ahead of its wake turn), so that retry cannot close the CLI before the reply. Co-authored-by: Lakshmi Tanmay Co-authored-by: Mika Hertler Co-Authored-By: Claude Fable 5.2 --- .../Adapters/ClaudeAdapterV2.test.ts | 25 +- .../Adapters/ClaudeAdapterV2.ts | 19 +- ...laudeAccountSwitchWake.integration.test.ts | 381 ++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 152 +++---- .../ProviderSwitch.integration.test.ts | 135 ++++--- 5 files changed, 575 insertions(+), 137 deletions(-) create mode 100644 apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 155370da1757d..c9fbcdfb99809 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11063,7 +11063,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); it.effect( - "uses task_started as an incremental roster fallback, keeps its tool call, and clears internal work on empty snapshot", + "uses task_started as an incremental roster fallback and still owes its wake after an empty snapshot", () => Effect.gen(function* () { const harness = yield* makeWakeHarness; @@ -11211,7 +11211,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ).map((task) => task.taskId), WAKE_TASK_ID, ); - assert.isFalse(yield* harness.hasPendingBackgroundWork); + assert.isTrue(yield* harness.hasPendingBackgroundWork); }).pipe( Effect.provide( Layer.mergeAll( @@ -13937,9 +13937,20 @@ describe("ClaudeAdapterV2 background wake turns", () => { assert.isTrue(yield* harness.hasPendingBackgroundWork); assert.lengthOf(harness.continuationRequests, 0); + // Claude may empty the roster before the notification; the session still owes the wake. + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "system", + subtype: "background_tasks_changed", + tasks: [], + uuid: "00000000-0000-4000-8000-000000000299", + session_id: WAKE_NATIVE_SESSION, + }), + ); + assert.isTrue(yield* harness.hasPendingBackgroundWork); yield* harness.offerAndWait(wakeNotification); assert.lengthOf(harness.continuationRequests, 0); - assert.isFalse(yield* harness.hasPendingBackgroundWork); + assert.isTrue(yield* harness.hasPendingBackgroundWork); yield* Queue.offer(harness.sdkMessages, wakeAssistant); yield* harness.awaitContinuation(1); @@ -14828,8 +14839,8 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* harness.offerAndWait(emptyRoster); yield* harness.offerAndWait(wakeNotification); if (emptyLevel === "after") yield* harness.offerAndWait(emptyRoster); - // Query replacement reads the internal roster, which the exit already cleared. - assert.isFalse(yield* harness.hasPendingBackgroundWork); + // The notification is buffered and its wake turn is still to come. + assert.isTrue(yield* harness.hasPendingBackgroundWork); yield* harness.offerAndWait(wakeTurnInit); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -19145,7 +19156,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); yield* Queue.offer(firstProcess, turnOneResult); yield* changes.until(() => events.some((event) => event.type === "turn.terminal")); - assert.isFalse(yield* hasPendingBackgroundWork); + assert.isTrue(yield* hasPendingBackgroundWork); const alternateModel = CLAUDE_TEST_RESTART_MODEL_SELECTION; const activeProviderThread = { @@ -19343,7 +19354,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* changes.until( () => events.filter((event) => event.type === "turn.terminal").length === 2, ); - assert.isFalse(yield* hasPendingBackgroundWork); + assert.isTrue(yield* hasPendingBackgroundWork); assert.lengthOf(continuationRequests, 0); yield* Queue.offer(secondProcess, wakeResult); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index b3c1fd1c1fed4..98e26849f3189 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -13161,7 +13161,24 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return true; } } - return false; + // Only a live process can still deliver a notification or the wake turn that follows it. + const live = yield* Ref.get(queryContext); + if (live === null || live.stopping) { + return false; + } + // The roster can empty before task_notification, and the notification before its wake turn. + return ( + taskIdSetForNativeThread( + yield* Ref.get(wakeEligibleBackgroundTasksByNativeThread), + live.nativeThreadId, + ).size > 0 || + (buffers + .get(live.nativeThreadId) + ?.messages.some( + (message) => message.type === "system" && message.subtype === "task_notification", + ) ?? + false) + ); }), // Untraced: the run's event stream calls this once per provider event. hasPendingBackgroundWorkForThread: Effect.fnUntraced(function* (providerThread, options) { diff --git a/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts new file mode 100644 index 0000000000000..0fee4be994428 --- /dev/null +++ b/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts @@ -0,0 +1,381 @@ +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + ClaudeSettings, + CommandId, + MessageId, + ProviderInstanceId, + ProjectId, + ThreadId, + type ModelSelection, + type OrchestrationV2DomainEvent, +} from "@t3tools/contracts"; +import * as SecureRandom from "effect/crypto/SecureRandom"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; +import type * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; +import * as ClaudeAdapterV2 from "./Adapters/ClaudeAdapterV2.ts"; +import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts"; +import * as EventSink from "./EventSink.ts"; +import * as EffectWorker from "./EffectWorker.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; + +const nativeSessionId = "account-switch-wake-session"; +const settings = Schema.decodeSync(ClaudeSettings)({}); +const firstAccount: ModelSelection = { + instanceId: ProviderInstanceId.make("claudeAgent"), + model: "claude-sonnet-4-6", +}; +const secondAccount: ModelSelection = { + instanceId: ProviderInstanceId.make("claude-second-account"), + model: "claude-sonnet-4-6", +}; + +function frame(value: Record): SDKMessage { + return { session_id: nativeSessionId, ...value } as unknown as SDKMessage; +} + +const resultFrame = (uuid: string, text: string, wake = false) => + frame({ + type: "result", + subtype: "success", + uuid, + is_error: false, + num_turns: 1, + result: text, + stop_reason: "end_turn", + terminal_reason: "completed", + permission_denials: [], + duration_ms: 1, + duration_api_ms: 1, + total_cost_usd: 0, + usage: { input_tokens: 1, output_tokens: 1 }, + modelUsage: {}, + ...(wake ? { origin: { kind: "task-notification" } } : {}), + }); + +const assistantFrame = (uuid: string, text: string) => + frame({ + type: "assistant", + uuid, + parent_tool_use_id: null, + message: { + id: `msg_${uuid}`, + role: "assistant", + type: "message", + model: firstAccount.model, + content: [{ type: "text", text }], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 1, output_tokens: 1 }, + }, + }); + +const backgroundBashStarted = frame({ + type: "system", + subtype: "task_started", + task_id: "bash-build", + tool_use_id: "toolu_bash_build", + description: "Build in the background", + is_backgrounded: true, + task_type: "local_bash", + uuid: "bash-started", +}); +// Claude may publish the emptied roster before the notification that owes a wake turn. +const emptyRoster = frame({ + type: "system", + subtype: "background_tasks_changed", + tasks: [], + uuid: "roster-empty", +}); +const bashNotification = frame({ + type: "system", + subtype: "task_notification", + task_id: "bash-build", + tool_use_id: "toolu_bash_build", + status: "completed", + output_file: "/tmp/claude-replay/tasks/bash-build.output", + summary: 'Background command "Build in the background" completed (exit code 0)', + uuid: "bash-notification", +}); +const WAKE_REPLY = "BUILD_REPORT_DELIVERED"; + +const makeAccount = Effect.fn("makeAccount")(function* (input: { + readonly instanceId: ProviderInstanceId; + readonly cwd: string; + readonly continuations: Queue.Queue; + readonly closed: Array; +}) { + const processes: Array> = []; + const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({ + instanceId: input.instanceId, + settings, + environment: {}, + attachmentsDir: input.cwd, + fileSystem: yield* FileSystem.FileSystem, + path: yield* Path.Path, + random: yield* SecureRandom.SecureRandom, + idAllocator: yield* IdAllocator.IdAllocatorV2, + continuationRequests: { + offer: (request) => Queue.offer(input.continuations, request).pipe(Effect.asVoid), + }, + queryRunner: { + allocateSessionId: Effect.succeed(nativeSessionId), + open: () => + Effect.gen(function* () { + const messages = yield* Queue.unbounded(); + processes.push(messages); + return { + messages: Stream.fromQueue(messages), + offer: () => Effect.void, + setModel: () => Effect.void, + applyFlagSettings: () => Effect.void, + setPermissionMode: () => Effect.void, + backgroundTasks: () => Effect.succeed(false), + mcpServerStatus: Effect.succeed([]), + readMcpResource: () => Effect.die("unexpected MCP resource read"), + callMcpTool: () => Effect.die("unexpected MCP tool call"), + listPermissionRules: Effect.die("unexpected permission rules read"), + stopTask: () => Effect.void, + interrupt: Effect.void, + close: Effect.sync(() => void input.closed.push(input.instanceId)).pipe( + Effect.andThen(Queue.shutdown(messages)), + ), + }; + }), + forkSession: () => Effect.die("unused"), + subagentLaunchToolUseId: () => Effect.succeed(null), + subagentModel: () => Effect.succeed(null), + assertComplete: Effect.void, + }, + }); + return { adapter, processes }; +}); + +// regression: #32296 an account switch must not close the CLI that still owes a wake reply +it.live.each(["sent after", "queued behind"] as const)( + "delivers a Claude background reply before an account switch %s the turn", + (variant) => + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("claude-account-switch-wake"); + const continuations = + yield* Queue.unbounded(); + const closed: Array = []; + const first = yield* makeAccount({ + instanceId: firstAccount.instanceId, + cwd, + continuations, + closed, + }); + const second = yield* makeAccount({ + instanceId: secondAccount.instanceId, + cwd, + continuations, + closed, + }); + const adapters = [first.adapter, second.adapter]; + // Two accounts of one Claude install resume each other's native history. + const registry = Layer.succeed( + ProviderAdapterRegistry.ProviderAdapterRegistryV2, + ProviderAdapterRegistry.ProviderAdapterRegistryV2.of({ + get: (instanceId) => { + const adapter = adapters.find((candidate) => candidate.instanceId === instanceId); + return adapter === undefined + ? Effect.fail( + new ProviderAdapterRegistry.ProviderAdapterRegistryLookupError({ instanceId }), + ) + : Effect.succeed(adapter); + }, + list: () => Effect.succeed(adapters.map((adapter) => adapter.instanceId)), + getMetadata: () => + Effect.succeed({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + continuationKey: "claude:shared-native-account-history", + enabled: true, + capabilities: ClaudeProviderCapabilitiesV2, + }), + }), + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("thread:claude-account-switch-wake"); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const runStatus = (messageId: string, status: string) => + watch( + (event) => + event.type === "run.updated" && + event.payload.userMessageId === messageId && + event.payload.status === status, + ); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make("project:claude-account-switch-wake"), + title: "Account switch during a background build", + modelSelection: firstAccount, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const queued = variant === "queued behind"; + const send = (ordinal: number, modelSelection: ModelSelection) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`send-${ordinal}`), + threadId, + messageId: MessageId.make(`send-${ordinal}`), + text: `Prompt ${ordinal}`, + attachments: [], + modelSelection, + dispatchMode: { type: queued ? "queue_after_active" : "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + + const firstRunning = yield* watch( + (event) => event.type === "provider-turn.updated" && event.payload.status === "running", + ); + yield* send(1, firstAccount); + yield* worker.drain(); + yield* Fiber.join(firstRunning); + if (queued) yield* send(2, secondAccount); + const firstSettling = yield* runStatus("send-1", "waiting"); + const firstProcess = first.processes[0]!; + yield* Queue.offer(firstProcess, backgroundBashStarted); + yield* Queue.offer(firstProcess, resultFrame("turn-1", "Started the build.")); + yield* Fiber.join(firstSettling); + yield* worker.drain(); + + if (!queued) yield* send(2, secondAccount); + yield* worker.drain(); + const waiting = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + waiting.runs.map((run) => [run.providerInstanceId, run.status]), + [ + [firstAccount.instanceId, "completed"], + [secondAccount.instanceId, "queued"], + ], + ); + + // The settled roster gives queue resumption its chance before the notification. + const rosterSettled = yield* orchestrator + .streamStoredEventsFrom({ + threadId, + afterSequence: yield* (yield* EventSink.EventSinkV2).latestSequence(), + eventType: "provider-thread.updated", + }) + .pipe(Stream.runHead, Effect.forkScoped); + yield* Queue.offer(firstProcess, emptyRoster); + yield* Fiber.join(rosterSettled); + yield* worker.drain(); + yield* orchestrator.resumeQueuedRuns; + yield* worker.drain(); + const afterRoster = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual(closed, [], "the notification is still owed by the first account"); + assert.equal(afterRoster.runs[1]?.status, "queued"); + assert.equal( + afterRoster.providerSessions.find( + (session) => session.providerInstanceId === firstAccount.instanceId, + )?.status, + "ready", + ); + + // The notification and wake reply arrive on the still-open process. + yield* Queue.offer(firstProcess, bashNotification); + yield* Queue.offer(firstProcess, assistantFrame("wake-assistant", WAKE_REPLY)); + yield* Queue.offer(firstProcess, resultFrame("wake-result", WAKE_REPLY, true)); + const request = yield* Queue.take(continuations); + const wakeSettling = yield* runStatus("wake", "waiting"); + const switched = yield* runStatus("send-2", "running"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("wake"), + threadId, + messageId: MessageId.make("wake"), + text: request.detail ?? "Background task completed.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "provider", + notification: request.notification ?? { + source: { kind: "background_task" }, + outcome: "updated", + summary: "Background activity updated", + }, + }); + yield* worker.drain(); + yield* Fiber.join(wakeSettling); + yield* worker.drain(); + yield* Fiber.join(switched); + yield* worker.drain(); + + const after = yield* orchestrator.getThreadProjection(threadId); + assert.isTrue( + after.messages.some((message) => message.text === WAKE_REPLY), + "the wake reply must reach the thread", + ); + const wakeRun = after.runs.find((run) => run.userMessageId === "wake"); + assert.equal(wakeRun?.providerInstanceId, firstAccount.instanceId); + assert.equal(wakeRun?.status, "completed"); + assert.equal( + after.runs.find((run) => run.userMessageId === "send-2")?.providerInstanceId, + secondAccount.instanceId, + ); + // Only now may the switch release the first account's process. + assert.deepEqual(closed, [firstAccount.instanceId]); + assert.lengthOf(second.processes, 1); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { + name: "claude-account-switch-wake", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registry, + { runEffectWorker: false }, + ), + ), + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), +); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 00f9dfae8d9e1..1fbfebd3dbed3 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2289,6 +2289,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return true; }); + // In memory: the sessions these queues wait on do not outlive the server either. + const queuesWaitingOnBackgroundWork = new Set(); const startNextQueuedRun = Effect.fn("Orchestrator.startNextQueuedRun")( function* ( threadId: ThreadId, @@ -2590,8 +2592,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio busySessionIds.add(session.id); } } - // Nobody waits on a queued start to refuse it. A handoff runs in its own session, so a - // busy one stays up; a restart resumes the busy session's native thread, so wait instead. + // A handoff runs in its own session, so a busy one stays up. A restart resumes the busy + // session's native thread, so it waits without a hold and starts when that work settles. // A fresh run resumes nothing, so it never waits. if ( busySessionIds.size > 0 && @@ -2599,23 +2601,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio switchPlan?.transition.type === "restart_and_resume" ) { yield* Effect.logInfo( - "Holding the queue: the replaced provider session still runs background work", + "The queue waits: the replaced provider session still runs background work", { threadId, runId: queuedRun.id, providerSessionIds: Array.from(busySessionIds) }, ); - yield* writeSystemEvents( - projection.runs - .filter((run) => run.status === "queued") - .map((run) => ({ - type: "run.updated" as const, - threadId, - runId: run.id, - providerInstanceId: run.providerInstanceId, - occurredAt: now, - payload: { ...run, queueHeld: true }, - })), - ); + queuesWaitingOnBackgroundWork.add(threadId); return; } + queuesWaitingOnBackgroundWork.delete(threadId); const sessionsToDetach = releasableSessions.filter( (session) => !busySessionIds.has(session.id), ); @@ -7216,12 +7208,66 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio projection.thread.worktreePath === null ? latestRun : undefined; - const queueBehindRun = activeRun ?? worktreeMissingRun; - const pendingMergeBackTransfers = pendingMergeBackTransfersForThread(projection); const startsOrQueuesRun = dispatchMode.type === "defer_start" || dispatchMode.type === "start_immediately" || dispatchMode.type === "queue_after_active"; + const queuesBehindRun = (activeRun ?? worktreeMissingRun) !== undefined && startsOrQueuesRun; + const isProviderSwitch = + activeProviderThread !== undefined && + 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. + // A send that queues behind a run is planned when its run starts. + const providerSwitchPlan = + isProviderSwitch && !queuesBehindRun + ? yield* providerSwitchService + .plan({ + projection: { + ...projection, + thread: { + ...projection.thread, + modelSelection: { + ...projection.thread.modelSelection, + instanceId: activeProviderThread.providerInstanceId, + }, + }, + }, + targetModelSelection: modelSelection, + }) + .pipe(mapDispatchError(command)) + : null; + const canResumeAcrossInstances = + !freshContext && + isProviderSwitch && + activeProviderThread.nativeThreadRef !== null && + providerSwitchPlan?.transition.type === "restart_and_resume"; + const releasableSessions = queuesBehindRun + ? [] + : uniqueSessions([ + ...projection.providerSessions.filter( + (session) => + providerSwitchPlan?.releaseProviderSessionIds.includes(session.id) && + session.status !== "stopped" && + session.status !== "error", + ), + ...(freshContext ? sessionsReleasedByFreshContext(projection, activeProviderThread) : []), + ]); + // Same rules as a queued start: a hand-off or fresh session runs beside a busy session and + // leaves it up; an account switch resumes that session's native thread, so it waits in the queue. + const sessionsToDetach: Array = []; + for (const session of releasableSessions) { + if (!(yield* sessionHasBackgroundWork(session, projection.providerThreads))) { + sessionsToDetach.push(session); + } + } + const waitsForBackgroundWork = + canResumeAcrossInstances && sessionsToDetach.length < releasableSessions.length; + const queueBehindRun = + activeRun ?? + worktreeMissingRun ?? + (waitsForBackgroundWork ? (latestRun ?? undefined) : undefined); + const pendingMergeBackTransfers = pendingMergeBackTransfersForThread(projection); const shouldQueue = queueBehindRun !== undefined && startsOrQueuesRun; // An explicit pick queued ahead reaches the thread only when its run starts. A goal or // restart command that stops the active run goes first, so nothing queued applies before it. @@ -7675,58 +7721,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio projection.thread.historyOrigin === "v1_import" && contextBoundary === 0 ? yield* readHandoffItems(command.threadId, [null]) : []; - const isProviderSwitch = - activeProviderThread !== undefined && - 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 = - !freshContext && - isProviderSwitch && - activeProviderThread.nativeThreadRef !== null && - providerSwitchPlan?.transition.type === "restart_and_resume"; - const releasableSessions = uniqueSessions([ - ...projection.providerSessions.filter( - (session) => - providerSwitchPlan?.releaseProviderSessionIds.includes(session.id) && - session.status !== "stopped" && - session.status !== "error", - ), - ...(freshContext ? sessionsReleasedByFreshContext(projection, activeProviderThread) : []), - ]); - const sessionsToDetach: Array = []; - for (const detaching of releasableSessions) { - if (yield* sessionHasBackgroundWork(detaching, projection.providerThreads)) { - // A schedule has nobody to wait or press Stop; like a queued fresh start, leave it running. - if (freshContext && command.scheduledTaskId !== undefined) continue; - return yield* new OrchestratorDispatchError({ - commandId: command.commandId, - commandType: command.type, - cause: freshContext - ? "The current session is still running background agents or commands, and a fresh session would end them. Wait for them to finish, or press Stop, then send the message again." - : "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.", - }); - } - sessionsToDetach.push(detaching); - } - if ( pendingForkTransfer === undefined && pendingMergeBackTransfer === undefined && @@ -15259,11 +15253,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( command.type === "queue.resume" || command.type === "queued-run.start" || + // A send can queue with no run ahead of it: an account switch waiting on background work. + (command.type === "message.dispatch" && + committed.storedEvents.some( + ({ event }) => event.type === "run.created" && event.payload.status === "queued", + )) || (command.type === "thread.runtime-mode.set" && runtimeModeCeiling !== undefined) ) { yield* mapDispatchError(command)( startNextQueuedRun(command.threadId, { - userStarted: command.type !== "thread.runtime-mode.set", + userStarted: command.type === "queue.resume" || command.type === "queued-run.start", ...(command.type === "queued-run.start" ? { queuedRunId: command.runId } : {}), }), ); @@ -15522,6 +15521,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), eventSink.stream({ afterSequence: terminalEventsAfterSequence, eventType: "subagent.updated" }), ).pipe( + // The probe decides whether the work ended; these events are only when to ask again. + Stream.tap(({ event }) => + queuesWaitingOnBackgroundWork.has(event.threadId) + ? threadDispatch.withLock(event.threadId, startNextQueuedRun(event.threadId)).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to start a queue that waited on background work", { + threadId: event.threadId, + cause, + }), + ), + ) + : Effect.void, + ), Stream.filter(settlesChildWork), Stream.runForEach((stored) => Effect.gen(function* () { 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 e4371b54ee4b5..4ff4bc632cb8e 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -31,6 +31,7 @@ import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as Schema from "effect/Schema"; @@ -130,6 +131,8 @@ function makeTestAdapter(input: { }; readonly backgroundTasksEndWithProcess?: true; readonly backgroundWork?: Ref.Ref; + /** Ends `backgroundWork` and reports the offered thread's roster, as an idle session does. */ + readonly backgroundWorkEnded?: Queue.Queue; readonly closedSessions?: Ref.Ref>; }): ProviderAdapter.ProviderAdapterV2["Service"] { return { @@ -146,6 +149,22 @@ function makeTestAdapter(input: { Ref.update(closedSessions, (ids) => [...ids, sessionInput.providerSessionId]), ); } + if (input.backgroundWorkEnded !== undefined && input.backgroundWork !== undefined) { + const backgroundWork = input.backgroundWork; + yield* Queue.take(input.backgroundWorkEnded).pipe( + Effect.tap(() => Ref.set(backgroundWork, false)), + Effect.flatMap((providerThread) => + PubSub.publish(events, { + type: "provider_thread.updated", + driver: input.driver, + providerThread: { ...providerThread, pendingBackgroundTasks: [] }, + sessionScoped: true, + }), + ), + Effect.forever, + Effect.forkScoped, + ); + } const now = yield* DateTime.now; const providerSession: OrchestrationV2ProviderSession = { id: sessionInput.providerSessionId, @@ -2795,7 +2814,7 @@ describe("orchestration v2 provider switching", () => { { background: false, queued: false, sharedHistory: false }, ], [ - "refuses the hand-off while the previous instance still runs background work", + "keeps a busy previous session running when a direct send hands off", { background: true, queued: false, sharedHistory: false }, ], [ @@ -2803,14 +2822,19 @@ describe("orchestration v2 provider switching", () => { { background: true, queued: true, sharedHistory: false }, ], [ - "holds a queued account switch while the replaced session still runs background work", + "starts a queued account switch once the replaced session's background work ends", { background: true, queued: true, sharedHistory: true }, ], + [ + "queues a direct account switch until the replaced session's background work ends", + { background: true, queued: false, sharedHistory: true }, + ], ] as const)("%s", ([, { background, queued, sharedHistory }]) => Effect.gen(function* () { const cwd = yield* checkpointWorkspace("handoff-releases-session"); const capturedTurns = yield* Ref.make>([]); const backgroundWork = yield* Ref.make(background); + const backgroundWorkEnded = yield* Queue.unbounded(); const closedSessions = yield* Ref.make>([]); const firstStarted = yield* Deferred.make(); const releaseFirst = yield* Deferred.make(); @@ -2827,6 +2851,7 @@ describe("orchestration v2 provider switching", () => { responseByRunOrdinal: { 1: "First account complete" }, capturedTurns, backgroundWork, + backgroundWorkEnded, closedSessions, ...(queued ? { holdFirstTurn: firstStarted, releaseFirstTurn: releaseFirst } : {}), }), @@ -2897,58 +2922,37 @@ describe("orchestration v2 provider switching", () => { worktreePath: cwd, }); yield* send(1, CLAUDE_MODEL_SELECTION); - - if (queued) { - yield* Deferred.await(firstStarted); - yield* send(2, secondSelection); - const queuedRunId = (yield* orchestrator.getThreadProjection(handoffThreadId)).runs[1] - ?.id; - assert.isDefined(queuedRunId); - yield* Deferred.succeed(releaseFirst, undefined); - yield* orchestrator.streamStoredEvents.pipe( + const runCompleted = (ordinal: number) => + orchestrator.streamStoredEvents.pipe( Stream.filter( ({ event }) => event.type === "run.updated" && - event.runId === queuedRunId && - (event.payload.queueHeld === true || event.payload.status === "completed"), + event.payload.ordinal === ordinal && + event.payload.status === "completed", ), Stream.runHead, ); - yield* worker.drain(); - const afterQueue = yield* orchestrator.getThreadProjection(handoffThreadId); - const firstSession = afterQueue.providerSessions.find( + // The account switch resumes the busy session's native thread, so it waits in the + // queue without a hold and starts by itself once that session reports its work ended. + const expectSwitchAfterBackgroundWork = Effect.gen(function* () { + const waiting = yield* orchestrator.getThreadProjection(handoffThreadId); + const firstSession = waiting.providerSessions.find( (session) => session.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, ); - // Detaching would end the background agents, so the busy session must stay up. - assert.deepEqual(yield* Ref.get(closedSessions), []); - assert.equal(firstSession?.status, "ready"); - if (!sharedHistory) { - assert.deepEqual( - afterQueue.runs.map((run) => [run.providerInstanceId, run.status]), - [ - [CLAUDE_MODEL_SELECTION.instanceId, "completed"], - [secondSelection.instanceId, "completed"], - ], - ); - return; - } - // Resuming the native thread beside the busy session would fork its history. assert.deepEqual( - afterQueue.runs.map((run) => [run.status, run.queueHeld === true]), + waiting.runs.map((run) => [run.status, run.queueHeld === true]), [ ["completed", false], - ["queued", true], + ["queued", false], ], ); assert.lengthOf(yield* Ref.get(capturedTurns), 1); - yield* Ref.set(backgroundWork, false); - yield* orchestrator.dispatch({ - type: "queue.resume", - commandId: CommandId.make("command:handoff-releases-session:resume"), - threadId: handoffThreadId, - }); + assert.deepEqual(yield* Ref.get(closedSessions), []); + assert.equal(firstSession?.status, "ready"); + yield* Queue.offer(backgroundWorkEnded, waiting.providerThreads[0]!); + yield* runCompleted(2); yield* worker.drain(); - const resumed = yield* waitForIdle(handoffThreadId); + const resumed = yield* orchestrator.getThreadProjection(handoffThreadId); assert.deepEqual( resumed.runs.map((run) => [run.providerInstanceId, run.status]), [ @@ -2957,6 +2961,29 @@ describe("orchestration v2 provider switching", () => { ], ); assert.deepEqual(yield* Ref.get(closedSessions), [firstSession!.id]); + }); + + if (queued) { + yield* Deferred.await(firstStarted); + yield* send(2, secondSelection); + yield* Deferred.succeed(releaseFirst, undefined); + yield* runCompleted(sharedHistory ? 1 : 2); + yield* worker.drain(); + if (sharedHistory) return yield* expectSwitchAfterBackgroundWork; + const afterQueue = yield* orchestrator.getThreadProjection(handoffThreadId); + const firstSession = afterQueue.providerSessions.find( + (session) => session.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, + ); + // Detaching would end the background agents, so the busy session must stay up. + assert.deepEqual(yield* Ref.get(closedSessions), []); + assert.equal(firstSession?.status, "ready"); + assert.deepEqual( + afterQueue.runs.map((run) => [run.providerInstanceId, run.status]), + [ + [CLAUDE_MODEL_SELECTION.instanceId, "completed"], + [secondSelection.instanceId, "completed"], + ], + ); return; } @@ -2966,26 +2993,13 @@ describe("orchestration v2 provider switching", () => { (session) => session.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, ); assert.equal(firstSession?.status, "ready"); - - if (background) { - // 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); - } + // 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 ?? [], []); yield* send(2, secondSelection); yield* worker.drain(); + if (sharedHistory) return yield* expectSwitchAfterBackgroundWork; const afterSecond = yield* waitForIdle(handoffThreadId); yield* worker.drain(); const projection = yield* orchestrator.getThreadProjection(handoffThreadId); @@ -2998,12 +3012,15 @@ describe("orchestration v2 provider switching", () => { ], ); assert.lengthOf(afterSecond.contextHandoffs, 1); + // The detached event alone hides the session; the runtime must also close. A busy + // session keeps its background work instead. assert.deepEqual( projection.providerSessions.map((session) => session.providerInstanceId), - [secondSelection.instanceId], + background + ? [CLAUDE_MODEL_SELECTION.instanceId, secondSelection.instanceId] + : [secondSelection.instanceId], ); - // The detached event alone hides the session; the runtime must also close. - assert.deepEqual(yield* Ref.get(closedSessions), [firstSession!.id]); + assert.deepEqual(yield* Ref.get(closedSessions), background ? [] : [firstSession!.id]); }).pipe( Effect.provide( ProviderReplayHarness.layerWithRegistry( From 2ec917d7fe1189fca0f0a286bdbfccac7dc18d1e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 20:09:11 +0000 Subject: [PATCH 2/7] fix(server): a kept session's background reply drains on its own provider thread A reply buffered by a session kept across a hand-off was dispatched against the selected target, so the target drained its own empty buffer and the reply never reached the thread. The continuation now carries its owning provider thread: dispatch runs it there on the selection that session last ran, a queued one starts there too, and neither moves the thread's selection. The enabled-provider guard checks that owner. Co-authored-by: Lakshmi Tanmay Co-Authored-By: Claude Fable 5.2 --- ...laudeAccountSwitchWake.integration.test.ts | 421 ++++++++++-------- .../src/orchestration-v2/Orchestrator.ts | 29 +- .../ProviderContinuationService.ts | 3 + .../ThreadManagementService.ts | 28 +- .../bufferedProviderContinuation.ts | 28 ++ packages/contracts/src/orchestrationV2.ts | 2 + 6 files changed, 301 insertions(+), 210 deletions(-) create mode 100644 apps/server/src/orchestration-v2/bufferedProviderContinuation.ts diff --git a/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts index 0fee4be994428..504e8639da7cd 100644 --- a/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts +++ b/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts @@ -166,216 +166,259 @@ const makeAccount = Effect.fn("makeAccount")(function* (input: { }); // regression: #32296 an account switch must not close the CLI that still owes a wake reply -it.live.each(["sent after", "queued behind"] as const)( - "delivers a Claude background reply before an account switch %s the turn", - (variant) => - Effect.gen(function* () { - const cwd = yield* checkpointWorkspace("claude-account-switch-wake"); - const continuations = - yield* Queue.unbounded(); - const closed: Array = []; - const first = yield* makeAccount({ - instanceId: firstAccount.instanceId, - cwd, - continuations, - closed, - }); - const second = yield* makeAccount({ - instanceId: secondAccount.instanceId, - cwd, - continuations, - closed, +it.live.each([ + "switch sent after", + "switch queued behind", + "hand-off sent after", + "hand-off still running after", +] as const)("delivers a Claude background reply across an account %s the turn", (variant) => + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("claude-account-switch-wake"); + const continuations = + yield* Queue.unbounded(); + const closed: Array = []; + const first = yield* makeAccount({ + instanceId: firstAccount.instanceId, + cwd, + continuations, + closed, + }); + const second = yield* makeAccount({ + instanceId: secondAccount.instanceId, + cwd, + continuations, + closed, + }); + const adapters = [first.adapter, second.adapter]; + // Two accounts of one Claude install resume each other's native history; separate installs hand off. + const handsOff = variant.startsWith("hand-off"); + const registry = Layer.succeed( + ProviderAdapterRegistry.ProviderAdapterRegistryV2, + ProviderAdapterRegistry.ProviderAdapterRegistryV2.of({ + get: (instanceId) => { + const adapter = adapters.find((candidate) => candidate.instanceId === instanceId); + return adapter === undefined + ? Effect.fail( + new ProviderAdapterRegistry.ProviderAdapterRegistryLookupError({ instanceId }), + ) + : Effect.succeed(adapter); + }, + list: () => Effect.succeed(adapters.map((adapter) => adapter.instanceId)), + getMetadata: (instanceId) => + Effect.succeed({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + continuationKey: handsOff + ? `claude:${instanceId}` + : "claude:shared-native-account-history", + enabled: true, + capabilities: ClaudeProviderCapabilitiesV2, + }), + }), + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("thread:claude-account-switch-wake"); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const runStatus = (messageId: string, status: string) => + watch( + (event) => + event.type === "run.updated" && + event.payload.userMessageId === messageId && + event.payload.status === status, + ); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make("project:claude-account-switch-wake"), + title: "Account switch during a background build", + modelSelection: firstAccount, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", }); - const adapters = [first.adapter, second.adapter]; - // Two accounts of one Claude install resume each other's native history. - const registry = Layer.succeed( - ProviderAdapterRegistry.ProviderAdapterRegistryV2, - ProviderAdapterRegistry.ProviderAdapterRegistryV2.of({ - get: (instanceId) => { - const adapter = adapters.find((candidate) => candidate.instanceId === instanceId); - return adapter === undefined - ? Effect.fail( - new ProviderAdapterRegistry.ProviderAdapterRegistryLookupError({ instanceId }), - ) - : Effect.succeed(adapter); - }, - list: () => Effect.succeed(adapters.map((adapter) => adapter.instanceId)), - getMetadata: () => - Effect.succeed({ - driver: ClaudeAdapterV2.CLAUDE_PROVIDER, - continuationKey: "claude:shared-native-account-history", - enabled: true, - capabilities: ClaudeProviderCapabilitiesV2, - }), - }), - ); - yield* Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const threadId = ThreadId.make("thread:claude-account-switch-wake"); - const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => - orchestrator.streamDomainEvents.pipe( - Stream.filter(predicate), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, - ); - const runStatus = (messageId: string, status: string) => - watch( - (event) => - event.type === "run.updated" && - event.payload.userMessageId === messageId && - event.payload.status === status, - ); - yield* orchestrator.dispatch({ - type: "thread.create", - commandId: CommandId.make("create"), + const queued = variant === "switch queued behind"; + const send = (ordinal: number, modelSelection: ModelSelection) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`send-${ordinal}`), threadId, - projectId: ProjectId.make("project:claude-account-switch-wake"), - title: "Account switch during a background build", - modelSelection: firstAccount, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: cwd, + messageId: MessageId.make(`send-${ordinal}`), + text: `Prompt ${ordinal}`, + attachments: [], + modelSelection, + dispatchMode: { type: queued ? "queue_after_active" : "start_immediately" }, createdBy: "user", creationSource: "web", }); - const queued = variant === "queued behind"; - const send = (ordinal: number, modelSelection: ModelSelection) => - orchestrator.dispatch({ + + const firstRunning = yield* watch( + (event) => event.type === "provider-turn.updated" && event.payload.status === "running", + ); + yield* send(1, firstAccount); + yield* worker.drain(); + yield* Fiber.join(firstRunning); + if (queued) yield* send(2, secondAccount); + const firstSettling = yield* runStatus("send-1", "waiting"); + const firstProcess = first.processes[0]!; + yield* Queue.offer(firstProcess, backgroundBashStarted); + yield* Queue.offer(firstProcess, resultFrame("turn-1", "Started the build.")); + yield* Fiber.join(firstSettling); + yield* worker.drain(); + + if (!queued) yield* send(2, secondAccount); + yield* worker.drain(); + const deliverWake = (afterDispatch: Effect.Effect = Effect.void) => + Effect.gen(function* () { + yield* Queue.offer(firstProcess, bashNotification); + yield* Queue.offer(firstProcess, assistantFrame("wake-assistant", WAKE_REPLY)); + yield* Queue.offer(firstProcess, resultFrame("wake-result", WAKE_REPLY, true)); + const request = yield* Queue.take(continuations); + const wakeSettling = yield* runStatus("wake", "waiting"); + yield* orchestrator.dispatch({ type: "message.dispatch", - commandId: CommandId.make(`send-${ordinal}`), + commandId: CommandId.make("wake"), threadId, - messageId: MessageId.make(`send-${ordinal}`), - text: `Prompt ${ordinal}`, + messageId: MessageId.make("wake"), + text: request.detail ?? "Background task completed.", attachments: [], - modelSelection, - dispatchMode: { type: queued ? "queue_after_active" : "start_immediately" }, - createdBy: "user", - creationSource: "web", + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "provider", + continuationProviderThreadId: request.providerThreadId, + notification: request.notification ?? { + source: { kind: "background_task" }, + outcome: "updated", + summary: "Background activity updated", + }, }); - - const firstRunning = yield* watch( - (event) => event.type === "provider-turn.updated" && event.payload.status === "running", - ); - yield* send(1, firstAccount); - yield* worker.drain(); - yield* Fiber.join(firstRunning); - if (queued) yield* send(2, secondAccount); - const firstSettling = yield* runStatus("send-1", "waiting"); - const firstProcess = first.processes[0]!; - yield* Queue.offer(firstProcess, backgroundBashStarted); - yield* Queue.offer(firstProcess, resultFrame("turn-1", "Started the build.")); - yield* Fiber.join(firstSettling); - yield* worker.drain(); - - if (!queued) yield* send(2, secondAccount); - yield* worker.drain(); - const waiting = yield* orchestrator.getThreadProjection(threadId); - assert.deepEqual( - waiting.runs.map((run) => [run.providerInstanceId, run.status]), - [ - [firstAccount.instanceId, "completed"], - [secondAccount.instanceId, "queued"], - ], - ); - - // The settled roster gives queue resumption its chance before the notification. - const rosterSettled = yield* orchestrator - .streamStoredEventsFrom({ - threadId, - afterSequence: yield* (yield* EventSink.EventSinkV2).latestSequence(), - eventType: "provider-thread.updated", - }) - .pipe(Stream.runHead, Effect.forkScoped); - yield* Queue.offer(firstProcess, emptyRoster); - yield* Fiber.join(rosterSettled); - yield* worker.drain(); - yield* orchestrator.resumeQueuedRuns; - yield* worker.drain(); - const afterRoster = yield* orchestrator.getThreadProjection(threadId); - assert.deepEqual(closed, [], "the notification is still owed by the first account"); - assert.equal(afterRoster.runs[1]?.status, "queued"); - assert.equal( - afterRoster.providerSessions.find( - (session) => session.providerInstanceId === firstAccount.instanceId, - )?.status, - "ready", - ); - - // The notification and wake reply arrive on the still-open process. - yield* Queue.offer(firstProcess, bashNotification); - yield* Queue.offer(firstProcess, assistantFrame("wake-assistant", WAKE_REPLY)); - yield* Queue.offer(firstProcess, resultFrame("wake-result", WAKE_REPLY, true)); - const request = yield* Queue.take(continuations); - const wakeSettling = yield* runStatus("wake", "waiting"); - const switched = yield* runStatus("send-2", "running"); - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make("wake"), - threadId, - messageId: MessageId.make("wake"), - text: request.detail ?? "Background task completed.", - attachments: [], - dispatchMode: { type: "queue_after_active" }, - createdBy: "agent", - creationSource: "provider", - notification: request.notification ?? { - source: { kind: "background_task" }, - outcome: "updated", - summary: "Background activity updated", - }, + yield* afterDispatch; + yield* worker.drain(); + yield* Fiber.join(wakeSettling); + yield* worker.drain(); }); - yield* worker.drain(); - yield* Fiber.join(wakeSettling); - yield* worker.drain(); - yield* Fiber.join(switched); - yield* worker.drain(); - + if (handsOff) { + // The second account runs beside the first, whose process still owes its reply. + const finishSecond = Effect.gen(function* () { + const secondSettling = yield* runStatus("send-2", "waiting"); + yield* Queue.offer(second.processes[0]!, resultFrame("turn-2", "Handoff complete.")); + yield* Fiber.join(secondSettling); + yield* worker.drain(); + }); + assert.deepEqual(closed, []); + // A reply that arrives during the second account's turn queues, then starts on its owner. + if (variant === "hand-off sent after") { + yield* finishSecond; + yield* deliverWake(); + } else { + yield* deliverWake(finishSecond); + } const after = yield* orchestrator.getThreadProjection(threadId); assert.isTrue( after.messages.some((message) => message.text === WAKE_REPLY), "the wake reply must reach the thread", ); - const wakeRun = after.runs.find((run) => run.userMessageId === "wake"); - assert.equal(wakeRun?.providerInstanceId, firstAccount.instanceId); - assert.equal(wakeRun?.status, "completed"); assert.equal( - after.runs.find((run) => run.userMessageId === "send-2")?.providerInstanceId, - secondAccount.instanceId, + after.runs.find((run) => run.userMessageId === "wake")?.providerInstanceId, + firstAccount.instanceId, ); - // Only now may the switch release the first account's process. - assert.deepEqual(closed, [firstAccount.instanceId]); - assert.lengthOf(second.processes, 1); - }).pipe( - Effect.provide( - ProviderReplayHarness.layerWithRegistry( - { - name: "claude-account-switch-wake", - runtimePolicyOverride: { - cwd, - approvalPolicy: "never", - sandboxPolicy: { - type: "readOnly", - access: { type: "fullAccess" }, - networkAccess: false, - }, - }, - }, - registry, - { runEffectWorker: false }, - ), - ), + // The reply drains on its owner; the user's pick stays on the second account. + assert.deepEqual(after.thread.modelSelection, secondAccount); + assert.equal(after.thread.providerInstanceId, secondAccount.instanceId); + assert.notInclude(closed, secondAccount.instanceId); + return; + } + const waiting = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + waiting.runs.map((run) => [run.providerInstanceId, run.status]), + [ + [firstAccount.instanceId, "completed"], + [secondAccount.instanceId, "queued"], + ], ); + + // The settled roster gives queue resumption its chance before the notification. + const rosterSettled = yield* orchestrator + .streamStoredEventsFrom({ + threadId, + afterSequence: yield* (yield* EventSink.EventSinkV2).latestSequence(), + eventType: "provider-thread.updated", + }) + .pipe(Stream.runHead, Effect.forkScoped); + yield* Queue.offer(firstProcess, emptyRoster); + yield* Fiber.join(rosterSettled); + yield* worker.drain(); + yield* orchestrator.resumeQueuedRuns; + yield* worker.drain(); + const afterRoster = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual(closed, [], "the notification is still owed by the first account"); + assert.equal(afterRoster.runs[1]?.status, "queued"); + assert.equal( + afterRoster.providerSessions.find( + (session) => session.providerInstanceId === firstAccount.instanceId, + )?.status, + "ready", + ); + + // The notification and wake reply arrive on the still-open process. + const switched = yield* runStatus("send-2", "running"); + yield* deliverWake(); + yield* Fiber.join(switched); + yield* worker.drain(); + + const after = yield* orchestrator.getThreadProjection(threadId); + assert.isTrue( + after.messages.some((message) => message.text === WAKE_REPLY), + "the wake reply must reach the thread", + ); + const wakeRun = after.runs.find((run) => run.userMessageId === "wake"); + assert.equal(wakeRun?.providerInstanceId, firstAccount.instanceId); + assert.equal(wakeRun?.status, "completed"); + assert.equal( + after.runs.find((run) => run.userMessageId === "send-2")?.providerInstanceId, + secondAccount.instanceId, + ); + // Only now may the switch release the first account's process. + assert.deepEqual(closed, [firstAccount.instanceId]); + assert.lengthOf(second.processes, 1); }).pipe( Effect.provide( - Layer.mergeAll( - IdAllocator.layer, - McpProviderSessions.layer, - NodeServices.layer, - NodeCrypto.layer, + ProviderReplayHarness.layerWithRegistry( + { + name: "claude-account-switch-wake", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registry, + { runEffectWorker: false }, ), ), + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), ), + ), ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 1fbfebd3dbed3..03127a1f6a5f7 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -81,6 +81,7 @@ import { import * as KeyedLock from "@t3tools/shared/KeyedLock"; import { modelSelectionsEqual } from "@t3tools/shared/model"; import { modelSwitchSince } from "./modelSwitch.ts"; +import { resolveBufferedProviderContinuation } from "./bufferedProviderContinuation.ts"; import { combinePullRequestWatchWakes, type PullRequestWatchWake } from "./pullRequestWatch.ts"; import { recordThreadOpen } from "@t3tools/shared/paletteSearch"; import { @@ -2501,8 +2502,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const activeProviderThread = dispatchProviderThread(projection, queuedMessage, freshContext); const contextBoundary = freshContext ? ordinal : contextBoundaryOrdinal(projection); // A run queued before a fresh run started is bound to the conversation that run left behind. + // Buffered output can only drain on the session that holds it. + const bufferedContinuation = isBufferedProviderNotification(queuedMessage); const followsLatestContext = !freshContext && + !bufferedContinuation && (followsCurrent || providerThreadPredatesBoundary(storedProviderThread, contextBoundary)); const targetProviderThread = followsLatestContext ? activeProviderThread?.providerInstanceId === queuedRun.providerInstanceId && @@ -2539,10 +2543,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio createdAt: now, updatedAt: now, }; - const selectionChanged = !modelSelectionsEqual( - projection.thread.modelSelection, - queuedRun.modelSelection, - ); + const selectionChanged = + !bufferedContinuation && + !modelSelectionsEqual(projection.thread.modelSelection, queuedRun.modelSelection); // The picker may already have moved ahead of the native conversation. const switchFromActiveProvider = followsCurrent && @@ -6850,7 +6853,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ]); projection = applyToProjection(projection, event); } - let modelSelection = command.modelSelection ?? projection.thread.modelSelection; + // A session kept across a hand-off still replies; its reply must not move the user's pick. + const continuation = resolveBufferedProviderContinuation(projection, command); + let modelSelection = + continuation?.modelSelection ?? command.modelSelection ?? projection.thread.modelSelection; let dispatchMode = resolveMessageDispatchIntent( projection, command.dispatchMode, @@ -7191,11 +7197,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return; } - const activeProviderThread = dispatchProviderThread( - projection, - { ...command, delegatedCompletion }, - freshContext, - ); + const activeProviderThread = + continuation?.providerThread ?? + dispatchProviderThread(projection, { ...command, delegatedCompletion }, freshContext); const activeRun = projection.runs.find(isActiveRun); // Messages after failed or stopped worktree setup must wait for a worktree, // rather than start in the project checkout. @@ -7680,8 +7684,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); const now = yield* DateTime.now; if ( - !modelSelectionsEqual(projection.thread.modelSelection, modelSelection) || - projection.thread.providerInstanceId !== modelSelection.instanceId + continuation === undefined && + (!modelSelectionsEqual(projection.thread.modelSelection, modelSelection) || + projection.thread.providerInstanceId !== modelSelection.instanceId) ) { yield* emit( events, diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts index 01f6e0acb989d..84a963fc005b7 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -258,6 +258,9 @@ export const layer = Layer.effectDiscard( createdBy: "agent", // Only buffered wakes use creationSource "provider"; Claude would drain an empty buffer instead of prompting message_text. creationSource: request.delivery === "message_text" ? "server" : "provider", + ...(request.delivery === "message_text" + ? {} + : { continuationProviderThreadId: request.providerThreadId }), }) .pipe(Effect.tap((dispatched) => releaseIfNeverStarted(request, messageId, dispatched))); // Catch after dispatchIfCurrent so adapters see raw failures first; clearIfCurrent is idempotent and covers others. diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 00e81061f7a13..dbd343804e3b0 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -69,6 +69,7 @@ import { DispatchModeLimit, exceededDispatchModeLimit } from "./DispatchModeLimi import { projectTurnItemForDetail } from "./WireProjection.ts"; import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts"; import * as ServerSettings from "../serverSettings.ts"; +import { resolveBufferedProviderContinuation } from "./bufferedProviderContinuation.ts"; import { ThreadWorktreeRestore } from "../workspace/restoreMissingWorktree.ts"; export type ThreadManagementSendMode = "auto" | "queue" | "steer" | "restart"; @@ -750,15 +751,24 @@ const make = Effect.gen(function* () { const assertDispatchTargetEnabled = (command: OrchestrationV2ServerCommand) => { switch (command.type) { case "message.dispatch": - return orchestrator.getThreadRecords(command.threadId, []).pipe( - Effect.flatMap((projection) => - assertProviderInstanceEnabled({ - projectId: projection.thread.projectId, - commandId: command.commandId, - instanceId: (command.modelSelection ?? projection.thread.modelSelection).instanceId, - }), - ), - ); + return orchestrator + .getThreadRecords( + command.threadId, + command.continuationProviderThreadId === undefined ? [] : ["providerThreads", "runs"], + ) + .pipe( + Effect.flatMap((projection) => + assertProviderInstanceEnabled({ + projectId: projection.thread.projectId, + commandId: command.commandId, + instanceId: ( + resolveBufferedProviderContinuation(projection, command)?.modelSelection ?? + command.modelSelection ?? + projection.thread.modelSelection + ).instanceId, + }), + ), + ); case "delegated_task.request": return orchestrator.getThreadRecords(command.parentThreadId, []).pipe( Effect.flatMap((parent) => diff --git a/apps/server/src/orchestration-v2/bufferedProviderContinuation.ts b/apps/server/src/orchestration-v2/bufferedProviderContinuation.ts new file mode 100644 index 0000000000000..3121624853231 --- /dev/null +++ b/apps/server/src/orchestration-v2/bufferedProviderContinuation.ts @@ -0,0 +1,28 @@ +import type { + OrchestrationV2ServerCommand, + OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; + +/** Buffered output drains on the session that produced it, on the selection that session last ran. */ +export function resolveBufferedProviderContinuation( + projection: Pick, + command: Extract, +) { + if ( + command.continuationProviderThreadId === undefined || + command.createdBy !== "agent" || + command.creationSource !== "provider" || + command.notification === undefined + ) { + return undefined; + } + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === command.continuationProviderThreadId, + ); + const modelSelection = projection.runs.findLast( + (run) => run.providerThreadId === providerThread?.id, + )?.modelSelection; + return providerThread === undefined || modelSelection === undefined + ? undefined + : { providerThread, modelSelection }; +} diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index bf6b53666f4d7..2cc0f96438d36 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3305,6 +3305,8 @@ export const OrchestrationV2Command = Schema.Union([ Schema.Struct({ type: Schema.Literal("message.dispatch"), notification: Schema.optional(OrchestrationV2Notification), + /** The provider thread whose session buffered this output; it drains there, not on the selection. */ + continuationProviderThreadId: Schema.optional(ProviderThreadId), /** Sends this pending guest message; the server uses the guest's text, not `text`. */ guestMessageId: Schema.optional(GuestMessageId), ...OrchestrationV2CreationFields, From 159db2f755a2c2ab38c9c0fb0078fd1c8d50e9e4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 20:20:25 +0000 Subject: [PATCH 3/7] test(server): type the wake helper's follow-up effect Co-Authored-By: Claude Fable 5.2 --- .../ClaudeAccountSwitchWake.integration.test.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts index 504e8639da7cd..e7ff0cb8ed764 100644 --- a/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts +++ b/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts @@ -278,7 +278,7 @@ it.live.each([ if (!queued) yield* send(2, secondAccount); yield* worker.drain(); - const deliverWake = (afterDispatch: Effect.Effect = Effect.void) => + const deliverWake = (afterDispatch: Effect.Effect) => Effect.gen(function* () { yield* Queue.offer(firstProcess, bashNotification); yield* Queue.offer(firstProcess, assistantFrame("wake-assistant", WAKE_REPLY)); @@ -319,7 +319,7 @@ it.live.each([ // A reply that arrives during the second account's turn queues, then starts on its owner. if (variant === "hand-off sent after") { yield* finishSecond; - yield* deliverWake(); + yield* deliverWake(Effect.void); } else { yield* deliverWake(finishSecond); } @@ -372,7 +372,7 @@ it.live.each([ // The notification and wake reply arrive on the still-open process. const switched = yield* runStatus("send-2", "running"); - yield* deliverWake(); + yield* deliverWake(Effect.void); yield* Fiber.join(switched); yield* worker.drain(); From 960f893db7c878f02cca6508b336a445974b89fa Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 20:53:44 +0000 Subject: [PATCH 4/7] fix(server): a waiting account switch also retries when a tool node settles A Codex tool that outlives its turn reports its end with a node update alone, so the queue that waited on it was never asked again. The retry now also listens to settled tool-call nodes, for waiting threads only. Co-Authored-By: Claude Fable 5.2 --- .../src/orchestration-v2/Orchestrator.ts | 23 ++++++++++++++----- 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 03127a1f6a5f7..8f26f5e404741 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -15520,15 +15520,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return providerThreadsWithBackgroundWork.delete(event.payload.id); }; yield* Stream.merge( - eventSink.stream({ - afterSequence: terminalEventsAfterSequence, - eventType: "provider-thread.updated", - }), - eventSink.stream({ afterSequence: terminalEventsAfterSequence, eventType: "subagent.updated" }), + Stream.merge( + eventSink.stream({ + afterSequence: terminalEventsAfterSequence, + eventType: "provider-thread.updated", + }), + eventSink.stream({ + afterSequence: terminalEventsAfterSequence, + eventType: "subagent.updated", + }), + ), + // A Codex tool that outlives its turn settles with a node update alone. + eventSink + .stream({ afterSequence: terminalEventsAfterSequence, eventType: "node.updated" }) + .pipe(Stream.filter(({ event }) => queuesWaitingOnBackgroundWork.has(event.threadId))), ).pipe( // The probe decides whether the work ended; these events are only when to ask again. Stream.tap(({ event }) => - queuesWaitingOnBackgroundWork.has(event.threadId) + queuesWaitingOnBackgroundWork.has(event.threadId) && + (event.type !== "node.updated" || + (event.payload.kind === "tool_call" && !isOrchestrationV2WorkActive(event.payload.status))) ? threadDispatch.withLock(event.threadId, startNextQueuedRun(event.threadId)).pipe( Effect.catchCause((cause) => Effect.logWarning("Failed to start a queue that waited on background work", { From bf3fa4e60e6c1b9d3ef427c65c641ede651ab21d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 21:11:04 +0000 Subject: [PATCH 5/7] fix(server): a buffered reply drains on the model its session last ran A send queued on the owning provider thread and never started was picked as that session's latest selection. Only started runs count now. Co-Authored-By: Claude Fable 5.2 --- .../bufferedProviderContinuation.test.ts | 48 +++++++++++++++++++ .../bufferedProviderContinuation.ts | 17 +++++-- 2 files changed, 62 insertions(+), 3 deletions(-) create mode 100644 apps/server/src/orchestration-v2/bufferedProviderContinuation.test.ts diff --git a/apps/server/src/orchestration-v2/bufferedProviderContinuation.test.ts b/apps/server/src/orchestration-v2/bufferedProviderContinuation.test.ts new file mode 100644 index 0000000000000..a494b97f8fac3 --- /dev/null +++ b/apps/server/src/orchestration-v2/bufferedProviderContinuation.test.ts @@ -0,0 +1,48 @@ +import { assert, it } from "@effect/vitest"; +import { + CommandId, + MessageId, + ProviderInstanceId, + ProviderThreadId, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; + +import { resolveBufferedProviderContinuation } from "./bufferedProviderContinuation.ts"; + +it("drains on the selection the owning session last ran, not one queued and never started", () => { + const providerThreadId = ProviderThreadId.make("provider-thread:owner"); + const instanceId = ProviderInstanceId.make("claudeAgent"); + const ran = { instanceId, model: "claude-sonnet-4-6" }; + const continuation = resolveBufferedProviderContinuation( + { + providerThreads: [{ id: providerThreadId }], + runs: [ + { providerThreadId, startedAt: DateTime.makeUnsafe(0), modelSelection: ran }, + { + providerThreadId, + startedAt: null, + modelSelection: { instanceId, model: "claude-opus-4-6" }, + }, + ], + }, + { + type: "message.dispatch", + commandId: CommandId.make("command:wake"), + threadId: ThreadId.make("thread:owner"), + messageId: MessageId.make("message:wake"), + text: "Background activity updated", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "provider", + continuationProviderThreadId: providerThreadId, + notification: { + source: { kind: "background_task" }, + outcome: "updated", + summary: "Background activity updated", + }, + }, + ); + assert.deepEqual(continuation?.modelSelection, ran); +}); diff --git a/apps/server/src/orchestration-v2/bufferedProviderContinuation.ts b/apps/server/src/orchestration-v2/bufferedProviderContinuation.ts index 3121624853231..4fd4e1e2bfe49 100644 --- a/apps/server/src/orchestration-v2/bufferedProviderContinuation.ts +++ b/apps/server/src/orchestration-v2/bufferedProviderContinuation.ts @@ -4,8 +4,18 @@ import type { } from "@t3tools/contracts"; /** Buffered output drains on the session that produced it, on the selection that session last ran. */ -export function resolveBufferedProviderContinuation( - projection: Pick, +export function resolveBufferedProviderContinuation< + ProviderThread extends Pick, +>( + projection: { + readonly providerThreads: ReadonlyArray; + readonly runs: ReadonlyArray< + Pick< + OrchestrationV2ThreadProjection["runs"][number], + "providerThreadId" | "startedAt" | "modelSelection" + > + >; + }, command: Extract, ) { if ( @@ -20,7 +30,8 @@ export function resolveBufferedProviderContinuation( (candidate) => candidate.id === command.continuationProviderThreadId, ); const modelSelection = projection.runs.findLast( - (run) => run.providerThreadId === providerThread?.id, + // A send queued on this thread and never started names a model the session never ran. + (run) => run.providerThreadId === providerThread?.id && run.startedAt !== null, )?.modelSelection; return providerThread === undefined || modelSelection === undefined ? undefined From e6b26bcfe242c8e03a576d94af9f75d530cc0986 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 21:19:58 +0000 Subject: [PATCH 6/7] fix(server): work that settles during the queue's probe still retries the start The thread was registered as waiting only after the probe reported busy, so a settlement delivered in between was dropped. It registers before the probe. Co-Authored-By: Claude Fable 5.2 --- apps/server/src/orchestration-v2/Orchestrator.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 8f26f5e404741..ec4415b2b1d59 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2590,6 +2590,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(freshContext ? sessionsReleasedByFreshContext(projection, activeProviderThread) : []), ]); const busySessionIds = new Set(); + // Registered before the probe: work that settles while it runs must still retry this start. + queuesWaitingOnBackgroundWork.add(threadId); for (const session of releasableSessions) { if (yield* sessionHasBackgroundWork(session, projection.providerThreads)) { busySessionIds.add(session.id); @@ -2607,7 +2609,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio "The queue waits: the replaced provider session still runs background work", { threadId, runId: queuedRun.id, providerSessionIds: Array.from(busySessionIds) }, ); - queuesWaitingOnBackgroundWork.add(threadId); return; } queuesWaitingOnBackgroundWork.delete(threadId); From 50ebeacef3f54c1b9264f94ada4c42f2fa5b291f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 21:27:26 +0000 Subject: [PATCH 7/7] test(server): the send after a late reply stays on the picked account Co-Authored-By: Claude Fable 5.2 --- .../ClaudeAccountSwitchWake.integration.test.ts | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts index e7ff0cb8ed764..106a46c63ea03 100644 --- a/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts +++ b/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts @@ -336,6 +336,20 @@ it.live.each([ assert.deepEqual(after.thread.modelSelection, secondAccount); assert.equal(after.thread.providerInstanceId, secondAccount.instanceId); assert.notInclude(closed, secondAccount.instanceId); + // The next send continues on the picked account, not on the session that just replied. + const thirdRunning = yield* runStatus("send-3", "running"); + yield* send(3, secondAccount); + yield* worker.drain(); + yield* Fiber.join(thirdRunning); + const next = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === "send-3", + ); + assert.equal(next?.providerInstanceId, secondAccount.instanceId); + assert.equal( + next?.providerThreadId, + after.runs.find((run) => run.userMessageId === "send-2")?.providerThreadId, + ); + assert.lengthOf(second.processes, 1); return; } const waiting = yield* orchestrator.getThreadProjection(threadId);