Add bb thread import for existing ACP sessions

apps/app/.ladle/model-picker-query-provider.tsx

@@ -63,6 +63,7 @@ const STORY_PROVIDER_INFOS: ProviderInfo[] = STORY_PROVIDER_OPTIONS.map(
       supportsServiceTier: STORY_SERVICE_TIER_SUPPORT[provider.value] ?? false,
       supportsUserQuestion: true,
       supportsFork: true,
+      supportsSessionImport: false,
       supportedPermissionModes: [...supportedPermissionModes],
     },
   }),

apps/app/src/hooks/useThreadCreationOptions.test.tsx

@@ -62,6 +62,7 @@ function executionOptionsResponse(): SystemExecutionOptionsResponse {
           supportsServiceTier: true,
           supportsUserQuestion: true,
           supportsFork: true,
+          supportsSessionImport: false,
           supportedPermissionModes: ["accept-edits", "auto", "full"],
         },
       },
@@ -77,6 +78,7 @@ function executionOptionsResponse(): SystemExecutionOptionsResponse {
           supportsServiceTier: true,
           supportsUserQuestion: true,
           supportsFork: true,
+          supportsSessionImport: false,
           supportedPermissionModes: ["accept-edits", "auto", "full"],
         },
       },
@@ -128,6 +130,7 @@ function claudeExecutionOptionsResponse(): SystemExecutionOptionsResponse {
           supportsServiceTier: true,
           supportsUserQuestion: true,
           supportsFork: true,
+          supportsSessionImport: false,
           supportedPermissionModes: ["accept-edits", "auto", "full"],
         },
       },

apps/cli/src/commands/thread/import.ts

@@ -0,0 +1,88 @@
+import { Command } from "commander";
+import { threadVisibilitySchema, type Thread } from "@bb/domain";
+import { action } from "../../action.js";
+import { createCliBbSdk } from "../../client.js";
+import { outputJson, prependErrorContext } from "../helpers.js";
+import { parsePermissionMode, PERMISSION_MODE_HELP } from "./helpers.js";
+
+interface ThreadImportCommandOptions {
+  project: string;
+  provider: string;
+  providerSession: string;
+  host?: string;
+  cwd: string;
+  json?: boolean;
+  permissionMode?: string;
+  title?: string;
+  visibility?: string;
+}
+
+export function registerImportCommand(
+  parent: Command,
+  getUrl: () => string,
+): void {
+  parent
+    .command("import")
+    .description(
+      "Import an existing external ACP provider session as a thread",
+    )
+    .requiredOption("--project <id>", "Project the imported thread belongs to")
+    .requiredOption(
+      "--provider <acp-provider>",
+      'ACP provider that owns the session (e.g. "acp-omp")',
+    )
+    .requiredOption(
+      "--provider-session <external-session-id>",
+      "External provider session ID to import",
+    )
+    .option("--host <id>", "Host the session lives on (default: primary host)")
+    .requiredOption(
+      "--cwd <path>",
+      "Working directory the session ran in; must match the project source " +
+        "path or an existing workspace of the project. If the provider is a " +
+        "configured custom ACP agent that pins its own cwd, this must match " +
+        "that pinned directory exactly",
+    )
+    .option("--title <title>", "Thread title")
+    .option("--permission-mode <mode>", PERMISSION_MODE_HELP)
+    .option("--visibility <visibility>", "Thread visibility: visible or hidden")
+    .option("--json", "Print machine-readable JSON output")
+    .action(
+      action(async (opts: ThreadImportCommandOptions) => {
+        const permissionMode = parsePermissionMode(opts.permissionMode);
+        const visibility =
+          opts.visibility === undefined
+            ? undefined
+            : threadVisibilitySchema.parse(opts.visibility);
+
+        let thread: Thread;
+        try {
+          thread = await createCliBbSdk(getUrl()).threads.import({
+            projectId: opts.project,
+            providerId: opts.provider,
+            providerSessionId: opts.providerSession,
+            origin: "cli",
+            ...(opts.host === undefined ? {} : { hostId: opts.host }),
+            cwd: opts.cwd,
+            ...(opts.title === undefined ? {} : { title: opts.title }),
+            ...(permissionMode === undefined ? {} : { permissionMode }),
+            ...(visibility === undefined ? {} : { visibility }),
+          });
+        } catch (error: unknown) {
+          throw prependErrorContext(
+            `Failed to import provider session ${opts.providerSession}`,
+            error,
+          );
+        }
+
+        if (outputJson(opts, thread)) return;
+        console.log(`Thread imported: ${thread.id}`);
+        console.log(`Provider: ${opts.provider}`);
+        console.log(`Provider session: ${opts.providerSession}`);
+        console.log(`Status: ${thread.status}`);
+        if (thread.visibility === "hidden") {
+          console.log("Visibility: hidden");
+        }
+      }),
+    );
+}

apps/cli/src/commands/thread/index.ts

@@ -8,6 +8,7 @@ import { registerOrganizationCommands } from "./organization.js";
 import { registerShowCommand } from "./show.js";
 import { registerSpawnCommand } from "./spawn.js";
 import { registerForkCommand } from "./fork.js";
+import { registerImportCommand } from "./import.js";
 import { registerWaitCommand } from "./wait.js";
 
 export function registerThreadCommands(
@@ -18,6 +19,7 @@ export function registerThreadCommands(
   registerWaitCommand(thread, getUrl);
   registerSpawnCommand(thread, getUrl);
   registerForkCommand(thread, getUrl);
+  registerImportCommand(thread, getUrl);
   registerListCommand(thread, getUrl);
   registerShowCommand(thread, getUrl);
   registerOpenCommand(thread, getUrl);

apps/host-daemon/src/command-dispatch-support.ts

@@ -52,9 +52,11 @@ export interface CommandDispatchOptions {
     providerId: string;
     acpLaunchSpec?: HostDaemonAcpLaunchSpec;
     cwd?: string;
+    probeSessionImport?: boolean;
   }) => Promise<{
     models: AvailableModel[];
     selectedOnlyModels: AvailableModel[];
+    supportsSessionImport?: boolean;
   }>;
   getProviderCliStatusForProvider?: (
     providerId: string,
@@ -116,11 +118,16 @@ export async function shutdownDefaultListModelsRuntimes(): Promise<void> {
 }
 
 export async function defaultListModels(
-  args: { providerId: string; acpLaunchSpec?: HostDaemonAcpLaunchSpec },
+  args: {
+    providerId: string;
+    acpLaunchSpec?: HostDaemonAcpLaunchSpec;
+    probeSessionImport?: boolean;
+  },
   options: { bridgeBundleDir?: AgentRuntimeOptions["bridgeBundleDir"] } = {},
 ): Promise<{
   models: AvailableModel[];
   selectedOnlyModels: AvailableModel[];
+  supportsSessionImport?: boolean;
 }> {
   const runtimeKey =
     `${options.bridgeBundleDir ?? ""}` +

apps/host-daemon/src/command-dispatch.ts

@@ -464,6 +464,9 @@ const onlineRpcHandlers: OnlineRpcHandlerMap = {
       ...(command.acpLaunchSpec !== undefined
         ? { acpLaunchSpec: command.acpLaunchSpec }
         : {}),
+      ...(command.probeSessionImport !== undefined
+        ? { probeSessionImport: command.probeSessionImport }
+        : {}),
     }),
   "known_acp_agents.status": async (command) =>
     getKnownAcpAgentsStatus({ agents: command.agents }),

apps/host-daemon/src/command-handlers/thread.ts

@@ -229,6 +229,7 @@ export async function startThread(
       disallowedTools: command.disallowedTools,
       instructionMode: command.instructionMode,
       ...(command.fork ? { fork: command.fork } : {}),
+      ...(command.sessionImport ? { sessionImport: command.sessionImport } : {}),
     });
     return result;
   } catch (error) {

apps/host-daemon/src/command-router.ts

@@ -238,11 +238,23 @@ export class CommandRouter {
   private async executeLiveDaemonCommandBody(
     command: HostDaemonCommand,
   ): Promise<HostDaemonCommandResultForCommand> {
-    const result = await dispatchCommand(command, this.createDispatchOptions());
-    // Commands that emit thread events before completing preserve the previous
-    // event-before-result ordering under live RPC.
-    if (shouldFlushEventsBeforeReportingCommandResult(command)) {
-      await this.options.eventSink.flush();
+    // Commands that emit thread events before completing preserve the
+    // previous event-before-result ordering under live RPC. The flush runs
+    // in `finally`, not only on success: events already emitted before a
+    // failure (e.g. a thread/identity from a bind that completed just
+    // before dispatch threw, such as a session/load timeout) must still
+    // reach the server before the failure result settles, or a caller
+    // inferring "bind never completed" from the server event log
+    // (releaseFailedSessionImportReservationInTransaction) can race ahead
+    // of them.
+    const shouldFlush = shouldFlushEventsBeforeReportingCommandResult(command);
+    let result: Awaited<ReturnType<typeof dispatchCommand>>;
+    try {
+      result = await dispatchCommand(command, this.createDispatchOptions());
+    } finally {
+      if (shouldFlush) {
+        await this.options.eventSink.flush();
+      }
     }
     return parseHostDaemonCommandResultForCommand(command, result);
   }

apps/host-daemon/test/command/command-router.test.ts

@@ -54,6 +54,7 @@ interface CreateTurnSubmitCommandArgs {
 }
 
 interface CreateRouterArgs {
+  eventSink?: CommandRouterOptions["eventSink"];
   logger?: CommandRouterOptions["logger"];
   runtimeManager?: RuntimeManager;
 }
@@ -91,7 +92,7 @@ function createRouter(
 ): CommandRouter {
   return new CommandRouter({
     dataDir: "/tmp/bb-router-test-data",
-    eventSink: noopEventSink,
+    eventSink: args.eventSink ?? noopEventSink,
     fetchProjectAttachment: unexpectedProjectAttachmentFetch,
     logger: {
       debug: () => undefined,
@@ -392,4 +393,36 @@ describe("CommandRouter", () => {
     const stopResponse = await stopTask;
     expect(stopResponse.ok).toBe(true);
   });
+
+  it("flushes buffered events before reporting a thread.start failure", async () => {
+    const harness = createHarness({ workspacePath: "/tmp/env-router" });
+    await harness.manager.ensureEnvironment({
+      environmentId: "env-router",
+      workspacePath: "/tmp/env-router",
+    });
+    harness.runtime.startThread = async () => {
+      throw new Error("provider session/load timed out");
+    };
+    const flush = vi.fn(async () => undefined);
+    const router = createRouter(harness, {
+      eventSink: { emit: vi.fn(), flush },
+    });
+
+    const response = await runRouterCommand({
+      command: createThreadStartCommand(),
+      requestId: "start-flush-on-failure",
+      router,
+    });
+
+    expect(response).toMatchObject({
+      ok: false,
+      errorMessage: "provider session/load timed out",
+    });
+    // thread.start declares flushEventsBeforeResult: true; a late-arriving
+    // thread/identity buffered before the failure must still reach the
+    // server, or a caller inferring "bind never completed" from the event
+    // log (releaseFailedSessionImportReservationInTransaction) can race
+    // ahead of it.
+    expect(flush).toHaveBeenCalledTimes(1);
+  });
 });

apps/server/src/internal/events.ts

@@ -374,6 +374,11 @@ async function applyEventEffects(
     try {
       const event = entry.event;
       if (event.type === "turn/started") {
+        // Replayed history from an imported provider session is persisted for
+        // display only; it must not drive thread lifecycle transitions.
+        if (event.historical) {
+          continue;
+        }
         const turnId = requireThreadEventScopeTurnId({
           type: event.type,
           scope: event.scope,
@@ -402,6 +407,11 @@ async function applyEventEffects(
       }
 
       if (event.type === "turn/completed") {
+        // See turn/started above: imported-history frames complete with no
+        // turn-completion side effects.
+        if (event.historical) {
+          continue;
+        }
         const turnId = requireThreadEventScopeTurnId({
           type: event.type,
           scope: event.scope,

apps/server/src/routes/threads/base.ts

@@ -44,6 +44,7 @@ import {
 } from "../../services/threads/thread-lifecycle.js";
 import { createThreadFromRequest } from "../../services/threads/thread-create.js";
 import { createThreadForkFromRequest } from "../../services/threads/thread-fork.js";
+import { createThreadImportFromRequest } from "../../services/threads/thread-import.js";
 import { requireChildThreadsConfirmation } from "../../services/threads/child-thread-confirmation.js";
 import {
   toThreadListEntryResponses,
@@ -275,6 +276,11 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void {
     return context.json(toThreadResponseFromThread(deps, { thread }), 201);
   });
 
+  post(routes.import, async (context, payload) => {
+    const thread = await createThreadImportFromRequest(deps, payload);
+    return context.json(toThreadResponseFromThread(deps, { thread }), 201);
+  });
+
   get(routes.get, (context, query) => {
     const thread = requirePublicThread(deps.db, context.req.param("id"));
     return context.json(

apps/server/src/services/environments/environment-provisioning-internal.ts

@@ -1,5 +1,6 @@
 import { and, desc, eq, isNull } from "drizzle-orm";
 import {
+  deleteProviderSessionReservation,
   events,
   type DbConnection,
   type DbNotifier,
@@ -27,6 +28,7 @@ import {
   appendSystemErrorEventInTransaction,
   appendThreadProvisioningEventInTransaction,
   buildCwdBranchEntries,
+  getLastProviderThreadId,
 } from "../threads/thread-events.js";
 import {
   buildEnvironmentProvisionCommand,
@@ -557,6 +559,19 @@ function recordEnvironmentProvisioningFailureInTransaction(
     if (outcome.applied) {
       deps.hub.notifyThread(thread.id, ["status-changed"]);
     }
+    // An import's provider session reservation is claimed synchronously
+    // when the thread is created, before provisioning ever runs. If
+    // provisioning fails before thread.start is ever dispatched (this
+    // path), the import's bind never completed, so the reservation must
+    // not outlive it (see failThreadProvisioning in
+    // thread-provisioning-environment.ts for the sibling path, and
+    // releaseFailedSessionImportReservationInTransaction in
+    // thread-lifecycle.ts for the async thread.start-failure path). Gated
+    // on the thread never having recorded a provider identity, which makes
+    // this a no-op for threads that were never an import.
+    if (getLastProviderThreadId(deps, thread.id) === null) {
+      deleteProviderSessionReservation(deps.db, thread.id);
+    }
   }
 
   return true;

apps/server/src/services/plugins/plugin-api.ts

@@ -52,7 +52,12 @@ import type {
   PluginUi,
   StandardSchemaV1,
 } from "@bb/plugin-sdk";
-import type { BbSdk, ThreadForkArgs, ThreadSpawnArgs } from "@bb/sdk";
+import type {
+  BbSdk,
+  ThreadForkArgs,
+  ThreadImportArgs,
+  ThreadSpawnArgs,
+} from "@bb/sdk";
 import type { ServerLogger } from "../../types.js";
 import type { PluginInteractionResult } from "../interactions/pending-interactions.js";
 import { appendPluginLogLine } from "./plugin-log.js";
@@ -492,6 +497,16 @@ function wrapSdkForPlugin(sdk: BbSdk, pluginId: string): BbSdk {
             : {}),
         });
       },
+      import(args: ThreadImportArgs) {
+        const origin = args.origin ?? "plugin";
+        return sdk.threads.import({
+          ...args,
+          origin,
+          ...(origin === "plugin"
+            ? { originPluginId: args.originPluginId ?? pluginId }
+            : {}),
+        });
+      },
       spawn(args: ThreadSpawnArgs) {
         const origin = args.origin ?? "plugin";
         return sdk.threads.spawn({

apps/server/src/services/skills/builtin-skills/bb-cli/SKILL.md

@@ -161,6 +161,19 @@ message agents, or inspect projects, providers, and environments.
   creates an idle fork by default; add `--prompt`, select `--workspace
 isolated|reuse`, or anchor with `--source-seq-end`. Permission mode inherits
   the source thread unless explicitly overridden.
+- Use `bb thread import --project <id> --provider <acp-provider>
+  --provider-session <external-session-id> --cwd <path>` to import an
+  existing external ACP agent session (for example an `omp acp` session) as a
+  bb thread. The agent must support ACP `session/load`; bb replays the
+  session's history into the thread timeline and the thread lands idle, ready
+  for follow-up turns. `--cwd` is required: it is the caller's assertion of
+  the working directory the session ran in (bb has no way to read this back
+  from the external session), and must match the project source path or an
+  existing workspace already attached to the project — anything else is
+  refused. If the provider resolves to a configured custom ACP agent that
+  pins its own cwd, `--cwd` must match that pinned directory exactly, or the
+  import is refused. Pass `--host` when the session lives on a non-primary
+  machine.
 - Pass `--visibility hidden` for background/plugin workers that should remain
   out of sidebar organization without contributing unread/pending favicon
   attention. `bb thread list` excludes them by