diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index 47c76d72136a8..4090f7b608782 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -183,8 +183,11 @@ function findCreationProject( ); } +// A refused settings change stays refused under its command id, so each retry +// 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}`); + return CommandId.make(`${message.commandId}:${setting}:${randomHex(8)}`); } export async function prepareQueuedMessageAttachments( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 77533f593e9a7..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; @@ -1332,8 +1335,14 @@ 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", + // 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 @@ -1972,6 +1981,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 @@ -2650,6 +2670,53 @@ 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, 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 || + (!pooled && hasPendingBackgroundWork !== 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 backgroundWorkRefusal = (command: { + readonly commandId: CommandId; + readonly type: string; + }) => + new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: BACKGROUND_WORK_REFUSAL, + }); + const dispatchThreadMutation = Effect.fn("orchestrationV2.dispatch.threadMutation")(function* ( command: Extract< OrchestrationV2ServerCommand, @@ -2990,6 +3057,43 @@ 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* backgroundWorkRefusal(command); + } + const now = yield* DateTime.now; if (command.type === "thread.metadata.update" && command.limitRecovery != null) { const projection = yield* loadProjectionForCommand( @@ -3699,28 +3803,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) => @@ -5666,6 +5748,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 03cdf5e85a642..9c1fe2d6223ac 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"; @@ -774,3 +776,366 @@ it.live.each(["active", "idle", "selection-command", "pooled", "separate-home"] }), ), ); + +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}-${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, + opened: [], + started: [], + 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 = { + ...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) => { + 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"]; + }); + 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, + }), + }); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + 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`), + threadId, + projectId: ProjectId.make(`project:${name}`), + title: name, + modelSelection: selection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + 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, + 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(started); + yield* worker.drain(); + const before = yield* orchestrator.getThreadProjection(threadId); + const providerThread = before.providerThreads.find( + (thread) => thread.id === before.thread.activeProviderThreadId, + )!; + const sessionStatus = (projection: typeof before) => + projection.providerSessions + .filter((session) => session.id === providerThread.providerSessionId) + .map((session) => session.status); + assert.deepEqual(sessionStatus(before), ["ready"]); + + 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, + 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); + 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))); + }), + ), +); + +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))); + }), + ), +); 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,