From 46b170942e2789e7217dc2c2b2df8079b7e04ae6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 12:54:31 +0000 Subject: [PATCH 1/4] test(server): reproduce issue 16000 background work lost on runtime-mode, instance and worktree changes Co-Authored-By: Claude Opus 5.5 --- .../SelectionRestart.integration.test.ts | 163 ++++++++++++++++++ 1 file changed, 163 insertions(+) diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index 03cdf5e85a642..3a76388050d94 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -774,3 +774,166 @@ it.live.each(["active", "idle", "selection-command", "pooled", "separate-home"] }), ), ); + +it.live.each(["runtime-mode", "provider-instance", "workspace"] as const)( + "keeps a session with running background work alive through a %s change", + (change) => + Effect.scoped( + Effect.gen(function* () { + const name = `background-work-${change}`; + const cwd = yield* checkpointWorkspace(name); + const otherCwd = yield* checkpointWorkspace(`${name}-other`); + const threadId = ThreadId.make(`thread:${name}`); + const targetId = ProviderInstanceId.make("codex-background-target"); + const state = yield* Ref.make({ + activeTurn: null, + opened: [], + started: [], + closedSessionCount: 0, + failedReplacementOpen: true, + }); + // Claude runs background agents and shells inside its CLI process and + // cannot switch runtime mode in place. + const capabilities: OrchestrationV2ProviderCapabilities = { + ...exclusiveCapabilities, + sessions: { + ...exclusiveCapabilities.sessions, + supportsRuntimeModeSwitchInSession: false, + }, + }; + const adapters = [providerInstanceId, targetId].map((instanceId) => { + const base = makeRestartAdapter(state, capabilities, instanceId); + return { + ...base, + openSession: (input) => + base.openSession(input).pipe( + Effect.map((session) => ({ + ...session, + backgroundTasksEndWithProcess: true as const, + hasPendingBackgroundWork: Effect.succeed(true), + hasPendingBackgroundWorkForThread: () => Effect.succeed(true), + })), + ), + } satisfies ProviderAdapter.ProviderAdapterV2["Service"]; + }); + const layerRegistry = Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistryV2, { + get: (instanceId) => + Effect.succeed(adapters.find((adapter) => adapter.instanceId === instanceId)!), + list: () => Effect.succeed([providerInstanceId, targetId]), + getMetadata: (instanceId) => + Effect.succeed({ + driver, + continuationKey: `codex:home:/${instanceId}`, + enabled: true, + capabilities, + }), + }); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const sink = yield* EventSink.EventSinkV2; + const selection = { ...initialSelection, model: "complete" }; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${name}:create`), + threadId, + projectId: ProjectId.make(`project:${name}`), + title: name, + modelSelection: selection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const completed = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => event.type === "run.updated" && event.payload.status === "completed", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${name}:first`), + threadId, + messageId: MessageId.make(`${name}:first`), + text: "Start a background subagent and a background shell", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(completed); + yield* worker.drain(); + const before = yield* orchestrator.getThreadProjection(threadId); + const providerThread = before.providerThreads.find( + (thread) => thread.id === before.thread.activeProviderThreadId, + )!; + assert.deepEqual( + before.providerSessions + .filter((session) => session.id === providerThread.providerSessionId) + .map((session) => session.status), + ["ready"], + ); + yield* sink.write({ + events: [ + { + id: EventId.make(`${name}:roster`), + type: "provider-thread.updated", + threadId, + occurredAt: yield* DateTime.now, + payload: { + ...providerThread, + pendingBackgroundTasks: [ + { taskId: "subagent", description: "Run 24 steps", kind: "subagent" }, + { taskId: "counter", description: "Append a counter", kind: "command" }, + ], + }, + }, + ], + }); + + // Refusing or deferring the change are both acceptable; ending the + // provider process while its background work runs is not. + yield* Effect.exit( + orchestrator.dispatch( + change === "runtime-mode" + ? { + type: "thread.runtime-mode.set", + commandId: CommandId.make(`${name}:change`), + threadId, + runtimeMode: "auto-accept-edits", + } + : change === "provider-instance" + ? { + type: "provider.switch", + commandId: CommandId.make(`${name}:change`), + threadId, + modelSelection: { instanceId: targetId, model: "complete" }, + } + : { + type: "thread.metadata.update", + commandId: CommandId.make(`${name}:change`), + threadId, + worktreePath: otherCwd, + }, + ), + ); + yield* worker.drain(); + + const after = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + after.providerSessions + .filter((session) => session.id === providerThread.providerSessionId) + .map((session) => session.status), + ["ready"], + ); + assert.equal((yield* Ref.get(state)).closedSessionCount, 0); + }).pipe(Effect.provide(ProviderReplayHarness.layerWithRegistry({ name }, layerRegistry))); + }), + ), +); From 48f89ea7a5c3d0e2f1fbbd486b1b7827382355e5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 13:05:51 +0000 Subject: [PATCH 2/4] fix(server): settings changes no longer end running background agents and shells Changing runtime mode, provider instance, or worktree detached the provider session even while its process ran the thread's background agents and shells, so that work died silently and the UI kept showing it as running (#16000). The orchestrator now asks the live adapter whether a session the change would detach still has this thread's background work, and refuses the change with a retryable message instead. Archive, settle, and Stop still end the work. The web branch picker no longer stops the session itself before a worktree change; the server already detaches on that path, and the eager stop bypassed the guard. Mobile outbox settings sync now uses a command id per retry attempt, so a refused change can apply once the work finishes. Co-Authored-By: Claude Opus 5.5 --- .../src/state/use-thread-outbox-drain.ts | 22 ++++- .../src/orchestration-v2/Orchestrator.ts | 93 ++++++++++++++----- .../BranchToolbarBranchSelector.tsx | 7 -- 3 files changed, 89 insertions(+), 33 deletions(-) diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index 47c76d72136a8..77880747dfb8e 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -183,8 +183,14 @@ function findCreationProject( ); } -function settingsCommandId(message: QueuedThreadMessage, setting: string): CommandId { - return CommandId.make(`${message.commandId}:${setting}`); +// A refused settings change stays refused under its command id, so each retry +// needs its own: the server can refuse one while background work runs. +function settingsCommandId( + message: QueuedThreadMessage, + setting: string, + attempt: number, +): CommandId { + return CommandId.make(`${message.commandId}:${setting}:${attempt}`); } export async function prepareQueuedMessageAttachments( @@ -758,7 +764,11 @@ export function useThreadOutboxDrain(): void { const runtimeResult = await setThreadRuntimeMode({ environmentId: queuedMessage.environmentId, input: { - commandId: settingsCommandId(queuedMessage, "runtime-mode"), + commandId: settingsCommandId( + queuedMessage, + "runtime-mode", + retryAttemptRef.current.get(queuedMessage.messageId) ?? 0, + ), threadId: queuedMessage.threadId, runtimeMode: settings.runtimeMode, createdAt: queuedMessage.createdAt, @@ -775,7 +785,11 @@ export function useThreadOutboxDrain(): void { const interactionResult = await setThreadInteractionMode({ environmentId: queuedMessage.environmentId, input: { - commandId: settingsCommandId(queuedMessage, "interaction-mode"), + commandId: settingsCommandId( + queuedMessage, + "interaction-mode", + retryAttemptRef.current.get(queuedMessage.messageId) ?? 0, + ), threadId: queuedMessage.threadId, interactionMode: settings.interactionMode, createdAt: queuedMessage.createdAt, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 77533f593e9a7..868865456c112 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2650,6 +2650,36 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, ); + const hasLiveBackgroundWork = Effect.fnUntraced(function* ( + threadId: ThreadId, + sessionIds: ReadonlySet, + ) { + let providerThreads: ReadonlyArray | undefined; + for (const sessionId of sessionIds) { + const session = yield* providerSessions + .get(sessionId) + .pipe(Effect.orElseSucceed(() => Option.none())); + if (Option.isNone(session)) continue; + const { hasPendingBackgroundWork, hasPendingBackgroundWorkForThread } = session.value; + // A pooled session can serve other threads; only this thread's work counts. + const probes = + hasPendingBackgroundWorkForThread === undefined + ? hasPendingBackgroundWork === undefined + ? [] + : [hasPendingBackgroundWork] + : (providerThreads ??= (yield* projectionStore.getThreadRecords(threadId, [ + "providerThreads", + ])).providerThreads) + .filter((thread) => thread.providerSessionId === sessionId) + .map(hasPendingBackgroundWorkForThread); + for (const probe of probes) { + // A failed probe cannot confirm that the work ended. + if (yield* probe.pipe(Effect.catchCause(() => Effect.succeed(true)))) return true; + } + } + return false; + }); + const dispatchThreadMutation = Effect.fn("orchestrationV2.dispatch.threadMutation")(function* ( command: Extract< OrchestrationV2ServerCommand, @@ -2990,6 +3020,47 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }) : null; + // Settle joins archive here: both mean "done with this + // thread", so a live provider session must not keep running background + // work (PR monitors, dev servers, subagent fleets) after any of them + // lands. The settle guard above already rejects active or blocked runs, + // so for settle this only ever stops an idle session; commands are + // decided serially against the projection, so a turn start that + // re-engages the thread cannot race this detach. + const detachSessionIds = new Set( + command.type === "thread.archive" || command.type === "thread.settle" + ? (providerContext?.providerSessions ?? []).map((session) => session.id) + : command.type === "thread.metadata.update" && + command.worktreePath !== undefined && + command.worktreePath !== thread.worktreePath + ? (providerContext?.providerSessions ?? []).map((session) => session.id) + : command.type === "thread.runtime-mode.set" + ? (providerContext?.providerSessions ?? []) + .filter( + (session) => !session.capabilities.sessions.supportsRuntimeModeSwitchInSession, + ) + .map((session) => session.id) + : (providerSwitchPlan?.releaseProviderSessionIds ?? []), + ); + // A settings change must not end the process running this thread's + // background agents and shells (#16000). Ask the live adapter: the + // projected roster is not authoritative while a turn runs. Archive and + // settle are deliberate ends, like Stop. + if ( + command.type !== "thread.archive" && + command.type !== "thread.settle" && + detachSessionIds.size > 0 && + (yield* hasLiveBackgroundWork(command.threadId, detachSessionIds).pipe( + mapDispatchError(command), + )) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} has background work running that this change would end. Try again once it finishes, or stop the thread first.`, + }); + } + const now = yield* DateTime.now; if (command.type === "thread.metadata.update" && command.limitRecovery != null) { const projection = yield* loadProjectionForCommand( @@ -3699,28 +3770,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } - // Settle joins archive here: both mean "done with this - // thread", so a live provider session must not keep running background - // work (PR monitors, dev servers, subagent fleets) after any of them - // lands. The settle guard above already rejects active or blocked runs, - // so for settle this only ever stops an idle session; commands are - // decided serially against the projection, so a turn start that - // re-engages the thread cannot race this detach. - const detachSessionIds = new Set( - command.type === "thread.archive" || command.type === "thread.settle" - ? (providerContext?.providerSessions ?? []).map((session) => session.id) - : command.type === "thread.metadata.update" && - command.worktreePath !== undefined && - command.worktreePath !== thread.worktreePath - ? (providerContext?.providerSessions ?? []).map((session) => session.id) - : command.type === "thread.runtime-mode.set" - ? (providerContext?.providerSessions ?? []) - .filter( - (session) => !session.capabilities.sessions.supportsRuntimeModeSwitchInSession, - ) - .map((session) => session.id) - : (providerSwitchPlan?.releaseProviderSessionIds ?? []), - ); if (detachSessionIds.size > 0) { const liveSessions = (providerContext?.providerSessions ?? []).filter( (session) => diff --git a/apps/web/src/components/BranchToolbarBranchSelector.tsx b/apps/web/src/components/BranchToolbarBranchSelector.tsx index b86517773207f..25b3c8f4221c3 100644 --- a/apps/web/src/components/BranchToolbarBranchSelector.tsx +++ b/apps/web/src/components/BranchToolbarBranchSelector.tsx @@ -115,7 +115,6 @@ export function BranchToolbarBranchSelector({ const composerFloatingLayerProps = useComposerMenuProps(); const canWriteSourceControl = useEnvironmentScope(environmentId, AuthSourceControlWriteScope); const canOperateThread = useEnvironmentScope(environmentId, AuthOrchestrationOperateScope); - const stopThreadSession = useAtomCommand(threadEnvironment.stopSession, "thread session stop"); const updateThreadMetadata = useAtomCommand( threadEnvironment.updateMetadata, "thread metadata update", @@ -176,12 +175,6 @@ export function BranchToolbarBranchSelector({ (hasServerThread && !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) ) return; - if (serverSession && worktreePath !== activeWorktreePath) { - void stopThreadSession({ - environmentId, - input: { threadId: activeThreadId }, - }); - } if (hasServerThread) { void (automatic ? updateEmptyThreadMetadata : updateThreadMetadata)({ environmentId, From c36d0a1f322695bacbb8f0ae670514c339e40b7e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 13:22:44 +0000 Subject: [PATCH 3/4] fix(server): sends and lone background agents no longer lose background work on account changes Exclusive sessions now ask the whole session for background work, since Claude's thread probe sees shells but not native background Agents. A send or queued send that would continue the native thread on another account refuses while that work runs, like the settings changes. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 55 ++++- .../SelectionRestart.integration.test.ts | 195 ++++++++++++------ 2 files changed, 177 insertions(+), 73 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 868865456c112..b3f3970282339 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1332,6 +1332,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : cause; const failure = makeProviderFailure({ cause: failureCause, + // Orchestrator refusals carry their user-facing reason as text. + ...(typeof failureCause === "string" ? { message: failureCause } : {}), code: handoffUnsupported ? "context_handoff_unsupported" : "queued_start_failed", class: handoffUnsupported ? "validation_error" : "unknown", }); @@ -1972,6 +1974,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio session.status !== "stopped" && session.status !== "error", ); + // Fails this run rather than holding the queue: nothing would restart it + // when the work ends, and the user can send again then (#16000). + if ( + sessionsToDetach.length > 0 && + (yield* hasLiveBackgroundWork( + threadId, + new Set(sessionsToDetach.map((session) => session.id)), + )) + ) { + return yield* backgroundWorkRefusal({ commandId, type: "message.dispatch" }); + } yield* writeSystemEvents( [ ...(selectionChanged @@ -2660,10 +2673,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .get(sessionId) .pipe(Effect.orElseSucceed(() => Option.none())); if (Option.isNone(session)) continue; - const { hasPendingBackgroundWork, hasPendingBackgroundWorkForThread } = session.value; - // A pooled session can serve other threads; only this thread's work counts. + const { hasPendingBackgroundWork, hasPendingBackgroundWorkForThread, providerSession } = + session.value; + // A pooled session can serve other threads, so only this thread's work + // counts. An exclusive session is all this thread's: ask the whole + // session, because Claude's thread probe sees shells but not its + // native background Agents. + const pooled = + providerSession.capabilities.sessions.supportsMultipleProviderThreadsPerSession; const probes = - hasPendingBackgroundWorkForThread === undefined + hasPendingBackgroundWorkForThread === undefined || + (!pooled && hasPendingBackgroundWork !== undefined) ? hasPendingBackgroundWork === undefined ? [] : [hasPendingBackgroundWork] @@ -2680,6 +2700,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return false; }); + const backgroundWorkRefusal = (command: { + readonly commandId: CommandId; + readonly type: string; + }) => + new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: + "Background work is still running in this thread, and this change would end it. Try again once it finishes, or stop the thread first.", + }); + const dispatchThreadMutation = Effect.fn("orchestrationV2.dispatch.threadMutation")(function* ( command: Extract< OrchestrationV2ServerCommand, @@ -3054,11 +3085,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio mapDispatchError(command), )) ) { - return yield* new OrchestratorDispatchError({ - commandId: command.commandId, - commandType: command.type, - cause: `Thread ${command.threadId} has background work running that this change would end. Try again once it finishes, or stop the thread first.`, - }); + return yield* backgroundWorkRefusal(command); } const now = yield* DateTime.now; @@ -5715,6 +5742,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio targetModelSelection: modelSelection, }) .pipe(mapDispatchError(command))).transition.type === "restart_and_resume"; + // Continuing the native thread on another account detaches the old + // session below, which would end its background work (#16000). + if ( + canResumeAcrossInstances && + activeProviderThread.providerSessionId !== null && + (yield* hasLiveBackgroundWork( + command.threadId, + new Set([activeProviderThread.providerSessionId]), + ).pipe(mapDispatchError(command))) + ) { + return yield* backgroundWorkRefusal(command); + } if ( pendingForkTransfer === undefined && diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index 3a76388050d94..cfc8a0e07f456 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -16,8 +16,10 @@ import { type RunId, ThreadId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; @@ -775,15 +777,29 @@ it.live.each(["active", "idle", "selection-command", "pooled", "separate-home"] ), ); -it.live.each(["runtime-mode", "provider-instance", "workspace"] as const)( - "keeps a session with running background work alive through a %s change", - (change) => +const backgroundWorkChanges = [ + "runtime-mode", + "provider-instance", + "workspace", + "send-to-other-account", + "queued-send-to-other-account", +] as const; + +it.live.each( + backgroundWorkChanges.flatMap((change) => [ + { change, backgroundWork: true }, + { change, backgroundWork: false }, + ]), +)( + "a $change change keeps a session alive only while it runs background work ($backgroundWork)", + ({ change, backgroundWork }) => Effect.scoped( Effect.gen(function* () { - const name = `background-work-${change}`; + const name = `background-work-${change}-${backgroundWork}`; const cwd = yield* checkpointWorkspace(name); const otherCwd = yield* checkpointWorkspace(`${name}-other`); const threadId = ThreadId.make(`thread:${name}`); + // Same home, so a send continues the native thread on the new account. const targetId = ProviderInstanceId.make("codex-background-target"); const state = yield* Ref.make({ activeTurn: null, @@ -792,6 +808,7 @@ it.live.each(["runtime-mode", "provider-instance", "workspace"] as const)( closedSessionCount: 0, failedReplacementOpen: true, }); + const sessions: Array = []; // Claude runs background agents and shells inside its CLI process and // cannot switch runtime mode in place. const capabilities: OrchestrationV2ProviderCapabilities = { @@ -807,12 +824,19 @@ it.live.each(["runtime-mode", "provider-instance", "workspace"] as const)( ...base, openSession: (input) => base.openSession(input).pipe( - Effect.map((session) => ({ - ...session, - backgroundTasksEndWithProcess: true as const, - hasPendingBackgroundWork: Effect.succeed(true), - hasPendingBackgroundWorkForThread: () => Effect.succeed(true), - })), + Effect.map((session) => { + const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = { + ...session, + backgroundTasksEndWithProcess: true, + // Claude's thread probe sees background shells but not + // native background Agents, so a lone Agent shows only + // session-wide. + hasPendingBackgroundWork: Effect.succeed(backgroundWork), + hasPendingBackgroundWorkForThread: () => Effect.succeed(false), + }; + sessions.push(runtime); + return runtime; + }), ), } satisfies ProviderAdapter.ProviderAdapterV2["Service"]; }); @@ -820,10 +844,10 @@ it.live.each(["runtime-mode", "provider-instance", "workspace"] as const)( get: (instanceId) => Effect.succeed(adapters.find((adapter) => adapter.instanceId === instanceId)!), list: () => Effect.succeed([providerInstanceId, targetId]), - getMetadata: (instanceId) => + getMetadata: () => Effect.succeed({ driver, - continuationKey: `codex:home:/${instanceId}`, + continuationKey: "codex:home:/shared", enabled: true, capabilities, }), @@ -831,8 +855,13 @@ it.live.each(["runtime-mode", "provider-instance", "workspace"] as const)( yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const sink = yield* EventSink.EventSinkV2; - const selection = { ...initialSelection, model: "complete" }; + const queued = change === "queued-send-to-other-account"; + // A queued send waits behind a running turn. + const selection = { + ...initialSelection, + model: queued ? initialSelection.model : "complete", + }; + const targetSelection = { instanceId: targetId, model: "complete" }; yield* orchestrator.dispatch({ type: "thread.create", commandId: CommandId.make(`${name}:create`), @@ -847,9 +876,11 @@ it.live.each(["runtime-mode", "provider-instance", "workspace"] as const)( createdBy: "user", creationSource: "web", }); - const completed = yield* orchestrator.streamDomainEvents.pipe( - Stream.filter( - (event) => event.type === "run.updated" && event.payload.status === "completed", + const started = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter((event) => + queued + ? event.type === "provider-turn.updated" && event.payload.status === "running" + : event.type === "run.updated" && event.payload.status === "completed", ), Stream.take(1), Stream.runDrain, @@ -867,71 +898,105 @@ it.live.each(["runtime-mode", "provider-instance", "workspace"] as const)( creationSource: "web", }); yield* worker.drain(); - yield* Fiber.join(completed); + yield* Fiber.join(started); yield* worker.drain(); const before = yield* orchestrator.getThreadProjection(threadId); const providerThread = before.providerThreads.find( (thread) => thread.id === before.thread.activeProviderThreadId, )!; - assert.deepEqual( - before.providerSessions + const sessionStatus = (projection: typeof before) => + projection.providerSessions .filter((session) => session.id === providerThread.providerSessionId) - .map((session) => session.status), - ["ready"], - ); - yield* sink.write({ - events: [ - { - id: EventId.make(`${name}:roster`), - type: "provider-thread.updated", - threadId, - occurredAt: yield* DateTime.now, - payload: { - ...providerThread, - pendingBackgroundTasks: [ - { taskId: "subagent", description: "Run 24 steps", kind: "subagent" }, - { taskId: "counter", description: "Append a counter", kind: "command" }, - ], - }, - }, - ], - }); + .map((session) => session.status); + assert.deepEqual(sessionStatus(before), ["ready"]); - // Refusing or deferring the change are both acceptable; ending the - // provider process while its background work runs is not. - yield* Effect.exit( - orchestrator.dispatch( - change === "runtime-mode" - ? { - type: "thread.runtime-mode.set", + const send = (dispatchMode: { + readonly type: "start_immediately" | "queue_after_active"; + }) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${name}:change`), + threadId, + messageId: MessageId.make(`${name}:change`), + text: "Continue on the other account", + attachments: [], + modelSelection: targetSelection, + dispatchMode, + createdBy: "user", + creationSource: "web", + }); + const result = yield* Effect.exit( + change === "runtime-mode" + ? orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make(`${name}:change`), + threadId, + runtimeMode: "auto-accept-edits", + }) + : change === "provider-instance" + ? orchestrator.dispatch({ + type: "provider.switch", commandId: CommandId.make(`${name}:change`), threadId, - runtimeMode: "auto-accept-edits", - } - : change === "provider-instance" - ? { - type: "provider.switch", - commandId: CommandId.make(`${name}:change`), - threadId, - modelSelection: { instanceId: targetId, model: "complete" }, - } - : { + modelSelection: targetSelection, + }) + : change === "workspace" + ? orchestrator.dispatch({ type: "thread.metadata.update", commandId: CommandId.make(`${name}:change`), threadId, worktreePath: otherCwd, - }, - ), + }) + : send({ type: queued ? "queue_after_active" : "start_immediately" }), ); yield* worker.drain(); + if (queued) { + assert.isTrue(Exit.isSuccess(result)); + // The first turn ends, and the queued send would start on the + // other account. + const queuedRunLeftQueue = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.userMessageId === `${name}:change` && + event.payload.status !== "queued", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* sessions[0]!.interruptTurn({ providerThread } as never); + yield* Fiber.join(queuedRunLeftQueue); + yield* worker.drain(); + } const after = yield* orchestrator.getThreadProjection(threadId); - assert.deepEqual( - after.providerSessions - .filter((session) => session.id === providerThread.providerSessionId) - .map((session) => session.status), - ["ready"], - ); + if (!backgroundWork) { + assert.isTrue(Exit.isSuccess(result)); + assert.notDeepEqual(sessionStatus(after), ["ready"]); + return; + } + const refusal = + "Background work is still running in this thread, and this change would end it. Try again once it finishes, or stop the thread first."; + if (queued) { + const failedRun = after.runs.find((run) => run.userMessageId === `${name}:change`); + assert.equal(failedRun?.status, "failed"); + assert.isTrue( + after.turnItems.some( + (item) => + item.runId === failedRun?.id && + item.type === "error" && + item.failure.message === refusal, + ), + ); + } else { + assert.isTrue(Exit.isFailure(result)); + if (Exit.isFailure(result)) { + const error = Cause.squash(result.cause); + assert.equal((error as Orchestrator.OrchestratorDispatchError).cause, refusal); + } + } + assert.deepEqual(sessionStatus(after), ["ready"]); assert.equal((yield* Ref.get(state)).closedSessionCount, 0); }).pipe(Effect.provide(ProviderReplayHarness.layerWithRegistry({ name }, layerRegistry))); }), From 4a3dab0fab00b242bd5b2b60827edc31b8acefac Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 13:36:11 +0000 Subject: [PATCH 4/4] fix(server): a send refused for background work no longer holds the queue A queued send refused because background work was still running was classed as an unknown provider failure, so the next queued message held the whole queue and nothing released it. The refusal is now a validation failure: only that send fails and the queue keeps going. Mobile settings command ids are now random per attempt, so a remount no longer replays ids the server already refused. Co-Authored-By: Claude Opus 5.5 --- .../src/state/use-thread-outbox-drain.ts | 23 +-- .../src/orchestration-v2/Orchestrator.ts | 12 +- .../SelectionRestart.integration.test.ts | 137 ++++++++++++++++++ 3 files changed, 152 insertions(+), 20 deletions(-) diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index 77880747dfb8e..4090f7b608782 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -184,13 +184,10 @@ function findCreationProject( } // A refused settings change stays refused under its command id, so each retry -// needs its own: the server can refuse one while background work runs. -function settingsCommandId( - message: QueuedThreadMessage, - setting: string, - attempt: number, -): CommandId { - return CommandId.make(`${message.commandId}:${setting}:${attempt}`); +// needs its own, even after a remount: the server can refuse one while +// background work runs. +function settingsCommandId(message: QueuedThreadMessage, setting: string): CommandId { + return CommandId.make(`${message.commandId}:${setting}:${randomHex(8)}`); } export async function prepareQueuedMessageAttachments( @@ -764,11 +761,7 @@ export function useThreadOutboxDrain(): void { const runtimeResult = await setThreadRuntimeMode({ environmentId: queuedMessage.environmentId, input: { - commandId: settingsCommandId( - queuedMessage, - "runtime-mode", - retryAttemptRef.current.get(queuedMessage.messageId) ?? 0, - ), + commandId: settingsCommandId(queuedMessage, "runtime-mode"), threadId: queuedMessage.threadId, runtimeMode: settings.runtimeMode, createdAt: queuedMessage.createdAt, @@ -785,11 +778,7 @@ export function useThreadOutboxDrain(): void { const interactionResult = await setThreadInteractionMode({ environmentId: queuedMessage.environmentId, input: { - commandId: settingsCommandId( - queuedMessage, - "interaction-mode", - retryAttemptRef.current.get(queuedMessage.messageId) ?? 0, - ), + commandId: settingsCommandId(queuedMessage, "interaction-mode"), threadId: queuedMessage.threadId, interactionMode: settings.interactionMode, createdAt: queuedMessage.createdAt, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b3f3970282339..d7067958960a1 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -917,6 +917,9 @@ function lastDeliveredRunForProviderThread( ); } +const BACKGROUND_WORK_REFUSAL = + "Background work is still running in this thread, and this change would end it. Try again once it finishes, or stop the thread first."; + const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(function* () { const checkpointService = yield* CheckpointServiceV2; const commandPolicy = yield* CommandPolicyV2; @@ -1335,7 +1338,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Orchestrator refusals carry their user-facing reason as text. ...(typeof failureCause === "string" ? { message: failureCause } : {}), code: handoffUnsupported ? "context_handoff_unsupported" : "queued_start_failed", - class: handoffUnsupported ? "validation_error" : "unknown", + // Background work refuses only this start; the queue behind it may run. + class: + handoffUnsupported || failureCause === BACKGROUND_WORK_REFUSAL + ? "validation_error" + : "unknown", }); yield* writeSystemEvents([ ...(attempt !== undefined && rootNode !== undefined @@ -2707,8 +2714,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, - cause: - "Background work is still running in this thread, and this change would end it. Try again once it finishes, or stop the thread first.", + cause: BACKGROUND_WORK_REFUSAL, }); const dispatchThreadMutation = Effect.fn("orchestrationV2.dispatch.threadMutation")(function* ( diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index cfc8a0e07f456..9c1fe2d6223ac 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -1002,3 +1002,140 @@ it.live.each( }), ), ); + +it.live("a refused queued send fails alone, and the queue behind it keeps running", () => + Effect.scoped( + Effect.gen(function* () { + const name = "background-work-queue"; + const cwd = yield* checkpointWorkspace(name); + const threadId = ThreadId.make(`thread:${name}`); + const targetId = ProviderInstanceId.make("codex-background-target"); + const state = yield* Ref.make({ + activeTurn: null, + opened: [], + started: [], + closedSessionCount: 0, + failedReplacementOpen: true, + }); + const sessions: Array = []; + const adapters = [providerInstanceId, targetId].map((instanceId) => { + const base = makeRestartAdapter(state, exclusiveCapabilities, instanceId); + return { + ...base, + openSession: (input) => + base.openSession(input).pipe( + Effect.map((session) => { + const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = { + ...session, + backgroundTasksEndWithProcess: true, + hasPendingBackgroundWork: Effect.succeed(true), + hasPendingBackgroundWorkForThread: () => Effect.succeed(false), + }; + sessions.push(runtime); + return runtime; + }), + ), + } satisfies ProviderAdapter.ProviderAdapterV2["Service"]; + }); + const layerRegistry = Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistryV2, { + get: (instanceId) => + Effect.succeed(adapters.find((adapter) => adapter.instanceId === instanceId)!), + list: () => Effect.succeed([providerInstanceId, targetId]), + getMetadata: () => + Effect.succeed({ + driver, + continuationKey: "codex:home:/shared", + enabled: true, + capabilities: exclusiveCapabilities, + }), + }); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${name}:create`), + threadId, + projectId: ProjectId.make(`project:${name}`), + title: name, + modelSelection: initialSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const send = ( + id: string, + dispatchMode: { readonly type: "start_immediately" | "queue_after_active" }, + modelSelection?: ModelSelection, + ) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${name}:${id}`), + threadId, + messageId: MessageId.make(`${name}:${id}`), + text: id, + attachments: [], + ...(modelSelection === undefined ? {} : { modelSelection }), + dispatchMode, + createdBy: "user", + creationSource: "web", + }); + // A held queue never starts another turn, so holding ends the wait too. + const runningTurnOrHeldQueue = () => + orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + (event.type === "provider-turn.updated" && event.payload.status === "running") || + (event.type === "run.updated" && event.payload.queueHeld === true), + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const firstRunning = yield* runningTurnOrHeldQueue(); + yield* send("first", { type: "start_immediately" }); + yield* worker.drain(); + yield* Fiber.join(firstRunning); + yield* worker.drain(); + const before = yield* orchestrator.getThreadProjection(threadId); + const providerThread = before.providerThreads.find( + (thread) => thread.id === before.thread.activeProviderThreadId, + )!; + + const targetSelection = { instanceId: targetId, model: initialSelection.model }; + yield* send("other-1", { type: "queue_after_active" }, targetSelection); + yield* send("other-2", { type: "queue_after_active" }, targetSelection); + yield* send("same", { type: "queue_after_active" }); + yield* worker.drain(); + + // Both sends to the other account are refused; the send that keeps + // the session starts on it. + const sameRunning = yield* runningTurnOrHeldQueue(); + yield* sessions[0]!.interruptTurn({ providerThread } as never); + yield* Fiber.join(sameRunning); + yield* worker.drain(); + + const after = yield* orchestrator.getThreadProjection(threadId); + const status = (id: string) => + after.runs.find((run) => run.userMessageId === `${name}:${id}`)?.status; + assert.deepEqual(["other-1", "other-2", "same"].map(status), [ + "failed", + "failed", + "running", + ]); + assert.isFalse(after.runs.some((run) => run.queueHeld === true)); + assert.deepEqual( + after.providerSessions + .filter((session) => session.id === providerThread.providerSessionId) + .map((session) => session.status), + ["ready"], + ); + assert.equal(sessions.length, 1); + assert.equal((yield* Ref.get(state)).closedSessionCount, 0); + }).pipe(Effect.provide(ProviderReplayHarness.layerWithRegistry({ name }, layerRegistry))); + }), + ), +);