Reconcile daemon turns the server lost track of on explicit stop

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(

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";
   }

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",

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<HostDaemonCommandResult<"thread.stop">> {
   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, {

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`,
     );
   }