From 067c2fd9a64d5dfe0438dbb1413f56333489eb06 Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Fri, 11 Sep 2026 17:23:37 +0000 Subject: [PATCH 1/8] Reconcile daemon turns the server lost track of on explicit stop A thread could wedge with every send refused as a competing turn while the app showed it idle or errored, and `bb thread stop` reported success without changing anything (#3462). Root cause, traced on four production Codex threads from 2026-09-01: the Codex bridge settled the spawn turn as a synthetic zero-work completion 250 ms after `turn/start` responded, before the real turn opened. The server went idle, a follow-up send was dispatched, and the real turn then started as an unrequested root turn. The daemon correctly refused the follow-up `turn.submit` as a competing turn, but the server settled that refusal as `run.failed`, moving the thread to `error` while a root turn it had already stored was still running. From `error` or `idle` an explicit stop sends `intent: "release"`, which the daemon deliberately skips when a turn is active, and the release result was indistinguishable from a completed release, so the stop was a silent no-op until the daemon turn ended. When it did, `run.succeeded` is illegal from `error`, so the thread stayed errored. The Codex zero-work settlement race that started the sequence is tracked separately in #2580 (candidate fix #2639); this change does not touch the bridge. It repairs the server and daemon sides so any provider turn the server lost track of can no longer wedge a thread or be reported as a successful stop. Changes: - Daemon: `thread.stop` with `intent: "release"` now reports `activeTurnRetained: true` when it keeps an active turn. Runtime and dispatcher refusals carry the `competing_turn` error code (`CompetingTurnError`, `COMPETING_TURN_ERROR_CODE`). - Server: explicit stops (stop route, context clear) that get a retained release re-activate the thread from the daemon's report, mark it stopping, and interrupt it through the normal stop path, so the stored turn is completed as interrupted and the thread settles idle. Automatic release paths and the daemon's release race guard are unchanged. - Server: a `turn.submit` refused with `competing_turn` while the thread has an uncompleted stored root turn records the rejection but keeps the thread active instead of failing the run; the running turn's completion settles it. - Protocol version 199 -> 200 for the new optional result field and error code. Verification: new server tests for idle/error retained-turn stops, context clear, concurrent stops, a failed escalated interrupt, and the competing-turn settlement with negative controls; all six behavioral tests fail on the previous lifecycle implementation. Daemon dispatch and error-code tests updated. Full @bb/server (2426), @bb/host-daemon (591), @bb/agent-runtime (318) and @bb/host-daemon-contract (51) suites pass. Live dev-app check with Codex: a running turn whose thread was flipped to error/idle is interrupted by `bb thread stop` and the next send succeeds; a send refused as competing keeps the thread active and it settles idle when the daemon turn completes. Co-Authored-By: Claude Fable 5.1 --- .../src/command-dispatch-support.test.ts | 20 ++ .../src/command-dispatch-support.ts | 29 ++- apps/host-daemon/src/command-dispatch.test.ts | 44 ++-- apps/host-daemon/src/command-dispatch.ts | 4 +- .../src/command-handlers/thread.ts | 5 +- .../src/services/threads/thread-lifecycle.ts | 81 ++++++- .../public/public-thread-stop-runtime.test.ts | 217 ++++++++++++++++++ .../test/threads/thread-send-dispatch.test.ts | 138 +++++++++++ packages/agent-runtime/src/index.ts | 6 +- packages/agent-runtime/src/runtime.ts | 13 +- packages/host-daemon-contract/src/commands.ts | 3 + packages/host-daemon-contract/src/protocol.ts | 2 +- .../test/contract.test.ts | 2 +- 13 files changed, 522 insertions(+), 42 deletions(-) diff --git a/apps/host-daemon/src/command-dispatch-support.test.ts b/apps/host-daemon/src/command-dispatch-support.test.ts index 95e785fee9..7a9f1741fe 100644 --- a/apps/host-daemon/src/command-dispatch-support.test.ts +++ b/apps/host-daemon/src/command-dispatch-support.test.ts @@ -1,10 +1,30 @@ +import { CompetingTurnError } from "@bb/agent-runtime"; +import { COMPETING_TURN_ERROR_CODE } from "@bb/host-daemon-contract"; import { describe, expect, it } from "vitest"; import { CommandDispatchError, + getErrorCode, isExpectedOnlineRpcFailureError, } from "./command-dispatch-support.js"; describe("command dispatch support", () => { + it("reports a runtime competing-turn refusal with the contract error code", () => { + expect(getErrorCode(new CompetingTurnError("thread-1"))).toBe( + COMPETING_TURN_ERROR_CODE, + ); + expect( + getErrorCode( + new CommandDispatchError( + COMPETING_TURN_ERROR_CODE, + "Refusing to start a competing turn while thread-1 is still starting", + ), + ), + ).toBe("competing_turn"); + expect(getErrorCode(new Error("Refusing to start a competing turn"))).toBe( + "command_failed", + ); + }); + it("classifies oversized file reads as expected RPC failures", () => { expect( isExpectedOnlineRpcFailureError( diff --git a/apps/host-daemon/src/command-dispatch-support.ts b/apps/host-daemon/src/command-dispatch-support.ts index e576a37163..0ead533c56 100644 --- a/apps/host-daemon/src/command-dispatch-support.ts +++ b/apps/host-daemon/src/command-dispatch-support.ts @@ -1,17 +1,21 @@ import type { DesktopBrowserBroker } from "./desktop-browser-broker.js"; -import type { AgentRuntimeBridgeLaunch } from "@bb/agent-runtime"; +import { + CompetingTurnError, + type AgentRuntimeBridgeLaunch, +} from "@bb/agent-runtime"; import type { AvailableModel } from "@bb/domain"; import type { EventSink } from "./event-sink.js"; -import type { - EnvironmentHookProgressMessage, - HostDaemonCommand, - ProviderHealthResult, - ProviderUsageResult, - HostDaemonBridgeLaunch, - HostDaemonInjectedSkillSource, - HostDaemonOnlineRpcCommand, - HostDaemonConnectTunnelIdentity, - WorkspaceContext, +import { + COMPETING_TURN_ERROR_CODE, + type EnvironmentHookProgressMessage, + type HostDaemonCommand, + type ProviderHealthResult, + type ProviderUsageResult, + type HostDaemonBridgeLaunch, + type HostDaemonInjectedSkillSource, + type HostDaemonOnlineRpcCommand, + type HostDaemonConnectTunnelIdentity, + type WorkspaceContext, } from "@bb/host-daemon-contract"; import type { ProviderInstallationCommand, @@ -186,6 +190,9 @@ export function getErrorCode(error: unknown): string { if (error instanceof CommandDispatchError) { return error.code; } + if (error instanceof CompetingTurnError) { + return COMPETING_TURN_ERROR_CODE; + } if (isStructuredSpawnMissingExecutableError(error)) { return "missing_executable"; } diff --git a/apps/host-daemon/src/command-dispatch.test.ts b/apps/host-daemon/src/command-dispatch.test.ts index 22d82e9b2b..50b6cf9770 100644 --- a/apps/host-daemon/src/command-dispatch.test.ts +++ b/apps/host-daemon/src/command-dispatch.test.ts @@ -1009,6 +1009,18 @@ describe("dispatchCommand", () => { workspacePath: "/tmp/bb-release-race", }); runtime.setActiveTurn("thread-1", "turn-new"); + const options = { + dataDir: "/tmp/bb-data", + logger: silentLogger, + eventSink: { emit: vi.fn(), flush: vi.fn(async () => undefined) }, + fetchProjectAttachment: async () => { + throw new Error("Unexpected project attachment fetch"); + }, + fetchPluginHostArtifact: fetchDispatchTestArtifact, + ...unexpectedProviderMaintenance, + runtimeManager: manager, + threadStorageRootPath: "/tmp/bb-thread-storage", + }; const result = await dispatchCommand( { @@ -1017,23 +1029,29 @@ describe("dispatchCommand", () => { environmentId: "env-release-race", threadId: "thread-1", }, - { - dataDir: "/tmp/bb-data", - logger: silentLogger, - eventSink: { emit: vi.fn(), flush: vi.fn(async () => undefined) }, - fetchProjectAttachment: async () => { - throw new Error("Unexpected project attachment fetch"); - }, - fetchPluginHostArtifact: fetchDispatchTestArtifact, - ...unexpectedProviderMaintenance, - runtimeManager: manager, - threadStorageRootPath: "/tmp/bb-thread-storage", - }, + options, ); expect(runtime.stopThread).not.toHaveBeenCalled(); expect(runtime.getActiveTurnId("thread-1")).toBe("turn-new"); - expect(result).toEqual({ providerCheckpointId: null }); + expect(result).toEqual({ + providerCheckpointId: null, + activeTurnRetained: true, + }); + + const interrupted = await dispatchCommand( + { + type: "thread.stop", + intent: "interrupt", + environmentId: "env-release-race", + threadId: "thread-1", + }, + options, + ); + + expect(runtime.stopThread).toHaveBeenCalledWith({ threadId: "thread-1" }); + expect(runtime.getActiveTurnId("thread-1")).toBeNull(); + expect(interrupted).toEqual({ providerCheckpointId: null }); }); it("treats thread.stop as successful when no runtime holds the thread", async () => { diff --git a/apps/host-daemon/src/command-dispatch.ts b/apps/host-daemon/src/command-dispatch.ts index 09ce9ec6f0..536979c500 100644 --- a/apps/host-daemon/src/command-dispatch.ts +++ b/apps/host-daemon/src/command-dispatch.ts @@ -98,7 +98,7 @@ type RuntimeStopCommand = async function stopThreadRuntime( command: RuntimeStopCommand, options: CommandDispatchOptions, -): Promise<{ providerCheckpointId: string | null }> { +): Promise> { const released = await options.runtimeManager.releaseThreadFromOtherEnvironments({ activeTurn: "interrupt", @@ -118,7 +118,7 @@ async function stopThreadRuntime( entry.runtime.getActiveTurnId(command.threadId) !== null ) { await options.eventSink.flush(); - return { providerCheckpointId }; + return { providerCheckpointId, activeTurnRetained: true }; } if (command.type !== "thread.stop" || command.intent !== "release") { await entry.runtime.waitForActiveTurn(command.threadId, { diff --git a/apps/host-daemon/src/command-handlers/thread.ts b/apps/host-daemon/src/command-handlers/thread.ts index 2b3728c0ff..7dda5f56b3 100644 --- a/apps/host-daemon/src/command-handlers/thread.ts +++ b/apps/host-daemon/src/command-handlers/thread.ts @@ -2,7 +2,7 @@ import fs from "node:fs/promises"; import path from "node:path"; import type { AgentRuntimeBridgeLaunch } from "@bb/agent-runtime"; import { flattenPromptInputGroups } from "@bb/domain"; -import type { HostDaemonCommandResult } from "@bb/host-daemon-contract"; +import { COMPETING_TURN_ERROR_CODE, type HostDaemonCommandResult } from "@bb/host-daemon-contract"; import type { RuntimeEntry } from "../runtime-manager.js"; import { CommandDispatchError, @@ -404,7 +404,8 @@ async function resolveLiveSubmittedTurnTarget( return refreshedTurnId; } if (entry.runtime.getLiveThreadIds().includes(command.threadId)) { - throw new Error( + throw new CommandDispatchError( + COMPETING_TURN_ERROR_CODE, `Refusing to start a competing turn while ${command.threadId} is still starting`, ); } diff --git a/apps/server/src/services/threads/thread-lifecycle.ts b/apps/server/src/services/threads/thread-lifecycle.ts index 48d36148a6..65923f5cfd 100644 --- a/apps/server/src/services/threads/thread-lifecycle.ts +++ b/apps/server/src/services/threads/thread-lifecycle.ts @@ -31,6 +31,7 @@ import { type DbTransaction, } from "@bb/db"; import { assertNever } from "@bb/core-ui"; +import { COMPETING_TURN_ERROR_CODE } from "@bb/host-daemon-contract"; import { type ProvisioningTranscriptEntry, type SystemThreadInterruptedReason, @@ -126,6 +127,10 @@ type TurnSubmitCommandResultReport = CommandResultReportForType<"turn.submit">; type ThreadStopCommandResultReport = CommandResultReportForType<"thread.stop">; type ThreadStorageDeleteCommandResultReport = CommandResultReportForType<"thread.storage.delete">; +type ThreadStopCommandResult = Extract< + ThreadStopCommandResultReport, + { ok: true } +>["result"]; type ThreadPlanCancelCommandResultReport = CommandResultReportForType<"thread.plan.cancel">; @@ -144,7 +149,10 @@ type PreparedReadyThreadTurnCommand = | PreparedReadyTurnSubmitCommand; const threadStartRequestDeduper = createAsyncDeduper(); -const threadStopRequestDeduper = createAsyncDeduper(); +const threadStopRequestDeduper = createAsyncDeduper< + string, + ThreadStopCommandResult | null +>(); type InFlightThreadRpcKind = | "thread.start" @@ -784,6 +792,12 @@ function settleThreadCommandFailure( message: args.report.errorMessage, }, }); + if ( + args.report.errorCode === COMPETING_TURN_ERROR_CODE && + getActiveTurnId(args.deps, thread.id) !== null + ) { + return emptyCommandResultSideEffects(); + } } if (hasExpectedTurnCompletedEvent(args.deps, args.command)) { return emptyCommandResultSideEffects(); @@ -1488,7 +1502,7 @@ export async function stopThreadForCurrentState( ) { return; } - await releaseIdleThreadRuntime(deps, thread.id, environment); + await releaseOrInterruptIdleThreadRuntime(deps, thread.id, environment); return; } @@ -1501,10 +1515,10 @@ export async function stopThreadForCurrentState( return; } - await releaseIdleThreadRuntime(deps, thread.id, environment); + await releaseOrInterruptIdleThreadRuntime(deps, thread.id, environment); } -async function releaseIdleThreadRuntime( +async function releaseOrInterruptIdleThreadRuntime( deps: RequestThreadStopForCurrentStateDeps, threadId: string, environment: RequestThreadStopForCurrentStateEnvironment | null, @@ -1512,7 +1526,7 @@ async function releaseIdleThreadRuntime( if (environment === null) { return; } - await runAwaitedThreadStopCommand(deps, { + const released = await runAwaitedThreadStopCommand(deps, { command: buildThreadStopCommand({ environmentId: environment.id, hostId: environment.hostId, @@ -1522,6 +1536,56 @@ async function releaseIdleThreadRuntime( hostId: environment.hostId, threadId, }); + if (released?.activeTurnRetained !== true) { + return; + } + deps.logger.warn( + { threadId }, + "Host daemon kept an active turn on release; interrupting it for the explicit stop", + ); + await interruptRetainedThreadRuntime(deps, threadId, environment); +} + +async function interruptRetainedThreadRuntime( + deps: RequestThreadStopForCurrentStateDeps, + threadId: string, + environment: RequestThreadStopForCurrentStateEnvironment, +): Promise { + const current = getThread(deps.db, threadId); + if (!current) { + return; + } + const args: RequestThreadStopArgs = { + environmentId: environment.id, + hostId: environment.hostId, + interruptionReason: "manual-stop", + threadId, + }; + if (current.status === "idle" || current.status === "error") { + const revived = applyLoggedThreadLifecycleEvent(deps, { + event: { type: "run.started" }, + threadId, + }); + if (!revived.applied) { + await runAwaitedThreadStopCommand(deps, { + command: buildThreadStopCommand({ ...args, intent: "interrupt" }), + hostId: args.hostId, + threadId, + }); + return; + } + } + if (getThread(deps.db, threadId)?.status !== "active") { + return; + } + if (!markThreadStopRequested(deps, args)) { + return; + } + await runAwaitedThreadStopCommand(deps, { + command: buildThreadStopCommand({ ...args, intent: "interrupt" }), + hostId: args.hostId, + threadId, + }); } async function runAwaitedThreadStopCommand( @@ -1532,11 +1596,11 @@ async function runAwaitedThreadStopCommand( hostId: string; threadId: string; }, -): Promise { - await threadStopRequestDeduper.run(args.threadId, async () => { +): Promise { + return threadStopRequestDeduper.run(args.threadId, async () => { inFlightThreadRpcGuard.claim(args.threadId, "thread.stop"); try { - await runLiveHostCommand(deps, { + return await runLiveHostCommand(deps, { command: args.command, hostId: args.hostId, timeoutMs: AWAITED_THREAD_STOP_TIMEOUT_MS, @@ -1553,6 +1617,7 @@ async function runAwaitedThreadStopCommand( ) { throw error; } + return null; } finally { inFlightThreadRpcGuard.release(args.threadId, "thread.stop"); } diff --git a/apps/server/test/public/public-thread-stop-runtime.test.ts b/apps/server/test/public/public-thread-stop-runtime.test.ts index 347c4dbd5e..8df302900e 100644 --- a/apps/server/test/public/public-thread-stop-runtime.test.ts +++ b/apps/server/test/public/public-thread-stop-runtime.test.ts @@ -23,6 +23,7 @@ import { seedStoredEvent, seedThread, seedThreadFixture, + seedTurnStarted, } from "../helpers/seed.js"; import { withTestHarness } from "../helpers/test-app.js"; import { runQueuedMessageDispatch } from "../../src/services/threads/queued-message-dispatch.js"; @@ -61,6 +62,222 @@ describe("thread runtime stop", () => { }); }); + for (const status of ["idle", "error"] as const) { + it(`interrupts a turn the daemon kept when the server believed the thread was ${status}`, async () => { + await withTestHarness(async (harness) => { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status, visibility: "hidden" }, + }); + seedTurnStarted(harness.deps, { + environmentId: environment.id, + providerThreadId: "provider-retained", + threadId: thread.id, + turnId: "turn-retained", + }); + + const responsePromise = harness.app.request( + `/api/v1/threads/${thread.id}/stop`, + { method: "POST" }, + ); + const release = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "release", + ); + await reportQueuedCommandSuccess(harness, release, { + providerCheckpointId: null, + activeTurnRetained: true, + }); + + const interrupt = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "interrupt", + ); + expect(getThread(harness.db, thread.id)?.status).toBe("stopping"); + await reportQueuedCommandSuccess(harness, interrupt, { + providerCheckpointId: null, + }); + + const response = await responsePromise; + expect(response.status).toBe(200); + expect(getThread(harness.db, thread.id)?.status).toBe("idle"); + const events = listEvents(harness.db, { threadId: thread.id }); + expect( + events.filter((event) => event.type === "system/thread/interrupted"), + ).toHaveLength(1); + const completion = events.find( + (event) => + event.type === "turn/completed" && event.turnId === "turn-retained", + ); + expect(completion).toBeDefined(); + expect(JSON.parse(completion?.data ?? "{}")).toMatchObject({ + status: "interrupted", + }); + }); + }); + } + + it("clears context after interrupting a turn the daemon kept", async () => { + await withTestHarness(async (harness) => { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "idle", visibility: "hidden" }, + }); + seedTurnStarted(harness.deps, { + environmentId: environment.id, + providerThreadId: "provider-retained", + threadId: thread.id, + turnId: "turn-retained", + }); + + const responsePromise = harness.app.request( + `/api/v1/threads/${thread.id}/context/clear`, + { method: "POST" }, + ); + const release = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "release", + ); + await reportQueuedCommandSuccess(harness, release, { + providerCheckpointId: null, + activeTurnRetained: true, + }); + const interrupt = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "interrupt", + ); + await reportQueuedCommandSuccess(harness, interrupt, { + providerCheckpointId: null, + }); + + const response = await responsePromise; + expect(response.status, await response.clone().text()).toBe(200); + expect(getThread(harness.db, thread.id)?.status).toBe("idle"); + expect( + listEvents(harness.db, { threadId: thread.id }).filter( + (event) => + event.type === "system/operation" && + JSON.parse(event.data).operation === "context_clear", + ), + ).toHaveLength(1); + }); + }); + + it("shares one release and one interrupt across concurrent stops of a kept turn", async () => { + await withTestHarness(async (harness) => { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "idle", visibility: "hidden" }, + }); + seedTurnStarted(harness.deps, { + environmentId: environment.id, + providerThreadId: "provider-retained", + threadId: thread.id, + turnId: "turn-retained", + }); + + const first = harness.app.request(`/api/v1/threads/${thread.id}/stop`, { + method: "POST", + }); + const release = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "release", + ); + const second = harness.app.request( + `/api/v1/threads/${thread.id}/stop`, + { method: "POST" }, + ); + expect(listQueuedCommands(harness, "thread.stop")).toHaveLength(1); + await reportQueuedCommandSuccess(harness, release, { + providerCheckpointId: null, + activeTurnRetained: true, + }); + + const interrupt = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "interrupt", + ); + expect(listQueuedCommands(harness, "thread.stop")).toHaveLength(1); + await reportQueuedCommandSuccess(harness, interrupt, { + providerCheckpointId: null, + }); + + expect((await first).status).toBe(200); + expect((await second).status).toBe(200); + expect(getThread(harness.db, thread.id)?.status).toBe("idle"); + expect( + listEvents(harness.db, { threadId: thread.id }).filter( + (event) => event.type === "system/thread/interrupted", + ), + ).toHaveLength(1); + }); + }); + + it("leaves a kept turn stopping when the escalated interrupt fails", async () => { + await withTestHarness(async (harness) => { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "idle", visibility: "hidden" }, + }); + seedTurnStarted(harness.deps, { + environmentId: environment.id, + providerThreadId: "provider-retained", + threadId: thread.id, + turnId: "turn-retained", + }); + + const responsePromise = harness.app.request( + `/api/v1/threads/${thread.id}/stop`, + { method: "POST" }, + ); + const release = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "release", + ); + await reportQueuedCommandSuccess(harness, release, { + providerCheckpointId: null, + activeTurnRetained: true, + }); + const interrupt = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "interrupt", + ); + await reportQueuedCommandError(harness, interrupt, { + errorCode: "test_interrupt_failure", + errorMessage: "Test interrupt failure", + }); + + expect((await responsePromise).status).toBe(200); + expect(getThread(harness.db, thread.id)?.status).toBe("stopping"); + expect( + listEvents(harness.db, { threadId: thread.id }).find( + (event) => + event.type === "turn/completed" && event.turnId === "turn-retained", + ), + ).toBeUndefined(); + }); + }); + it("settles background commands terminated by an idle runtime release", async () => { await withTestHarness(async (harness) => { const { environment, thread } = seedThreadFixture(harness, { diff --git a/apps/server/test/threads/thread-send-dispatch.test.ts b/apps/server/test/threads/thread-send-dispatch.test.ts index 9059eaed91..c083e1a7e7 100644 --- a/apps/server/test/threads/thread-send-dispatch.test.ts +++ b/apps/server/test/threads/thread-send-dispatch.test.ts @@ -1453,3 +1453,141 @@ describe("concurrent idle dispatch regression", () => { }); }); }); + +describe("competing turn refusals", () => { + const competingTurnMessage = (threadId: string) => + `Refusing to start a competing turn for thread "${threadId}" while another turn is active or starting`; + + async function sendStartFromIdle( + harness: TestAppHarness, + fixture: IdleThreadFixture, + ) { + await sendThreadMessage(harness.deps, { + environment: fixture.environment, + payload: { + input: textInput("send while the daemon runs a turn"), + mode: "start", + model: "gpt-5", + permissionMode: "full", + reasoningLevel: "medium", + serviceTier: "default", + }, + thread: fixture.thread, + trigger: "user", + }); + expect(getThread(harness.db, fixture.thread.id)?.status).toBe("active"); + const queued = await waitForQueuedCommand( + harness, + (candidate) => + candidate.command.type === "turn.submit" && + candidate.command.threadId === fixture.thread.id, + ); + if (queued.command.type !== "turn.submit") { + throw new Error("Expected a turn.submit command"); + } + return queued; + } + + it("keeps the thread active when the daemon refuses a competing turn while a root turn is running", async () => { + await withTestHarness(async (harness) => { + const fixture = seedProviderThreadFixture({ + harness, + status: "idle", + value: 61, + }); + const queued = await sendStartFromIdle(harness, fixture); + seedTurnStarted(harness.deps, { + environmentId: fixture.environment.id, + providerThreadId: "provider-send-dispatch-61", + threadId: fixture.thread.id, + turnId: "turn-unrequested", + }); + + await reportQueuedCommandError(harness, queued, { + errorCode: "competing_turn", + errorMessage: competingTurnMessage(fixture.thread.id), + }); + + const events = listEvents(harness.db, { threadId: fixture.thread.id }); + const rejection = events.find( + (event) => event.type === "client/turn/rejected", + ); + expect(JSON.parse(rejection?.data ?? "{}")).toEqual({ + requestId: queued.command.requestId, + reason: "competing_turn", + message: competingTurnMessage(fixture.thread.id), + }); + expect(events.some((event) => event.type === "system/error")).toBe(false); + expect(getThread(harness.db, fixture.thread.id)?.status).toBe("active"); + + const completed = await harness.app.request("/internal/session/events", { + method: "POST", + headers: internalAuthHeaders(harness), + body: JSON.stringify({ + sessionId: fixture.sessionId, + eventGroups: groupHostDaemonEvents([ + { + threadId: fixture.thread.id, + event: { + type: "turn/completed", + threadId: fixture.thread.id, + providerThreadId: "provider-send-dispatch-61", + scope: turnScope("turn-unrequested"), + status: "completed", + }, + }, + ]), + }), + }); + expect(completed.status, await completed.clone().text()).toBe(200); + expect(getThread(harness.db, fixture.thread.id)?.status).toBe("idle"); + }); + }); + + it("fails the run when a competing-turn refusal arrives without a running root turn", async () => { + await withTestHarness(async (harness) => { + const fixture = seedProviderThreadFixture({ + harness, + status: "idle", + value: 62, + }); + const queued = await sendStartFromIdle(harness, fixture); + + await reportQueuedCommandError(harness, queued, { + errorCode: "competing_turn", + errorMessage: competingTurnMessage(fixture.thread.id), + }); + + const events = listEvents(harness.db, { threadId: fixture.thread.id }); + expect(events.some((event) => event.type === "client/turn/rejected")).toBe( + true, + ); + expect(events.some((event) => event.type === "system/error")).toBe(true); + expect(getThread(harness.db, fixture.thread.id)?.status).toBe("error"); + }); + }); + + it("still fails the run for other refusals while a root turn is stored", async () => { + await withTestHarness(async (harness) => { + const fixture = seedProviderThreadFixture({ + harness, + status: "idle", + value: 63, + }); + const queued = await sendStartFromIdle(harness, fixture); + seedTurnStarted(harness.deps, { + environmentId: fixture.environment.id, + providerThreadId: "provider-send-dispatch-63", + threadId: fixture.thread.id, + turnId: "turn-unrequested", + }); + + await reportQueuedCommandError(harness, queued, { + errorCode: "provider_rpc_error", + errorMessage: "Provider rejected the turn", + }); + + expect(getThread(harness.db, fixture.thread.id)?.status).toBe("error"); + }); + }); +}); diff --git a/packages/agent-runtime/src/index.ts b/packages/agent-runtime/src/index.ts index 70061e13f7..9f932abd79 100644 --- a/packages/agent-runtime/src/index.ts +++ b/packages/agent-runtime/src/index.ts @@ -1,4 +1,8 @@ -export { AgentRuntimeRecoveryError, createAgentRuntime } from "./runtime.js"; +export { + AgentRuntimeRecoveryError, + CompetingTurnError, + createAgentRuntime, +} from "./runtime.js"; export { bridgeLaunchProcessKey } from "./bridge-launch-process-key.js"; export type { AgentRuntime, diff --git a/packages/agent-runtime/src/runtime.ts b/packages/agent-runtime/src/runtime.ts index f65aa5f63f..eee218c351 100644 --- a/packages/agent-runtime/src/runtime.ts +++ b/packages/agent-runtime/src/runtime.ts @@ -138,6 +138,15 @@ interface RequestRecoveryArgs { threadId: string; } +export class CompetingTurnError extends Error { + constructor(threadId: string) { + super( + `Refusing to start a competing turn for thread "${threadId}" while another turn is active or starting`, + ); + this.name = "CompetingTurnError"; + } +} + export class AgentRuntimeRecoveryError extends Error { readonly code: "auth_required" | "rate_limited"; readonly recovery: AgentRuntimeProviderRecoveryHint; @@ -765,9 +774,7 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { turnState.getActiveTurnId(threadId) !== null || pendingTurnStarts.has(threadId) ) { - throw new Error( - `Refusing to start a competing turn for thread "${threadId}" while another turn is active or starting`, - ); + throw new CompetingTurnError(threadId); } } diff --git a/packages/host-daemon-contract/src/commands.ts b/packages/host-daemon-contract/src/commands.ts index 1626577a64..6102aabbb6 100644 --- a/packages/host-daemon-contract/src/commands.ts +++ b/packages/host-daemon-contract/src/commands.ts @@ -1216,9 +1216,12 @@ const threadStartResultSchema = z.object({ const turnSubmitResultSchema = z.object({ appliedAs: z.enum(["new-turn", "steer"]), }); +export const COMPETING_TURN_ERROR_CODE = "competing_turn" as const; + const threadStopResultSchema = z .object({ providerCheckpointId: z.string().min(1).nullable(), + activeTurnRetained: z.boolean().optional(), }) .strict(); const emptyCommandResultSchema = z.object({}); diff --git a/packages/host-daemon-contract/src/protocol.ts b/packages/host-daemon-contract/src/protocol.ts index 63f7f40e86..acc5e5d3a3 100644 --- a/packages/host-daemon-contract/src/protocol.ts +++ b/packages/host-daemon-contract/src/protocol.ts @@ -1,3 +1,3 @@ -export const HOST_DAEMON_PROTOCOL_VERSION = 211 as const; +export const HOST_DAEMON_PROTOCOL_VERSION = 212 as const; export const HOST_ARTIFACT_MAX_BYTES = 256 * 1024 * 1024; diff --git a/packages/host-daemon-contract/test/contract.test.ts b/packages/host-daemon-contract/test/contract.test.ts index 02f023c293..971f7f31d5 100644 --- a/packages/host-daemon-contract/test/contract.test.ts +++ b/packages/host-daemon-contract/test/contract.test.ts @@ -1066,7 +1066,7 @@ const CONTRIBUTED_ENV = [ describe("host-daemon command schemas", () => { it("uses the current host-daemon protocol version", () => { - expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(211); + expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(212); expect(HOST_ARTIFACT_MAX_BYTES).toBe(256 * 1024 * 1024); }); From 2e5eafd30c408369b69fc4814fcac380381867cf Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Fri, 11 Sep 2026 17:33:32 +0000 Subject: [PATCH 2/8] Narrow the turn.submit request id in the competing-turn tests The test helper returned the queued command as the RPC union, so the caller's requestId read failed typecheck in CI. Co-Authored-By: Claude Fable 5.1 --- apps/server/test/threads/thread-send-dispatch.test.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/apps/server/test/threads/thread-send-dispatch.test.ts b/apps/server/test/threads/thread-send-dispatch.test.ts index c083e1a7e7..2a543446c6 100644 --- a/apps/server/test/threads/thread-send-dispatch.test.ts +++ b/apps/server/test/threads/thread-send-dispatch.test.ts @@ -1485,7 +1485,7 @@ describe("competing turn refusals", () => { if (queued.command.type !== "turn.submit") { throw new Error("Expected a turn.submit command"); } - return queued; + return { queued, requestId: queued.command.requestId }; } it("keeps the thread active when the daemon refuses a competing turn while a root turn is running", async () => { @@ -1495,7 +1495,7 @@ describe("competing turn refusals", () => { status: "idle", value: 61, }); - const queued = await sendStartFromIdle(harness, fixture); + const { queued, requestId } = await sendStartFromIdle(harness, fixture); seedTurnStarted(harness.deps, { environmentId: fixture.environment.id, providerThreadId: "provider-send-dispatch-61", @@ -1513,7 +1513,7 @@ describe("competing turn refusals", () => { (event) => event.type === "client/turn/rejected", ); expect(JSON.parse(rejection?.data ?? "{}")).toEqual({ - requestId: queued.command.requestId, + requestId, reason: "competing_turn", message: competingTurnMessage(fixture.thread.id), }); @@ -1551,7 +1551,7 @@ describe("competing turn refusals", () => { status: "idle", value: 62, }); - const queued = await sendStartFromIdle(harness, fixture); + const { queued } = await sendStartFromIdle(harness, fixture); await reportQueuedCommandError(harness, queued, { errorCode: "competing_turn", @@ -1574,7 +1574,7 @@ describe("competing turn refusals", () => { status: "idle", value: 63, }); - const queued = await sendStartFromIdle(harness, fixture); + const { queued } = await sendStartFromIdle(harness, fixture); seedTurnStarted(harness.deps, { environmentId: fixture.environment.id, providerThreadId: "provider-send-dispatch-63", From 7a381bb3c70f2eb88189461470b8177a3ec6aa3e Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Fri, 11 Sep 2026 17:34:34 +0000 Subject: [PATCH 3/8] Format the competing-turn and stop-runtime tests Co-Authored-By: Claude Fable 5.1 --- apps/server/test/public/public-thread-stop-runtime.test.ts | 7 +++---- apps/server/test/threads/thread-send-dispatch.test.ts | 6 +++--- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/apps/server/test/public/public-thread-stop-runtime.test.ts b/apps/server/test/public/public-thread-stop-runtime.test.ts index 8df302900e..cc1a7757b0 100644 --- a/apps/server/test/public/public-thread-stop-runtime.test.ts +++ b/apps/server/test/public/public-thread-stop-runtime.test.ts @@ -195,10 +195,9 @@ describe("thread runtime stop", () => { command.threadId === thread.id && command.intent === "release", ); - const second = harness.app.request( - `/api/v1/threads/${thread.id}/stop`, - { method: "POST" }, - ); + const second = harness.app.request(`/api/v1/threads/${thread.id}/stop`, { + method: "POST", + }); expect(listQueuedCommands(harness, "thread.stop")).toHaveLength(1); await reportQueuedCommandSuccess(harness, release, { providerCheckpointId: null, diff --git a/apps/server/test/threads/thread-send-dispatch.test.ts b/apps/server/test/threads/thread-send-dispatch.test.ts index 2a543446c6..f3a8d0ed1b 100644 --- a/apps/server/test/threads/thread-send-dispatch.test.ts +++ b/apps/server/test/threads/thread-send-dispatch.test.ts @@ -1559,9 +1559,9 @@ describe("competing turn refusals", () => { }); const events = listEvents(harness.db, { threadId: fixture.thread.id }); - expect(events.some((event) => event.type === "client/turn/rejected")).toBe( - true, - ); + expect( + events.some((event) => event.type === "client/turn/rejected"), + ).toBe(true); expect(events.some((event) => event.type === "system/error")).toBe(true); expect(getThread(harness.db, fixture.thread.id)?.status).toBe("error"); }); From 764b6cc02232780fa7c43b9d3b8d1f285a58ac2c Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Fri, 11 Sep 2026 21:28:27 +0000 Subject: [PATCH 4/8] Make an explicit stop wait for every turn it finds An explicit stop (the stop route, context clear, and machine preservation) is now one single-flight operation per thread that re-reads the thread row each round: it interrupts a live turn, releases an idle runtime, and after a release the daemon declined it re-activates the thread and interrupts the retained turn. A concurrent stop joins the whole operation, so it no longer returns when a pending release resolves while the escalated interrupt is still outstanding. Command failures are returned to each caller, and callers that require the thread stopped get the failure of the escalated interrupt. An explicit stop wins over work that is still running: a turn that starts while the stop's release is in flight is interrupted too. This is documented in the thread guide and the SDK declaration and covered by a regression. The stop guard now counts holders and tracks releases separately, so a pending awaited release no longer suppresses an interrupt requested through the non-awaited path, and concurrent awaited stops no longer drop each other's claim. Co-Authored-By: Claude Code --- apps/host-daemon/src/command-dispatch.test.ts | 11 +- .../src/services/threads/thread-lifecycle.ts | 284 ++++++++++-------- .../public/public-thread-stop-runtime.test.ts | 262 +++++++++++++++- packages/sdk/src/areas/threads.ts | 6 + .../src/templates/bb-guide-threads.md | 5 +- 5 files changed, 432 insertions(+), 136 deletions(-) diff --git a/apps/host-daemon/src/command-dispatch.test.ts b/apps/host-daemon/src/command-dispatch.test.ts index 50b6cf9770..e775cc2063 100644 --- a/apps/host-daemon/src/command-dispatch.test.ts +++ b/apps/host-daemon/src/command-dispatch.test.ts @@ -821,8 +821,7 @@ describe("dispatchCommand", () => { .mockReturnValueOnce(newRuntime); const manager = new RuntimeManager({ createRuntime: createRuntimeSpy, - provisionWorkspace: async (args) => - createWorkspace(args.path), + provisionWorkspace: async (args) => createWorkspace(args.path), }); await manager.ensureEnvironment({ environmentId: "env-old", @@ -998,7 +997,7 @@ describe("dispatchCommand", () => { }); }); - it("skips a release when a turn started after the server read the thread", async () => { + it("reports a retained turn instead of releasing a turn that started after the server read the thread", async () => { const runtime = createRuntime(); const manager = new RuntimeManager({ createRuntime: () => runtime, @@ -1165,8 +1164,7 @@ describe("dispatchCommand", () => { .mockReturnValueOnce(newRuntime); const manager = new RuntimeManager({ createRuntime: createRuntimeSpy, - provisionWorkspace: async (args) => - createWorkspace(args.path), + provisionWorkspace: async (args) => createWorkspace(args.path), }); await manager.ensureEnvironment({ environmentId: "env-old", @@ -1216,8 +1214,7 @@ describe("dispatchCommand", () => { .mockReturnValueOnce(newRuntime); const manager = new RuntimeManager({ createRuntime: createRuntimeSpy, - provisionWorkspace: async (args) => - createWorkspace(args.path), + provisionWorkspace: async (args) => createWorkspace(args.path), }); await manager.ensureEnvironment({ environmentId: "env-old", diff --git a/apps/server/src/services/threads/thread-lifecycle.ts b/apps/server/src/services/threads/thread-lifecycle.ts index 65923f5cfd..75a5cbd5c5 100644 --- a/apps/server/src/services/threads/thread-lifecycle.ts +++ b/apps/server/src/services/threads/thread-lifecycle.ts @@ -86,6 +86,7 @@ import { type ThreadStopCommandArgs, } from "./thread-commands.js"; import { ensureHostSessionReadyForWork } from "../hosts/host-lifecycle.js"; +import { ApiError } from "../../errors.js"; import { isHostUnavailableApiError } from "../hosts/online-rpc.js"; import { LIVE_DAEMON_COMMAND_TIMEOUT_MS, @@ -149,19 +150,32 @@ type PreparedReadyThreadTurnCommand = | PreparedReadyTurnSubmitCommand; const threadStartRequestDeduper = createAsyncDeduper(); +interface ThreadStopFailure { + error: unknown; + intent: ThreadStopCommand["intent"]; +} + +interface AwaitedThreadStopOutcome { + failure: ThreadStopFailure | null; + result: ThreadStopCommandResult | null; +} + const threadStopRequestDeduper = createAsyncDeduper< string, - ThreadStopCommandResult | null + ThreadStopFailure | null >(); +const MAX_EXPLICIT_THREAD_STOP_ROUNDS = 4; + type InFlightThreadRpcKind = | "thread.start" | "thread.start.title-sync" | "thread.stop" - | "thread.storage.delete"; + | "thread.storage.delete" + | "thread.stop.release"; class InFlightRpcGuard { - private readonly held = new Set(); + private readonly holders = new Map(); private key(threadId: string, kind: InFlightThreadRpcKind): string { return `${kind}:${threadId}`; @@ -169,19 +183,30 @@ class InFlightRpcGuard { claim(threadId: string, kind: InFlightThreadRpcKind): boolean { const key = this.key(threadId, kind); - if (this.held.has(key)) { + if (this.holders.has(key)) { return false; } - this.held.add(key); + this.holders.set(key, 1); return true; } + share(threadId: string, kind: InFlightThreadRpcKind): void { + const key = this.key(threadId, kind); + this.holders.set(key, (this.holders.get(key) ?? 0) + 1); + } + release(threadId: string, kind: InFlightThreadRpcKind): void { - this.held.delete(this.key(threadId, kind)); + const key = this.key(threadId, kind); + const count = this.holders.get(key); + if (count === undefined || count <= 1) { + this.holders.delete(key); + return; + } + this.holders.set(key, count - 1); } isHeld(threadId: string, kind: InFlightThreadRpcKind): boolean { - return this.held.has(this.key(threadId, kind)); + return this.holders.has(this.key(threadId, kind)); } } @@ -192,7 +217,10 @@ export function hasLiveThreadStartInFlight(threadId: string): boolean { } export function hasLiveThreadStopInFlight(threadId: string): boolean { - return inFlightThreadRpcGuard.isHeld(threadId, "thread.stop"); + return ( + inFlightThreadRpcGuard.isHeld(threadId, "thread.stop") || + inFlightThreadRpcGuard.isHeld(threadId, "thread.stop.release") + ); } interface ThreadStartSuccessActivationArgs { @@ -1476,152 +1504,154 @@ export async function stopThreadForCurrentState( options?: { requireStopped: true }, ): Promise { await revokeThreadDesktopBrowserControl(deps, thread.id); - if (hasLiveThreadRuntime(deps, thread)) { - if (environment === null) { - return; - } - const args: RequestThreadStopArgs = { - environmentId: environment.id, - hostId: environment.hostId, - interruptionReason: "manual-stop", - threadId: thread.id, - }; - if (markThreadStopRequested(deps, args)) { - await runAwaitedThreadStopCommand(deps, { - requireStopped: options?.requireStopped, - command: buildThreadStopCommand({ ...args, intent: "interrupt" }), - hostId: args.hostId, - threadId: thread.id, - }); - return; - } - const settledThread = getThread(deps.db, thread.id); - if ( - settledThread === null || - (settledThread.status !== "idle" && settledThread.status !== "error") - ) { - return; - } - await releaseOrInterruptIdleThreadRuntime(deps, thread.id, environment); + const failure = await threadStopRequestDeduper.run(thread.id, () => + stopThreadUntilSettled(deps, thread.id, environment), + ); + if (failure === null) { return; } - + if (options?.requireStopped) { + throw failure.error; + } if ( - isPreStartThreadStatus(thread.status) || - thread.status === "stopping" || - hasActiveThreadProvisioningContext(deps, thread.id) + failure.intent === "release" && + !isHostUnavailableApiError(failure.error) ) { - requestPreStartThreadStop(deps, thread); - return; + throw failure.error; } +} - await releaseOrInterruptIdleThreadRuntime(deps, thread.id, environment); +function manualThreadStopArgs( + threadId: string, + environment: RequestThreadStopForCurrentStateEnvironment, +): RequestThreadStopArgs { + return { + environmentId: environment.id, + hostId: environment.hostId, + interruptionReason: "manual-stop", + threadId, + }; } -async function releaseOrInterruptIdleThreadRuntime( +async function stopThreadUntilSettled( deps: RequestThreadStopForCurrentStateDeps, threadId: string, environment: RequestThreadStopForCurrentStateEnvironment | null, -): Promise { - if (environment === null) { - return; - } - const released = await runAwaitedThreadStopCommand(deps, { - command: buildThreadStopCommand({ - environmentId: environment.id, - hostId: environment.hostId, - intent: "release", - threadId, - }), - hostId: environment.hostId, - threadId, - }); - if (released?.activeTurnRetained !== true) { - return; +): Promise { + for (let round = 0; round < MAX_EXPLICIT_THREAD_STOP_ROUNDS; round += 1) { + const current = getThread(deps.db, threadId); + if (current === null) { + return null; + } + if (hasLiveThreadRuntime(deps, current)) { + if (environment === null) { + return null; + } + const args = manualThreadStopArgs(threadId, environment); + if (markThreadStopRequested(deps, args)) { + const interrupted = await runAwaitedThreadStopCommand(deps, { + command: buildThreadStopCommand({ ...args, intent: "interrupt" }), + hostId: args.hostId, + }); + return interrupted.failure; + } + const settled = getThread(deps.db, threadId); + if ( + settled === null || + (settled.status !== "idle" && settled.status !== "error") + ) { + return null; + } + } else if ( + isPreStartThreadStatus(current.status) || + current.status === "stopping" || + hasActiveThreadProvisioningContext(deps, threadId) + ) { + requestPreStartThreadStop(deps, current); + return null; + } + if (environment === null) { + return null; + } + const args = manualThreadStopArgs(threadId, environment); + const released = await runAwaitedThreadStopCommand(deps, { + command: buildThreadStopCommand({ ...args, intent: "release" }), + hostId: args.hostId, + }); + if (released.failure !== null) { + return released.failure; + } + if (released.result?.activeTurnRetained === true) { + deps.logger.warn( + { threadId }, + "Host daemon kept an active turn on release; interrupting it for the explicit stop", + ); + if (!reviveThreadFromRetainedTurn(deps, threadId)) { + const interrupted = await runAwaitedThreadStopCommand(deps, { + command: buildThreadStopCommand({ ...args, intent: "interrupt" }), + hostId: args.hostId, + }); + return interrupted.failure; + } + continue; + } + const afterRelease = getThread(deps.db, threadId); + if (afterRelease === null || !hasLiveThreadRuntime(deps, afterRelease)) { + return null; + } } - deps.logger.warn( - { threadId }, - "Host daemon kept an active turn on release; interrupting it for the explicit stop", - ); - await interruptRetainedThreadRuntime(deps, threadId, environment); + return { + error: new ApiError( + 409, + "invalid_request", + `Thread ${threadId} kept starting turns while it was being stopped`, + ), + intent: "interrupt", + }; } -async function interruptRetainedThreadRuntime( +function reviveThreadFromRetainedTurn( deps: RequestThreadStopForCurrentStateDeps, threadId: string, - environment: RequestThreadStopForCurrentStateEnvironment, -): Promise { +): boolean { const current = getThread(deps.db, threadId); - if (!current) { - return; + if (current === null) { + return false; } - const args: RequestThreadStopArgs = { - environmentId: environment.id, - hostId: environment.hostId, - interruptionReason: "manual-stop", - threadId, - }; if (current.status === "idle" || current.status === "error") { - const revived = applyLoggedThreadLifecycleEvent(deps, { + applyLoggedThreadLifecycleEvent(deps, { event: { type: "run.started" }, threadId, }); - if (!revived.applied) { - await runAwaitedThreadStopCommand(deps, { - command: buildThreadStopCommand({ ...args, intent: "interrupt" }), - hostId: args.hostId, - threadId, - }); - return; - } - } - if (getThread(deps.db, threadId)?.status !== "active") { - return; } - if (!markThreadStopRequested(deps, args)) { - return; - } - await runAwaitedThreadStopCommand(deps, { - command: buildThreadStopCommand({ ...args, intent: "interrupt" }), - hostId: args.hostId, - threadId, - }); + const revived = getThread(deps.db, threadId); + return revived !== null && hasLiveThreadRuntime(deps, revived); } async function runAwaitedThreadStopCommand( deps: RequestThreadStopForCurrentStateDeps, - args: { - command: ThreadStopCommand; - requireStopped?: boolean; - hostId: string; - threadId: string; - }, -): Promise { - return threadStopRequestDeduper.run(args.threadId, async () => { - inFlightThreadRpcGuard.claim(args.threadId, "thread.stop"); - try { - return await runLiveHostCommand(deps, { - command: args.command, - hostId: args.hostId, - timeoutMs: AWAITED_THREAD_STOP_TIMEOUT_MS, - }); - } catch (error) { - deps.logger.warn( - { err: error, intent: args.command.intent, threadId: args.threadId }, - "Awaited thread stop command failed", - ); - if (args.requireStopped) throw error; - if ( - args.command.intent === "release" && - !isHostUnavailableApiError(error) - ) { - throw error; - } - return null; - } finally { - inFlightThreadRpcGuard.release(args.threadId, "thread.stop"); - } - }); + args: { command: ThreadStopCommand; hostId: string }, +): Promise { + const { command } = args; + const kind = + command.intent === "release" ? "thread.stop.release" : "thread.stop"; + inFlightThreadRpcGuard.share(command.threadId, kind); + try { + const result = await runLiveHostCommand(deps, { + command, + hostId: args.hostId, + timeoutMs: AWAITED_THREAD_STOP_TIMEOUT_MS, + }); + return { failure: null, result }; + } catch (error) { + deps.logger.warn( + { err: error, intent: command.intent, threadId: command.threadId }, + "Awaited thread stop command failed", + ); + return { failure: { error, intent: command.intent }, result: null }; + } finally { + inFlightThreadRpcGuard.release(command.threadId, kind); + } } function interruptActiveTurnForThreadInTransaction( diff --git a/apps/server/test/public/public-thread-stop-runtime.test.ts b/apps/server/test/public/public-thread-stop-runtime.test.ts index cc1a7757b0..88b89a2ff0 100644 --- a/apps/server/test/public/public-thread-stop-runtime.test.ts +++ b/apps/server/test/public/public-thread-stop-runtime.test.ts @@ -26,8 +26,12 @@ import { seedTurnStarted, } from "../helpers/seed.js"; import { withTestHarness } from "../helpers/test-app.js"; +import { applyLoggedThreadLifecycleEvent } from "../../src/services/threads/lifecycle-outcome.js"; import { runQueuedMessageDispatch } from "../../src/services/threads/queued-message-dispatch.js"; -import { stopThreadForCurrentState } from "../../src/services/threads/thread-lifecycle.js"; +import { + requestActiveRuntimeThreadStopIfNeeded, + stopThreadForCurrentState, +} from "../../src/services/threads/thread-lifecycle.js"; describe("thread runtime stop", () => { it("releases an idle runtime without changing thread state", async () => { @@ -277,6 +281,262 @@ describe("thread runtime stop", () => { }); }); + it("interrupts a turn that starts while an explicit stop's release is pending", async () => { + await withTestHarness(async (harness) => { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "idle", visibility: "hidden" }, + }); + + const responsePromise = Promise.resolve( + harness.app.request(`/api/v1/threads/${thread.id}/stop`, { + method: "POST", + }), + ); + const release = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "release", + ); + + applyLoggedThreadLifecycleEvent(harness.deps, { + event: { type: "run.started" }, + threadId: thread.id, + }); + seedTurnStarted(harness.deps, { + environmentId: environment.id, + providerThreadId: "provider-new", + threadId: thread.id, + turnId: "turn-started-during-release", + }); + await reportQueuedCommandSuccess(harness, release, { + providerCheckpointId: null, + activeTurnRetained: true, + }); + + const interrupt = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "interrupt", + ); + const settledEarly = await Promise.race([ + responsePromise.then(() => "settled"), + new Promise((resolve) => setTimeout(() => resolve("pending"), 50)), + ]); + expect(settledEarly).toBe("pending"); + await reportQueuedCommandSuccess(harness, interrupt, { + providerCheckpointId: null, + }); + + expect((await responsePromise).status).toBe(200); + expect(getThread(harness.db, thread.id)?.status).toBe("idle"); + const completion = listEvents(harness.db, { threadId: thread.id }).find( + (event) => + event.type === "turn/completed" && + event.turnId === "turn-started-during-release", + ); + expect(JSON.parse(completion?.data ?? "{}")).toMatchObject({ + status: "interrupted", + }); + }); + }); + + it("makes a stop that joins a pending release wait for the escalated interrupt", async () => { + await withTestHarness(async (harness) => { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "idle", visibility: "hidden" }, + }); + + const first = harness.app.request(`/api/v1/threads/${thread.id}/stop`, { + method: "POST", + }); + const release = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "release", + ); + applyLoggedThreadLifecycleEvent(harness.deps, { + event: { type: "run.started" }, + threadId: thread.id, + }); + seedTurnStarted(harness.deps, { + environmentId: environment.id, + providerThreadId: "provider-new", + threadId: thread.id, + turnId: "turn-active-at-second-stop", + }); + const second = Promise.resolve( + harness.app.request(`/api/v1/threads/${thread.id}/stop`, { + method: "POST", + }), + ); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect( + listQueuedThreadCommands(harness, "thread.stop", thread.id), + ).toEqual([expect.objectContaining({ intent: "release" })]); + + await reportQueuedCommandSuccess(harness, release, { + providerCheckpointId: null, + activeTurnRetained: true, + }); + const interrupt = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "interrupt", + ); + const settledEarly = await Promise.race([ + second.then(() => "settled"), + new Promise((resolve) => setTimeout(() => resolve("pending"), 50)), + ]); + expect(settledEarly).toBe("pending"); + expect( + listQueuedThreadCommands(harness, "thread.stop", thread.id), + ).toHaveLength(1); + await reportQueuedCommandSuccess(harness, interrupt, { + providerCheckpointId: null, + }); + + expect((await first).status).toBe(200); + expect((await second).status).toBe(200); + expect(getThread(harness.db, thread.id)?.status).toBe("idle"); + expect( + listEvents(harness.db, { threadId: thread.id }).filter( + (event) => event.type === "system/thread/interrupted", + ), + ).toHaveLength(1); + }); + }); + + it("still dispatches an interrupt requested while an explicit stop's release is pending", async () => { + await withTestHarness(async (harness) => { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "idle", visibility: "hidden" }, + }); + + const responsePromise = harness.app.request( + `/api/v1/threads/${thread.id}/stop`, + { method: "POST" }, + ); + const release = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "release", + ); + applyLoggedThreadLifecycleEvent(harness.deps, { + event: { type: "run.started" }, + threadId: thread.id, + }); + seedTurnStarted(harness.deps, { + environmentId: environment.id, + providerThreadId: "provider-new", + threadId: thread.id, + turnId: "turn-stopped-by-request", + }); + requestActiveRuntimeThreadStopIfNeeded( + harness.deps, + { id: thread.id, status: "active" }, + { hostId: environment.hostId, id: environment.id }, + ); + const dispatched = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "interrupt", + ); + expect(getThread(harness.db, thread.id)?.status).toBe("stopping"); + + await reportQueuedCommandSuccess(harness, release, { + providerCheckpointId: null, + }); + const awaited = await waitForQueuedCommand( + harness, + (queued) => + queued.command.type === "thread.stop" && + queued.command.threadId === thread.id && + queued.command.intent === "interrupt" && + queued.row.cursor !== dispatched.row.cursor, + ); + await reportQueuedCommandSuccess(harness, dispatched, { + providerCheckpointId: null, + }); + await reportQueuedCommandSuccess(harness, awaited, { + providerCheckpointId: null, + }); + + expect((await responsePromise).status).toBe(200); + expect(getThread(harness.db, thread.id)?.status).toBe("idle"); + const completion = listEvents(harness.db, { threadId: thread.id }).find( + (event) => + event.type === "turn/completed" && + event.turnId === "turn-stopped-by-request", + ); + expect(JSON.parse(completion?.data ?? "{}")).toMatchObject({ + status: "interrupted", + }); + }); + }); + + it("rejects a caller that requires a stopped thread when the escalated interrupt fails", async () => { + await withTestHarness(async (harness) => { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "idle", visibility: "hidden" }, + }); + seedTurnStarted(harness.deps, { + environmentId: environment.id, + providerThreadId: "provider-retained", + threadId: thread.id, + turnId: "turn-retained", + }); + + const stopPromise = stopThreadForCurrentState( + harness.deps, + thread, + environment, + { requireStopped: true }, + ); + const settled = stopPromise.then( + () => ({ ok: true as const }), + (error: unknown) => ({ ok: false as const, error }), + ); + const release = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "release", + ); + await reportQueuedCommandSuccess(harness, release, { + providerCheckpointId: null, + activeTurnRetained: true, + }); + const interrupt = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "interrupt", + ); + await reportQueuedCommandError(harness, interrupt, { + errorCode: "test_interrupt_failure", + errorMessage: "Test interrupt failure", + }); + + const outcome = await settled; + expect(outcome.ok).toBe(false); + expect(getThread(harness.db, thread.id)?.status).toBe("stopping"); + }); + }); + it("settles background commands terminated by an idle runtime release", async () => { await withTestHarness(async (harness) => { const { environment, thread } = seedThreadFixture(harness, { diff --git a/packages/sdk/src/areas/threads.ts b/packages/sdk/src/areas/threads.ts index 63c723634b..ef3ec8d68c 100644 --- a/packages/sdk/src/areas/threads.ts +++ b/packages/sdk/src/areas/threads.ts @@ -597,6 +597,12 @@ export interface ThreadsArea { search(args: ThreadSearchArgs): Promise; send(args: ThreadSendArgs): Promise; spawn(args: ThreadSpawnArgs): Promise; + /** + * Stop the thread's work and release its loaded runtime. An explicit stop + * wins over running work: a turn the machine still runs while the thread + * looks idle or failed, or a turn that starts while the stop is delivered, + * is interrupted, and the call resolves after that turn has settled. + */ stop(args: ThreadActionArgs): Promise; tabs: ThreadTabsArea; context(args: ThreadStatusArgs): Promise; diff --git a/packages/templates/src/templates/bb-guide-threads.md b/packages/templates/src/templates/bb-guide-threads.md index 4361e8bb75..fe24d38baa 100644 --- a/packages/templates/src/templates/bb-guide-threads.md +++ b/packages/templates/src/templates/bb-guide-threads.md @@ -399,7 +399,10 @@ Lifecycle: The command succeeds when no runtime is loaded. Archive a finished hidden worker first, then stop it to release memory promptly. A stop that only releases an idle runtime adds no interruption: it leaves the timeline and any - pending interaction of that thread untouched. + pending interaction of that thread untouched. An explicit stop wins over work + that is still running: when the machine still runs a turn for a thread the app + shows as idle or failed, or a turn starts while the stop is being delivered, + the stop interrupts that turn and returns only after it has settled. bb thread unarchive [id] Unarchive a thread --self Unarchive current thread From 723bf21c6ce238ce31fcf6ca3a780b366a34aca0 Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Fri, 11 Sep 2026 23:02:16 +0000 Subject: [PATCH 5/8] Bump @get-bb/plugin-sdk to 0.4.85 main published 0.4.84, and this layer changes the SDK's bundled type declarations (the thread.stop result's activeTurnRetained field and the threads.stop declaration), so the npm version guard requires a new version. Co-Authored-By: Claude Code --- packages/domain/src/plugin-sdk-version.ts | 2 +- packages/plugin-sdk/package.json | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/domain/src/plugin-sdk-version.ts b/packages/domain/src/plugin-sdk-version.ts index 8ce10095c0..d8eedd8d3a 100644 --- a/packages/domain/src/plugin-sdk-version.ts +++ b/packages/domain/src/plugin-sdk-version.ts @@ -1,3 +1,3 @@ -export const PLUGIN_SDK_VERSION = "0.4.98"; +export const PLUGIN_SDK_VERSION = "0.4.99"; export const PLUGIN_SDK_MAJOR = Number(PLUGIN_SDK_VERSION.split(".", 1)[0]); diff --git a/packages/plugin-sdk/package.json b/packages/plugin-sdk/package.json index a10003bdd2..1b834c3593 100644 --- a/packages/plugin-sdk/package.json +++ b/packages/plugin-sdk/package.json @@ -1,6 +1,6 @@ { "name": "@get-bb/plugin-sdk", - "version": "0.4.98", + "version": "0.4.99", "homepage": "https://github.com/get-bb/bb#readme", "bugs": { "url": "https://github.com/get-bb/bb/issues" From ad31ba0d3f19a787f5d93f04bb9c3e44b7eb84fc Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Sat, 12 Sep 2026 00:34:25 +0000 Subject: [PATCH 6/8] Keep retained daemon background work live until interrupt succeeds --- .../src/command-handlers/thread.ts | 5 +- .../src/services/threads/thread-lifecycle.ts | 2 +- .../public/public-thread-stop-runtime.test.ts | 97 ++++++++++++++++++- 3 files changed, 99 insertions(+), 5 deletions(-) diff --git a/apps/host-daemon/src/command-handlers/thread.ts b/apps/host-daemon/src/command-handlers/thread.ts index 7dda5f56b3..2d402ec50e 100644 --- a/apps/host-daemon/src/command-handlers/thread.ts +++ b/apps/host-daemon/src/command-handlers/thread.ts @@ -2,7 +2,10 @@ import fs from "node:fs/promises"; import path from "node:path"; import type { AgentRuntimeBridgeLaunch } from "@bb/agent-runtime"; import { flattenPromptInputGroups } from "@bb/domain"; -import { COMPETING_TURN_ERROR_CODE, type HostDaemonCommandResult } from "@bb/host-daemon-contract"; +import { + COMPETING_TURN_ERROR_CODE, + type HostDaemonCommandResult, +} from "@bb/host-daemon-contract"; import type { RuntimeEntry } from "../runtime-manager.js"; import { CommandDispatchError, diff --git a/apps/server/src/services/threads/thread-lifecycle.ts b/apps/server/src/services/threads/thread-lifecycle.ts index 75a5cbd5c5..54eb586feb 100644 --- a/apps/server/src/services/threads/thread-lifecycle.ts +++ b/apps/server/src/services/threads/thread-lifecycle.ts @@ -1000,7 +1000,7 @@ export async function prepareReadyThreadTurnCommand( export function settleThreadStopCommandResult( args: SettleThreadStopCommandResultArgs, ): CommandResultSideEffectsResult { - if (args.report.ok) { + if (args.report.ok && args.report.result.activeTurnRetained !== true) { settleDanglingBackgroundTasksForStoppedThreadInTransaction(args.deps, { threadId: args.command.threadId, }); diff --git a/apps/server/test/public/public-thread-stop-runtime.test.ts b/apps/server/test/public/public-thread-stop-runtime.test.ts index 88b89a2ff0..e7c42edc33 100644 --- a/apps/server/test/public/public-thread-stop-runtime.test.ts +++ b/apps/server/test/public/public-thread-stop-runtime.test.ts @@ -29,7 +29,7 @@ import { withTestHarness } from "../helpers/test-app.js"; import { applyLoggedThreadLifecycleEvent } from "../../src/services/threads/lifecycle-outcome.js"; import { runQueuedMessageDispatch } from "../../src/services/threads/queued-message-dispatch.js"; import { - requestActiveRuntimeThreadStopIfNeeded, + requestThreadStopForCurrentState, stopThreadForCurrentState, } from "../../src/services/threads/thread-lifecycle.js"; @@ -441,9 +441,9 @@ describe("thread runtime stop", () => { threadId: thread.id, turnId: "turn-stopped-by-request", }); - requestActiveRuntimeThreadStopIfNeeded( + requestThreadStopForCurrentState( harness.deps, - { id: thread.id, status: "active" }, + { ...thread, status: "active" }, { hostId: environment.hostId, id: environment.id }, ); const dispatched = await waitForQueuedCommand( @@ -537,6 +537,97 @@ describe("thread runtime stop", () => { }); }); + it("keeps background work running after a retained release and failed interrupt", async () => { + await withTestHarness(async (harness) => { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "idle", visibility: "hidden" }, + }); + seedTurnStarted(harness.deps, { + environmentId: environment.id, + providerThreadId: "provider-retained", + threadId: thread.id, + turnId: "turn-retained", + }); + seedStoredEvent(harness.deps, { + threadId: thread.id, + environmentId: environment.id, + sequence: 2, + type: "item/started", + scope: turnScope("turn-retained"), + providerThreadId: "provider-retained", + itemId: "task:retained-command", + itemKind: "backgroundTask", + data: { + providerThreadId: "provider-retained", + item: { + type: "backgroundTask", + id: "task:retained-command", + taskType: "local_bash", + description: "Running command", + status: "pending", + taskStatus: "running", + skipTranscript: false, + }, + }, + }); + const responsePromise = harness.app.request( + `/api/v1/threads/${thread.id}/stop`, + { + method: "POST", + }, + ); + const release = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "release", + ); + await reportQueuedCommandSuccess(harness, release, { + providerCheckpointId: null, + activeTurnRetained: true, + }); + const interrupt = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "interrupt", + ); + await reportQueuedCommandError(harness, interrupt, { + errorCode: "test_interrupt_failure", + errorMessage: "Test interrupt failure", + }); + await responsePromise; + expect( + listEvents(harness.db, { threadId: thread.id }).filter( + (event) => event.type === "item/backgroundTask/completed", + ), + ).toHaveLength(0); + expect(getThread(harness.db, thread.id)?.status).toBe("stopping"); + const retry = harness.app.request(`/api/v1/threads/${thread.id}/stop`, { + method: "POST", + }); + const retriedInterrupt = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "thread.stop" && + command.threadId === thread.id && + command.intent === "interrupt", + ); + await reportQueuedCommandSuccess(harness, retriedInterrupt, { + providerCheckpointId: null, + }); + expect((await retry).status).toBe(200); + expect(getThread(harness.db, thread.id)?.status).toBe("idle"); + expect( + listEvents(harness.db, { threadId: thread.id }).filter( + (event) => event.type === "item/backgroundTask/completed", + ), + ).toHaveLength(1); + }); + }); + it("settles background commands terminated by an idle runtime release", async () => { await withTestHarness(async (harness) => { const { environment, thread } = seedThreadFixture(harness, { From da8f31e11ccc2e5de3c48ee35a5312ed54b94a06 Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Wed, 16 Sep 2026 20:02:19 +0000 Subject: [PATCH 7/8] Ignore Codex configuration flags in the scripted app server --- .../src/bridge/fake-codex-app-server.mjs | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs b/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs index d65c3bedf6..8e6c26ec6c 100644 --- a/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs +++ b/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs @@ -138,7 +138,20 @@ function runScriptedTurn(threadId) { // argv, not an env var: the bridge builds its child's environment from an // allowlist, so an env var set by a test never reaches this process. -const scriptPath = process.argv[2]; +function scriptPathFromArgs(args) { + for (let index = 0; index < args.length; index += 1) { + const argument = args[index]; + if (argument === "-c" || argument === "--config") { + index += 1; + continue; + } + if (argument.startsWith("-")) continue; + return argument; + } + return undefined; +} + +const scriptPath = scriptPathFromArgs(process.argv.slice(2)); const script = scriptPath ? JSON.parse(readFileSync(scriptPath, "utf8")) : null; const scriptedTurns = script?.turns ?? null; const requestLogPath = script?.requestLogPath ?? null; From 8e083db9a287e8c5ca35e6406643d1a827f8ef2e Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Wed, 16 Sep 2026 20:12:56 +0000 Subject: [PATCH 8/8] Document confirmation limits when a thread interrupt fails --- packages/sdk/src/areas/threads.ts | 4 +++- packages/templates/src/templates/bb-guide-threads.md | 4 +++- .../bb-guide/skills/bb-cli/references/failure-recovery.md | 5 ++++- 3 files changed, 10 insertions(+), 3 deletions(-) diff --git a/packages/sdk/src/areas/threads.ts b/packages/sdk/src/areas/threads.ts index ef3ec8d68c..292420f438 100644 --- a/packages/sdk/src/areas/threads.ts +++ b/packages/sdk/src/areas/threads.ts @@ -601,7 +601,9 @@ export interface ThreadsArea { * Stop the thread's work and release its loaded runtime. An explicit stop * wins over running work: a turn the machine still runs while the thread * looks idle or failed, or a turn that starts while the stop is delivered, - * is interrupted, and the call resolves after that turn has settled. + * is interrupted. The call waits for the interrupt attempt; if the machine + * cannot confirm it, the thread remains stopping. Inspect its status before + * treating the stop as confirmed. */ stop(args: ThreadActionArgs): Promise; tabs: ThreadTabsArea; diff --git a/packages/templates/src/templates/bb-guide-threads.md b/packages/templates/src/templates/bb-guide-threads.md index fe24d38baa..1aede931a0 100644 --- a/packages/templates/src/templates/bb-guide-threads.md +++ b/packages/templates/src/templates/bb-guide-threads.md @@ -402,7 +402,9 @@ Lifecycle: pending interaction of that thread untouched. An explicit stop wins over work that is still running: when the machine still runs a turn for a thread the app shows as idle or failed, or a turn starts while the stop is being delivered, - the stop interrupts that turn and returns only after it has settled. + the stop interrupts that turn and waits for the attempt. If the interrupt + fails, the thread remains stopping; check `bb thread show --json` before + treating the stop as confirmed. bb thread unarchive [id] Unarchive a thread --self Unarchive current thread diff --git a/plugins/bb-guide/skills/bb-cli/references/failure-recovery.md b/plugins/bb-guide/skills/bb-cli/references/failure-recovery.md index ca8219b62d..6292f44584 100644 --- a/plugins/bb-guide/skills/bb-cli/references/failure-recovery.md +++ b/plugins/bb-guide/skills/bb-cli/references/failure-recovery.md @@ -22,7 +22,10 @@ sendAt?, reason? })`. thread, treat that as intentional unless they ask you to continue. - Use `bb thread stop ` when a thread is stuck or no longer needed. - `bb thread stop ` also releases an idle or stuck agent runtime. The - command is idempotent and preserves thread history. + command is idempotent and preserves thread history. An explicit stop also + interrupts a turn the machine retains while the server sees idle or failed, + including a turn that starts during the stop. If interruption fails, the + thread remains stopping; inspect its status before treating Stop as confirmed. - Use `bb thread compact ` to send the built-in `/compact` command to an idle or errored thread. Completion or failure appears in the timeline. Provider support varies; consult its skill and reported capabilities. - Use `bb thread clear ` on an idle or failed thread to reset its active timeline and model context in place while keeping the same BB thread,