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..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, @@ -1009,6 +1008,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 +1028,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 () => { @@ -1147,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", @@ -1198,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/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..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 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 +407,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..54eb586feb 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, @@ -85,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, @@ -126,6 +128,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,16 +150,32 @@ type PreparedReadyThreadTurnCommand = | PreparedReadyTurnSubmitCommand; const threadStartRequestDeduper = createAsyncDeduper(); -const threadStopRequestDeduper = createAsyncDeduper(); +interface ThreadStopFailure { + error: unknown; + intent: ThreadStopCommand["intent"]; +} + +interface AwaitedThreadStopOutcome { + failure: ThreadStopFailure | null; + result: ThreadStopCommandResult | null; +} + +const threadStopRequestDeduper = createAsyncDeduper< + string, + 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}`; @@ -161,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)); } } @@ -184,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 { @@ -784,6 +820,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(); @@ -958,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, }); @@ -1462,101 +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 releaseIdleThreadRuntime(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 releaseIdleThreadRuntime(deps, thread.id, environment); } -async function releaseIdleThreadRuntime( - deps: RequestThreadStopForCurrentStateDeps, +function manualThreadStopArgs( threadId: string, - environment: RequestThreadStopForCurrentStateEnvironment | null, -): Promise { - if (environment === null) { - return; - } - await runAwaitedThreadStopCommand(deps, { - command: buildThreadStopCommand({ - environmentId: environment.id, - hostId: environment.hostId, - intent: "release", - threadId, - }), + environment: RequestThreadStopForCurrentStateEnvironment, +): RequestThreadStopArgs { + return { + environmentId: environment.id, hostId: environment.hostId, + interruptionReason: "manual-stop", threadId, - }); + }; } -async function runAwaitedThreadStopCommand( +async function stopThreadUntilSettled( deps: RequestThreadStopForCurrentStateDeps, - args: { - command: ThreadStopCommand; - requireStopped?: boolean; - hostId: string; - threadId: string; - }, -): Promise { - await threadStopRequestDeduper.run(args.threadId, async () => { - inFlightThreadRpcGuard.claim(args.threadId, "thread.stop"); - try { - 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; + threadId: string, + environment: RequestThreadStopForCurrentStateEnvironment | null, +): 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 ( - args.command.intent === "release" && - !isHostUnavailableApiError(error) + settled === null || + (settled.status !== "idle" && settled.status !== "error") ) { - throw error; + return null; } - } finally { - inFlightThreadRpcGuard.release(args.threadId, "thread.stop"); + } 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; + } + } + return { + error: new ApiError( + 409, + "invalid_request", + `Thread ${threadId} kept starting turns while it was being stopped`, + ), + intent: "interrupt", + }; +} + +function reviveThreadFromRetainedTurn( + deps: RequestThreadStopForCurrentStateDeps, + threadId: string, +): boolean { + const current = getThread(deps.db, threadId); + if (current === null) { + return false; + } + if (current.status === "idle" || current.status === "error") { + applyLoggedThreadLifecycleEvent(deps, { + event: { type: "run.started" }, + threadId, + }); + } + const revived = getThread(deps.db, threadId); + return revived !== null && hasLiveThreadRuntime(deps, revived); +} + +async function runAwaitedThreadStopCommand( + deps: RequestThreadStopForCurrentStateDeps, + 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 347c4dbd5e..e7c42edc33 100644 --- a/apps/server/test/public/public-thread-stop-runtime.test.ts +++ b/apps/server/test/public/public-thread-stop-runtime.test.ts @@ -23,10 +23,15 @@ import { seedStoredEvent, seedThread, seedThreadFixture, + 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 { + requestThreadStopForCurrentState, + stopThreadForCurrentState, +} from "../../src/services/threads/thread-lifecycle.js"; describe("thread runtime stop", () => { it("releases an idle runtime without changing thread state", async () => { @@ -61,6 +66,568 @@ 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("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", + }); + requestThreadStopForCurrentState( + harness.deps, + { ...thread, 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("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, { diff --git a/apps/server/test/threads/thread-send-dispatch.test.ts b/apps/server/test/threads/thread-send-dispatch.test.ts index 9059eaed91..f3a8d0ed1b 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, requestId: queued.command.requestId }; + } + + 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, requestId } = 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, + 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/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/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); }); 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" diff --git a/packages/sdk/src/areas/threads.ts b/packages/sdk/src/areas/threads.ts index 63c723634b..292420f438 100644 --- a/packages/sdk/src/areas/threads.ts +++ b/packages/sdk/src/areas/threads.ts @@ -597,6 +597,14 @@ 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. 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; 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..1aede931a0 100644 --- a/packages/templates/src/templates/bb-guide-threads.md +++ b/packages/templates/src/templates/bb-guide-threads.md @@ -399,7 +399,12 @@ 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 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, 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;