feat: bootstrap cli-agent runtime and wire executor, transport, chat, and recovery seams

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.changeset/cli-agent-runtime-bootstrap.md

@@ -0,0 +1,11 @@
+---
+"@runfusion/fusion": minor
+---
+
+Bootstrap the CLI Agent Executor runtime and wire it end-to-end.
+
+A new `createCliAgentRuntime` factory (engine) constructs the per-project bundle — a `CliSessionStore` over the project's existing core Database, a per-runtime adapter registry with all five bundled adapters, the `CliSessionManager` (PTY lifecycle), the `TelemetryHub` (per-session token registry rebuilt from live records), and the `CliResumeCoordinator` (relaunch re-mints a hook token + rewrites hook scripts) — returning the executor bundle, the `isWorktreeResumeReserved` / `isCliSessionWaitingOnInput` predicates, and a scoped `dispose`.
+
+The runtime is instantiated per project in `InProcessRuntime` behind the `experimentalFeatures.cliAgentExecutor` flag (opt-in, matching the `workflowGraphExecutor` precedent): the bundle threads into `TaskExecutorOptions.cliAgentRuntime`, the predicates feed the self-healing idle-worktree sweep and the stuck-task detector, and `resumeCoordinator.recoverOnStart()` runs non-blocking after engine start (errors logged, never thrown). The dashboard hook endpoint URL is derived from a server-threaded option, falling back to a localhost URL from `FUSION_DASHBOARD_PORT` (default 4040).
+
+The dashboard now resolves the project's `TelemetryHub` via `cliAgentHubResolver`, mounts the cli-sessions transport from the runtime's manager + store, and brokers cli-backed chat sends: a chat session with a `cliExecutorAdapterId` routes composer sends to a `CliChatSessionRunner` (instead of the model agent loop), and the hub's sanitized telemetry is routed per-session into the runner's transcript handler.

packages/cli/src/commands/dashboard.ts

@@ -25,6 +25,9 @@ import {
 } from "@fusion/core";
 import {
   createServer,
+  AttachTicketStore,
+  CliInputAttributionLog,
+  CliConfirmAdvanceRegistry,
   GitHubClient,
   createSkillsAdapter,
   getCliPackageVersion,
@@ -1741,9 +1744,33 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
     // to createServer — routes derived from getPluginRoutes() rely on it.
     await phaseTime("pluginLoadingPromise (await)", () => pluginLoadingPromise);
 
+    // ── CLI Agent Executor: hub resolver + session transport ─────────────
+    //
+    // The hook route validates a per-session token against the project's live
+    // TelemetryHub; resolve it from that project's engine. The cli-sessions
+    // transport (REST + WS attach) is supplied from the cwd project's runtime
+    // (the canonical single-project surface) when the experimental flag is on.
+    //
+    const cliAgentHubResolver = (projectId: string | undefined, _sessionId: string) => {
+      const engine = projectId ? engineManager.getEngine(projectId) : cwdEngine;
+      return engine?.getCliAgentRuntime()?.bundle.hub;
+    };
+    const cwdCliAgentRuntime = cwdEngine?.getCliAgentRuntime();
+    const cliSessionTransport = cwdCliAgentRuntime
+      ? {
+          manager: cwdCliAgentRuntime.bundle.manager,
+          store: cwdCliAgentRuntime.bundle.store,
+          ticketStore: new AttachTicketStore(),
+          attributionLog: new CliInputAttributionLog(),
+          confirmAdvance: new CliConfirmAdvanceRegistry(),
+        }
+      : undefined;
+
     app = createServer(store, {
       engine: cwdEngine,
       engineManager,
+      cliAgentHubResolver,
+      cliSessionTransport,
       hybridExecutor,
       centralCore: centralCoreForEngine,
       authStorage: dashboardAuthStorage,

packages/dashboard/src/__tests__/chat-manager-cli-send.test.ts

@@ -0,0 +1,71 @@
+/**
+ * ChatManager.sendMessage cli-agent send-branch (CLI Agent Executor integration).
+ *
+ * When a chat session selects a cli-agent executor (`cliExecutorAdapterId`),
+ * sendMessage must broker the composer text to the injected CliChatSessionRunner
+ * (ensureSession + send) rather than running the model agent loop. Narrow fakes:
+ * no real ChatStore, no pi-ai agent, no PTY, no network, no port 4040.
+ */
+
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { ChatManager } from "../chat.js";
+
+const mockChatStore = {
+  getSession: vi.fn(),
+  createSession: vi.fn(),
+  addMessage: vi.fn(),
+  getMessages: vi.fn(),
+  updateSession: vi.fn(),
+  setCliSessionFile: vi.fn(),
+  setInFlightGeneration: vi.fn(),
+  getRoomMessages: vi.fn(),
+};
+
+function makeManager(): ChatManager {
+  return new ChatManager(mockChatStore as never, "/tmp/test");
+}
+
+describe("ChatManager.sendMessage — cli-agent send branch", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("routes a cli-executor chat session's composer send to runner.send", async () => {
+    mockChatStore.getSession.mockReturnValue({
+      id: "chat-cli",
+      cliExecutorAdapterId: "claude-code",
+      projectId: "proj-1",
+    });
+
+    const ensureSession = vi.fn(async () => "cli-session-1");
+    const send = vi.fn(async () => "sent" as const);
+    const manager = makeManager();
+    manager.setCliChatRunner({ ensureSession, send }, "proj-1");
+
+    await manager.sendMessage("chat-cli", "hello agent");
+
+    expect(ensureSession).toHaveBeenCalledWith("chat-cli", { projectId: "proj-1" });
+    expect(send).toHaveBeenCalledWith("chat-cli", "hello agent");
+    // The model-agent path persists in-flight generation state; the cli branch
+    // must NOT touch it.
+    expect(mockChatStore.setInFlightGeneration).not.toHaveBeenCalled();
+  });
+
+  it("uses the session's projectId when no explicit runner projectId is set", async () => {
+    mockChatStore.getSession.mockReturnValue({
+      id: "chat-cli2",
+      cliExecutorAdapterId: "codex",
+      projectId: "proj-from-session",
+    });
+    const ensureSession = vi.fn(async () => "cli-session-2");
+    const send = vi.fn(async () => "queued" as const);
+    const manager = makeManager();
+    // No projectId passed to setCliChatRunner → falls back to session.projectId.
+    manager.setCliChatRunner({ ensureSession, send });
+
+    await manager.sendMessage("chat-cli2", "queued please");
+
+    expect(ensureSession).toHaveBeenCalledWith("chat-cli2", { projectId: "proj-from-session" });
+    expect(send).toHaveBeenCalledWith("chat-cli2", "queued please");
+  });
+});

packages/dashboard/src/__tests__/cli-agent-runtime-wiring.test.ts

@@ -0,0 +1,122 @@
+// @vitest-environment node
+
+/**
+ * CLI Agent Executor server-wiring contract (integration bootstrap).
+ *
+ * Proves that a real `createCliAgentRuntime` bundle (over a temp in-memory DB,
+ * PTY mocked at the loadPty seam) satisfies the shapes the dashboard ServerOptions
+ * consume:
+ *   - `cliAgentHubResolver(projectId, sessionId)` resolves the project's live
+ *     TelemetryHub from the runtime bundle.
+ *   - `cliSessionTransport` accepts the runtime's manager + store and the
+ *     transport-owned ticket/attribution/confirm singletons, and the
+ *     cli-sessions router mounts against that dep without error.
+ *
+ * No real PTY, no network, no port 4040.
+ */
+
+import express from "express";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { mkdtempSync } from "node:fs";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { rm } from "node:fs/promises";
+import { Database } from "@fusion/core";
+import type { IPty } from "node-pty";
+import { createCliAgentRuntime, type BootstrappedCliAgentRuntime } from "@fusion/engine";
+import {
+  AttachTicketStore,
+  CliInputAttributionLog,
+  CliConfirmAdvanceRegistry,
+} from "../cli-session-transport.js";
+import { createCliSessionsRouter } from "../routes/cli-sessions.js";
+import { request } from "../test-request.js";
+
+function mockPty(): typeof import("node-pty") {
+  return {
+    spawn() {
+      return {
+        pid: 1,
+        onData: () => ({ dispose() {} }),
+        onExit: () => ({ dispose() {} }),
+        write() {},
+        resize() {},
+        pause() {},
+        resume() {},
+        kill() {},
+        clear() {},
+      } as unknown as IPty;
+    },
+  } as unknown as typeof import("node-pty");
+}
+
+describe("cli-agent runtime server wiring", () => {
+  let runtime: BootstrappedCliAgentRuntime;
+  let db: Database;
+  let tmpDir: string;
+
+  beforeEach(() => {
+    tmpDir = mkdtempSync(join(tmpdir(), "fn-cli-wiring-"));
+    const fusionDir = join(tmpDir, ".fusion");
+    db = new Database(fusionDir, { inMemory: true });
+    db.init();
+    runtime = createCliAgentRuntime({
+      fusionDir,
+      db,
+      projectId: "proj-a",
+      hookEndpointUrl: "http://127.0.0.1:4040/api/cli-agent/hooks",
+      managerOptions: { loadPty: async () => mockPty() },
+    });
+  });
+
+  afterEach(async () => {
+    runtime.dispose();
+    db.close();
+    await rm(tmpDir, { recursive: true, force: true });
+  });
+
+  it("cliAgentHubResolver resolves the project's hub from the runtime bundle", () => {
+    const engines = new Map([["proj-a", { getCliAgentRuntime: () => runtime }]]);
+    const cliAgentHubResolver = (projectId: string | undefined, _sessionId: string) => {
+      const engine = projectId ? engines.get(projectId) : undefined;
+      return engine?.getCliAgentRuntime()?.bundle.hub;
+    };
+
+    expect(cliAgentHubResolver("proj-a", "cli-1")).toBe(runtime.bundle.hub);
+    expect(cliAgentHubResolver("missing", "cli-1")).toBeUndefined();
+    expect(cliAgentHubResolver(undefined, "cli-1")).toBeUndefined();
+  });
+
+  it("cliSessionTransport dep is satisfied by the runtime manager + store, and the router mounts", async () => {
+    // Seed a session so a transport-backed list route returns it.
+    runtime.bundle.store.createSession({
+      adapterId: runtime.bundle.registry.ids()[0],
+      projectId: "proj-a",
+      purpose: "execute",
+      taskId: "FN-1",
+      worktreePath: "/tmp/wt",
+      agentState: "busy",
+    });
+
+    const transport = {
+      manager: runtime.bundle.manager,
+      store: runtime.bundle.store,
+      ticketStore: new AttachTicketStore(),
+      attributionLog: new CliInputAttributionLog(),
+      confirmAdvance: new CliConfirmAdvanceRegistry(),
+    };
+
+    const app = express();
+    app.use(express.json());
+    app.use("/api/cli-sessions", createCliSessionsRouter(transport));
+
+    const res = await request(
+      app as unknown as (req: import("http").IncomingMessage, res: import("http").ServerResponse) => void,
+      "GET",
+      "/api/cli-sessions?projectId=proj-a",
+    );
+    expect(res.status).toBe(200);
+    const sessions = res.body.sessions as Array<{ taskId?: string }>;
+    expect(sessions.some((s) => s.taskId === "FN-1")).toBe(true);
+  });
+});

packages/dashboard/src/chat.ts

@@ -735,6 +735,29 @@ export class ChatManager {
     private messageStore?: MessageStore,
   ) {}
 
+  /**
+   * Runner for CLI-agent-backed chat sessions (CLI Agent Executor). When a chat
+   * session selects a cli-agent executor (`cliExecutorAdapterId`), composer sends
+   * are brokered to the live PTY through this runner instead of the model agent
+   * loop. Injected post-construction (the runtime is built per-project at boot,
+   * after the ChatManager) so the positional ctor stays stable.
+   */
+  private cliChatRunner?: {
+    ensureSession(chatSessionId: string, opts: { projectId: string; worktreePath?: string | null }): Promise<string>;
+    send(chatSessionId: string, text: string): Promise<"sent" | "queued">;
+  };
+  /** Project id used when the runner spawns a CLI session for a chat. */
+  private cliChatProjectId?: string;
+
+  /** Wire (or clear) the CLI-agent chat runner and its owning project id. */
+  setCliChatRunner(
+    runner: ChatManager["cliChatRunner"] | undefined,
+    projectId?: string,
+  ): void {
+    this.cliChatRunner = runner;
+    this.cliChatProjectId = projectId;
+  }
+
   private queueInFlightGenerationPersist(sessionId: string, snapshot: ChatInFlightGenerationState | null): void {
     const existingTimer = this.inFlightPersistTimers.get(sessionId);
     if (existingTimer) {
@@ -1392,6 +1415,26 @@ export class ChatManager {
     const broadcastOptions = { generationId };
 
     const session = this.chatStore.getSession(sessionId);
+
+    // CLI-agent-backed chat: a session that selected a cli-agent executor brokers
+    // its composer sends to the live PTY (via the runner) rather than running the
+    // model agent loop. The runner persists the user message + the transcript.
+    if (session?.cliExecutorAdapterId && this.cliChatRunner) {
+      const runner = this.cliChatRunner;
+      try {
+        await runner.ensureSession(sessionId, {
+          projectId: this.cliChatProjectId ?? session.projectId ?? "",
+        });
+        await runner.send(sessionId, content);
+      } finally {
+        const current = this.activeGenerations.get(sessionId);
+        if (current?.generationId === generationId) {
+          this.activeGenerations.delete(sessionId);
+        }
+      }
+      return;
+    }
+
     let agentResult: AgentResult | undefined;
     let accumulatedThinking = "";
     let accumulatedText = "";

packages/dashboard/src/index.ts

@@ -77,3 +77,14 @@ export {
   MAX_CARRY_LENGTH,
   type NeutralizeResult,
 } from "./cli-session-output-filter.js";
+
+// CLI Agent Executor transport dependencies — re-exported so the CLI boot
+// (packages/cli dashboard command) can construct the per-session attach-ticket
+// store, input-attribution log, and confirm-advance registry that the
+// cli-sessions transport routes require, then thread them into ServerOptions.
+export {
+  AttachTicketStore,
+  CliInputAttributionLog,
+  CliConfirmAdvanceRegistry,
+  type CliSessionTransportDeps,
+} from "./cli-session-transport.js";

packages/dashboard/src/server.ts

@@ -56,6 +56,7 @@ import {
   rehydrateFromStore as rehydrateMilestoneSliceSessions,
 } from "./milestone-slice-interview.js";
 import { ChatManager } from "./chat.js";
+import { CliChatSessionRunner } from "./cli-chat.js";
 import { stopAllDevServers } from "./dev-server-routes.js";
 import type { SkillsAdapter } from "./skills-adapter.js";
 import { createAuthMiddleware, authenticateUpgradeRequest, getDaemonToken } from "./auth-middleware.js";
@@ -1136,6 +1137,73 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
     options?.engine?.getMessageStore(),
   );
 
+  // CLI Agent Executor — chat surface wiring. When the cli-session transport is
+  // supplied (the runtime is live), broker cli-backed chat sends to the PTY and
+  // route the project hub's sanitized telemetry into the runner's transcript
+  // handler. The listener is keyed per-session inside one closure so it composes
+  // safely even if other taps exist.
+  if (options?.cliSessionTransport && options.cliAgentHubResolver) {
+    try {
+      const cliTransportStore = options.cliSessionTransport.store;
+      // The transport's `manager` is typed for the attach/inject transport slice;
+      // the chat runner additionally needs `spawn`. The concrete engine
+      // CliSessionManager provides both — widen via a structural cast to the
+      // spawn/inject slice the runner consumes.
+      const spawnInject = options.cliSessionTransport.manager as unknown as {
+        spawn: (opts: {
+          adapterId: string;
+          projectId: string;
+          purpose: "chat";
+          chatSessionId: string;
+          worktreePath?: string | null;
+          resume?: { sessionId: string; nativeSessionId: string };
+        }) => Promise<{ id: string; nativeSessionId: string | null; agentState: string }>;
+        inject: (sessionId: string, text: string) => Promise<void>;
+      };
+      // The runner needs spawn/inject (manager) + a fresh session record getter
+      // (store). Compose the slice the runner expects so flush decisions read
+      // authoritative records.
+      const cliChatRunner = new CliChatSessionRunner({
+        store: chatStore,
+        manager: {
+          spawn: (opts) => spawnInject.spawn(opts),
+          inject: (sessionId, text) => spawnInject.inject(sessionId, text),
+          getSession: (sessionId) => {
+            const r = cliTransportStore.getSession(sessionId);
+            return r
+              ? { id: r.id, nativeSessionId: r.nativeSessionId, agentState: r.agentState }
+              : undefined;
+          },
+        },
+      });
+      chatManager.setCliChatRunner(cliChatRunner, options.engine?.getProjectId?.());
+      const hub = options.cliAgentHubResolver(undefined, "");
+      if (hub) {
+        hub.setEventListener((cliSessionId, event) => {
+          // Per-session routing inside one listener: map the CLI session id to its
+          // owning chat session (only chat-purpose sessions carry chatSessionId);
+          // non-chat sessions (task/validator) are ignored here.
+          const record = cliTransportStore.getSession(cliSessionId);
+          const chatSessionId = record?.chatSessionId;
+          if (!chatSessionId) return;
+          void cliChatRunner
+            .handleTelemetry(chatSessionId, {
+              kind: event.kind,
+              text: event.text,
+              nativeSessionId: event.nativeSessionId,
+            })
+            .catch(() => {
+              // best-effort: a transcript-handler throw must never break ingest.
+            });
+        });
+      }
+    } catch (err) {
+      runtimeLogger.warn?.("CLI-agent chat runner wiring failed", {
+        message: err instanceof Error ? err.message : String(err),
+      });
+    }
+  }
+
   const runAiSessionCleanup = (maxAgeMs: number, source: "initial" | "scheduled") => {
     const result = aiSessionStore.cleanupStaleSessions(maxAgeMs);
     runtimeLogger.info("AI session cleanup summary", {

packages/engine/src/cli-agent/__tests__/runtime.test.ts

@@ -0,0 +1,152 @@
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { mkdtempSync } from "node:fs";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { rm } from "node:fs/promises";
+import { Database } from "@fusion/core";
+import type { IPty } from "node-pty";
+import { createCliAgentRuntime, type BootstrappedCliAgentRuntime } from "../runtime.js";
+import { BUNDLED_CLI_ADAPTERS } from "../adapters/index.js";
+
+// ── Mock PTY at the loadPtyModule seam (runtime construction must not touch a
+// real PTY; spawning is not exercised here). ───────────────────────────────────
+
+function makeMockPtyModule(): typeof import("node-pty") {
+  return {
+    spawn() {
+      const mock = {
+        pid: 4242,
+        cols: 80,
+        rows: 24,
+        process: "mock",
+        handleFlowControl: false,
+        onData: () => ({ dispose: () => {} }),
+        onExit: () => ({ dispose: () => {} }),
+        write: () => {},
+        resize: () => {},
+        pause: () => {},
+        resume: () => {},
+        kill: () => {},
+        clear: () => {},
+      };
+      return mock as unknown as IPty;
+    },
+  } as unknown as typeof import("node-pty");
+}
+
+interface Harness {
+  runtime: BootstrappedCliAgentRuntime;
+  db: Database;
+  tmpDir: string;
+  fusionDir: string;
+}
+
+function makeHarness(): Harness {
+  const tmpDir = mkdtempSync(join(tmpdir(), "fn-cli-runtime-test-"));
+  const fusionDir = join(tmpDir, ".fusion");
+  const db = new Database(fusionDir, { inMemory: true });
+  db.init();
+  const runtime = createCliAgentRuntime({
+    fusionDir,
+    db,
+    projectId: "proj-1",
+    hookEndpointUrl: "http://127.0.0.1:4040/api/cli-agent/hooks",
+    managerOptions: { loadPty: async () => makeMockPtyModule() },
+  });
+  return { runtime, db, tmpDir, fusionDir };
+}
+
+describe("createCliAgentRuntime", () => {
+  let h: Harness;
+
+  beforeEach(() => {
+    h = makeHarness();
+  });
+
+  afterEach(async () => {
+    h.runtime.dispose();
+    h.db.close();
+    await rm(h.tmpDir, { recursive: true, force: true });
+  });
+
+  it("constructs the full bundle with manager, hub, registry, and store", () => {
+    const { bundle } = h.runtime;
+    expect(bundle.manager).toBeDefined();
+    expect(bundle.hub).toBeDefined();
+    expect(bundle.registry).toBeDefined();
+    expect(bundle.store).toBeDefined();
+    expect(bundle.projectId).toBe("proj-1");
+    expect(bundle.hookEndpointUrl).toBe("http://127.0.0.1:4040/api/cli-agent/hooks");
+  });
+
+  it("registers all bundled adapters into a per-runtime registry", () => {
+    const ids = h.runtime.bundle.registry.ids().sort();
+    const expected = BUNDLED_CLI_ADAPTERS.map((a) => a.id).sort();
+    expect(ids).toEqual(expected);
+    expect(ids).toHaveLength(5);
+  });
+
+  it("does not pollute a second runtime's registry (no duplicate-registration)", () => {
+    // A second runtime over the SAME process registers the same adapters again;
+    // a per-runtime registry means no DuplicateCliAdapterError is thrown.
+    const second = createCliAgentRuntime({
+      fusionDir: h.fusionDir,
+      db: h.db,
+      projectId: "proj-2",
+      hookEndpointUrl: "http://127.0.0.1:4040/api/cli-agent/hooks",
+      managerOptions: { loadPty: async () => makeMockPtyModule() },
+    });
+    expect(second.bundle.registry.ids()).toHaveLength(5);
+    second.dispose();
+  });
+
+  it("isWorktreeResumeReserved reflects resume-eligible session records", () => {
+    const adapterId = BUNDLED_CLI_ADAPTERS[0].id;
+    // A live-on-restart record (busy) reserves its worktree.
+    h.runtime.bundle.store.createSession({
+      adapterId,
+      projectId: "proj-1",
+      purpose: "execute",
+      taskId: "FN-1",
+      worktreePath: "/wt/reserved",
+      agentState: "busy",
+    });
+    expect(h.runtime.isWorktreeResumeReserved("/wt/reserved")).toBe(true);
+    expect(h.runtime.isWorktreeResumeReserved("/wt/other")).toBe(false);
+  });
+
+  it("isCliSessionWaitingOnInput is true only when a task's session is waitingOnInput", () => {
+    const adapterId = BUNDLED_CLI_ADAPTERS[0].id;
+    h.runtime.bundle.store.createSession({
+      adapterId,
+      projectId: "proj-1",
+      purpose: "execute",
+      taskId: "FN-busy",
+      worktreePath: "/wt/busy",
+      agentState: "busy",
+    });
+    h.runtime.bundle.store.createSession({
+      adapterId,
+      projectId: "proj-1",
+      purpose: "execute",
+      taskId: "FN-wait",
+      worktreePath: "/wt/wait",
+      agentState: "waitingOnInput",
+    });
+    expect(h.runtime.isCliSessionWaitingOnInput("FN-wait")).toBe(true);
+    expect(h.runtime.isCliSessionWaitingOnInput("FN-busy")).toBe(false);
+    expect(h.runtime.isCliSessionWaitingOnInput("FN-unknown")).toBe(false);
+  });
+
+  it("exposes a resume coordinator whose recoverOnStart runs cleanly with no orphans", async () => {
+    const results = await h.runtime.resumeCoordinator.recoverOnStart();
+    expect(Array.isArray(results)).toBe(true);
+    expect(results).toHaveLength(0);
+  });
+
+  it("dispose tears down the manager without throwing and is idempotent", () => {
+    expect(() => h.runtime.dispose()).not.toThrow();
+    expect(() => h.runtime.dispose()).not.toThrow();
+    // Re-dispose in afterEach is also safe.
+  });
+});

packages/engine/src/cli-agent/runtime.ts

@@ -0,0 +1,176 @@
+/**
+ * createCliAgentRuntime — the per-project bootstrap that wires the CLI Agent
+ * Executor subsystem together (U-final integration).
+ *
+ * Every component (PTY session manager, telemetry hub, adapter registry, resume
+ * coordinator) is built with injection seams and tested in isolation; this
+ * factory is the single place that actually instantiates the live bundle and
+ * stitches the seams:
+ *
+ * - Builds a {@link CliSessionStore} over the project's EXISTING core Database
+ *   (never opens a second connection — the store is a thin query layer).
+ * - Registers all bundled adapters into a fresh {@link CliAdapterRegistry} (a
+ *   per-runtime registry, NOT the process-wide `defaultCliAdapterRegistry`, so
+ *   multi-project boots never collide on duplicate-registration).
+ * - Constructs the {@link CliSessionManager} (PTY lifecycle) and
+ *   {@link TelemetryHub} (per-session token registry, rebuilt from live store
+ *   records on construction).
+ * - Constructs the {@link CliResumeCoordinator}, wiring `reattachTelemetry` to
+ *   re-mint a hook token + rewrite the session's hook scripts on relaunch.
+ *
+ * It returns the {@link CliAgentRuntime} bundle the {@link TaskExecutor}
+ * consumes, plus the two narrow predicates the self-healing / stuck-task seams
+ * read, plus a `dispose` that tears the manager down cleanly (scoped SIGKILL of
+ * the runtime's own PTYs only — never the dashboard / port 4040).
+ */
+
+import { CliSessionStore } from "@fusion/core";
+import type { Database } from "@fusion/core";
+import { CliAdapterRegistry } from "./adapter.js";
+import { BUNDLED_CLI_ADAPTERS } from "./adapters/index.js";
+import { CliSessionManager, type CliSessionManagerOptions } from "./session-manager.js";
+import { TelemetryHub, type TelemetryHubOptions } from "./telemetry-hub.js";
+import { CliResumeCoordinator } from "./resume-coordinator.js";
+import { writeSessionHookScripts } from "./hook-scripts.js";
+import type { CliAgentRuntime } from "../executor.js";
+
+/** Options for {@link createCliAgentRuntime}. */
+export interface CreateCliAgentRuntimeOptions {
+  /** The project's `.fusion` dir (scratch root for hook scripts). */
+  fusionDir: string;
+  /** The project's already-open core Database (reused, never re-opened). */
+  db: Database;
+  /** Project this runtime drives (`cli_sessions.projectId`). */
+  projectId: string;
+  /**
+   * Absolute URL of the dashboard hook ingestion endpoint the generated hook
+   * scripts POST to (e.g. `http://127.0.0.1:4040/api/cli-agent/hooks`).
+   */
+  hookEndpointUrl: string;
+  /** Optional override for the hook scratch-dir root (tests). */
+  hookDirRoot?: string;
+  /** Optional notification dispatch forwarded to the TelemetryHub. */
+  onNotification?: TelemetryHubOptions["onNotification"];
+  /**
+   * Test seams forwarded to the {@link CliSessionManager} (e.g. a mocked node-pty
+   * loader so runtime construction never touches a real PTY).
+   */
+  managerOptions?: Pick<
+    CliSessionManagerOptions,
+    "loadPty" | "scrollbackBytes" | "concurrencyCeiling" | "highWatermark" | "injectionQuietWindowMs"
+  >;
+}
+
+/**
+ * The full bootstrapped CLI-agent runtime: the executor bundle, the predicates
+ * the self-healing + stuck-task seams read, the resume coordinator, and dispose.
+ */
+export interface BootstrappedCliAgentRuntime {
+  /** The bundle threaded into {@link TaskExecutorOptions.cliAgentRuntime}. */
+  bundle: CliAgentRuntime;
+  /** Engine-start orphan recovery sweep (call after engine start; errors logged). */
+  resumeCoordinator: CliResumeCoordinator;
+  /**
+   * Self-healing seam: whether a worktree path backs a resume-eligible session
+   * record (so idle-worktree sweeps treat it as in-use). Delegates to the resume
+   * coordinator's reservation set.
+   */
+  isWorktreeResumeReserved: (worktreePath: string) => boolean;
+  /**
+   * Stuck-task seam: whether a task's live CLI session is `waitingOnInput`
+   * (expected idleness — suppress stuck flagging). Reads the live store.
+   */
+  isCliSessionWaitingOnInput: (taskId: string) => boolean;
+  /** Tear down the PTY manager (scoped SIGKILL of this runtime's PTYs only). */
+  dispose: () => void;
+}
+
+/**
+ * Construct the per-project CLI-agent runtime bundle. Pure construction — no IO
+ * beyond the store's reads against the supplied Database; spawning a PTY or
+ * running recovery is the caller's job (`resumeCoordinator.recoverOnStart()`).
+ */
+export function createCliAgentRuntime(
+  options: CreateCliAgentRuntimeOptions,
+): BootstrappedCliAgentRuntime {
+  const { fusionDir, db, projectId, hookEndpointUrl } = options;
+
+  // 1. Store over the project's existing Database (thin query layer; no new conn).
+  const store = new CliSessionStore(fusionDir, db);
+
+  // 2. A per-runtime registry with every bundled adapter (not the process-wide
+  //    singleton — avoids duplicate-registration across multi-project boots).
+  const registry = new CliAdapterRegistry();
+  for (const adapter of BUNDLED_CLI_ADAPTERS) {
+    registry.register(adapter);
+  }
+
+  // 3. PTY session manager.
+  const manager = new CliSessionManager({
+    registry,
+    store,
+    ...options.managerOptions,
+  });
+
+  // 4. Telemetry hub — rebuilds its per-session token registry from live store
+  //    records on construction.
+  const hub = new TelemetryHub({
+    store,
+    onNotification: options.onNotification,
+  });
+
+  // 5. Resume coordinator. On relaunch, re-mint a hook token and rewrite the
+  //    session's hook scripts so the resumed CLI POSTs with a fresh, valid token.
+  const resumeCoordinator = new CliResumeCoordinator({
+    store,
+    manager,
+    registry,
+    reattachTelemetry: async (session) => {
+      const token = hub.issueToken(session.id);
+      await writeSessionHookScripts({
+        sessionId: session.id,
+        token,
+        endpointUrl: hookEndpointUrl,
+        dir: hookScriptDir(options, session.id),
+      });
+    },
+  });
+
+  const bundle: CliAgentRuntime = {
+    manager,
+    hub,
+    registry,
+    store,
+    projectId,
+    hookEndpointUrl,
+    hookDirRoot: options.hookDirRoot,
+  };
+
+  return {
+    bundle,
+    resumeCoordinator,
+    isWorktreeResumeReserved: (worktreePath: string) =>
+      resumeCoordinator.resumeReservedWorktrees().has(worktreePath),
+    isCliSessionWaitingOnInput: (taskId: string) => {
+      // A task's live session is "waiting on input" when any of its session
+      // records is in the waitingOnInput state. Defensive: a store error means
+      // "not waiting" (the stuck detector's own guard re-asserts this too).
+      try {
+        return store
+          .listByTask(taskId)
+          .some((s) => s.agentState === "waitingOnInput");
+      } catch {
+        return false;
+      }
+    },
+    dispose: () => {
+      manager.dispose();
+    },
+  };
+}
+
+/** Resolve the per-session hook scratch dir under the configured root. */
+function hookScriptDir(options: CreateCliAgentRuntimeOptions, sessionId: string): string {
+  const root = options.hookDirRoot ?? `${options.fusionDir}/cli-agent/hooks`;
+  return `${root}/${sessionId}`;
+}

packages/engine/src/index.ts

@@ -637,6 +637,12 @@ export {
   type CliSessionManagerOptions,
   type SpawnCliSessionOptions,
 } from "./cli-agent/session-manager.js";
+// CLI Agent Executor — per-project runtime bootstrap (integration).
+export {
+  createCliAgentRuntime,
+  type CreateCliAgentRuntimeOptions,
+  type BootstrappedCliAgentRuntime,
+} from "./cli-agent/runtime.js";
 // CLI Agent Executor — resume coordinator + self-healing/stuck integration (U8).
 export {
   CliResumeCoordinator,

packages/engine/src/project-engine.ts

@@ -800,6 +800,16 @@ export class ProjectEngine {
     return this.runtime.getSelfHealingManager();
   }
 
+  /**
+   * Get the bootstrapped CLI Agent Executor runtime (PTY manager + telemetry hub
+   * + adapter registry + resume coordinator), or undefined when the experimental
+   * flag is off. The dashboard reads this to resolve the project's TelemetryHub
+   * (hook route) and supply the cli-session transport dependency.
+   */
+  getCliAgentRuntime() {
+    return this.runtime.getCliAgentRuntime();
+  }
+
   /** Get the project working directory. */
   getWorkingDirectory(): string {
     return this.config.workingDirectory;

packages/engine/src/project-runtime.ts

@@ -55,6 +55,14 @@ export interface ProjectRuntimeConfig {
    * Useful when the caller (e.g. dashboard.ts) owns and watches the store.
    */
   externalTaskStore?: TaskStore;
+  /**
+   * Absolute URL of the dashboard's CLI-agent hook ingestion endpoint that
+   * generated hook scripts POST to (e.g. `http://127.0.0.1:4040/api/cli-agent/hooks`).
+   * Threaded from the dashboard boot once the listening port is known. When
+   * absent, the runtime derives a localhost URL from `FUSION_DASHBOARD_PORT`
+   * (falling back to 4040).
+   */
+  cliAgentHookEndpointUrl?: string;
 }
 
 /**

packages/engine/src/runtimes/in-process-runtime.ts

@@ -17,6 +17,8 @@ import { Scheduler } from "../scheduler.js";
 import type { PrMonitor, PrComment } from "../pr-monitor.js";
 import type { PrInfo } from "@fusion/core";
 import { TaskExecutor, type TaskExecutorOptions } from "../executor.js";
+import { isExperimentalFeatureEnabled } from "@fusion/core";
+import { createCliAgentRuntime, type BootstrappedCliAgentRuntime } from "../cli-agent/runtime.js";
 import { WorktreePool, isGitRepository, type PoolInvariantViolation } from "../worktree-pool.js";
 import { AgentSemaphore } from "../concurrency.js";
 import { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "../agent-heartbeat.js";
@@ -98,6 +100,13 @@ export class InProcessRuntime
   private worktreePool!: WorktreePool;
   private globalSemaphore?: AgentSemaphore;
   private stuckTaskDetector?: StuckTaskDetector;
+  /**
+   * Per-project CLI Agent Executor runtime bundle (PTY manager + telemetry hub +
+   * adapter registry + resume coordinator). Built in `start()` when the
+   * `cliAgentExecutor` experimental flag is on; threaded into the executor +
+   * self-healing/stuck seams; disposed in `stop()`.
+   */
+  private cliAgentRuntime?: BootstrappedCliAgentRuntime;
   private usageLimitPauser?: UsageLimitPauser;
   private selfHealingManager?: SelfHealingManager;
   private leaseManager?: MeshLeaseManager;
@@ -385,8 +394,29 @@ export class InProcessRuntime
 
       await yieldEventLoop();
 
+      // 5a-cli. Initialize the CLI Agent Executor runtime (behind the
+      // `cliAgentExecutor` experimental flag). Reuses the project's existing core
+      // Database; predicates feed the self-healing + stuck-task seams below.
+      if (isExperimentalFeatureEnabled(settings, "cliAgentExecutor")) {
+        try {
+          this.cliAgentRuntime = createCliAgentRuntime({
+            fusionDir: this.taskStore.getFusionDir(),
+            db: this.taskStore.getDatabase(),
+            projectId: this.config.projectId,
+            hookEndpointUrl: this.resolveCliAgentHookEndpointUrl(),
+          });
+          runtimeLog.log("CLI Agent Executor runtime initialized");
+        } catch (cliErr) {
+          runtimeLog.warn(
+            `CLI Agent Executor runtime initialization failed (cli-agent nodes will report a config error):`,
+            cliErr instanceof Error ? cliErr.message : cliErr,
+          );
+        }
+      }
+
       // 5b. Initialize TaskExecutor
       this.stuckTaskDetector = new StuckTaskDetector(this.taskStore, {
+        isCliSessionWaitingOnInput: this.cliAgentRuntime?.isCliSessionWaitingOnInput,
         beforeRequeue: (taskId, reason, event) => this.selfHealingManager?.checkStuckBudget(taskId, reason, event) ?? Promise.resolve(true),
         onLoopDetected: (event) => this.executor?.handleLoopDetected(event) ?? Promise.resolve(false),
         onStuck: (event) => {
@@ -447,6 +477,7 @@ export class InProcessRuntime
         pool: this.worktreePool,
         usageLimitPauser: this.usageLimitPauser,
         stuckTaskDetector: this.stuckTaskDetector,
+        cliAgentRuntime: this.cliAgentRuntime?.bundle,
         pluginRunner: this.pluginRunner,
         messageStore: this.messageStore,
         missionStore,
@@ -723,6 +754,7 @@ export class InProcessRuntime
       this.selfHealingManager = new SelfHealingManager(this.taskStore, {
         rootDir: this.config.workingDirectory,
         agentStore: this.agentStore,
+        isWorktreeResumeReserved: this.cliAgentRuntime?.isWorktreeResumeReserved,
         recoverCompletedTask: (task) => this.executor.recoverCompletedTask(task),
         recoverFailedPreMergeStep: (task) => this.executor.recoverFailedPreMergeWorkflowStep(task),
         getExecutingTaskIds: () => this.executor.getExecutingTaskIds(),
@@ -839,6 +871,25 @@ export class InProcessRuntime
         runtimeLog.warn(`Failed to stamp engineActiveSinceMs on runtime start: ${message}`);
       }
 
+      // 15. CLI Agent Executor: recover orphaned-live sessions left by a prior
+      // engine death. Non-blocking; errors are logged, never thrown.
+      if (this.cliAgentRuntime) {
+        const cliRuntime = this.cliAgentRuntime;
+        void cliRuntime.resumeCoordinator
+          .recoverOnStart()
+          .then((results) => {
+            if (results.length > 0) {
+              runtimeLog.log(`CLI Agent Executor recovered ${results.length} orphaned session(s) on start`);
+            }
+          })
+          .catch((err) => {
+            runtimeLog.warn(
+              `CLI Agent Executor recoverOnStart failed (continuing):`,
+              err instanceof Error ? err.message : err,
+            );
+          });
+      }
+
       this.setStatus("active");
       runtimeLog.log(`InProcessRuntime started for project ${this.config.projectId}`);
     } catch (error) {
@@ -890,6 +941,21 @@ export class InProcessRuntime
         runtimeLog.log("SelfHealingManager stopped");
       }
 
+      // 2c. Dispose the CLI Agent Executor runtime (scoped SIGKILL of this
+      // runtime's own PTYs only — never the dashboard / port 4040).
+      if (this.cliAgentRuntime) {
+        try {
+          this.cliAgentRuntime.dispose();
+          runtimeLog.log("CLI Agent Executor runtime disposed");
+        } catch (cliErr) {
+          runtimeLog.warn(
+            `CLI Agent Executor dispose failed:`,
+            cliErr instanceof Error ? cliErr.message : cliErr,
+          );
+        }
+        this.cliAgentRuntime = undefined;
+      }
+
       // 2. Stop routine scheduler (stops new routine triggers; in-flight executions continue)
       if (this.routineScheduler) {
         this.routineScheduler.stop();
@@ -1208,6 +1274,28 @@ export class InProcessRuntime
     return this.selfHealingManager;
   }
 
+  /**
+   * Get the bootstrapped CLI Agent Executor runtime (if the experimental flag is
+   * on and construction succeeded). The dashboard reads this to resolve the
+   * project's TelemetryHub (hook route) and supply the cli-session transport.
+   */
+  getCliAgentRuntime(): BootstrappedCliAgentRuntime | undefined {
+    return this.cliAgentRuntime;
+  }
+
+  /**
+   * Resolve the dashboard CLI-agent hook ingestion endpoint URL. Prefers the
+   * value threaded from server boot (once the listening port is known); falls
+   * back to a localhost URL derived from `FUSION_DASHBOARD_PORT` (default 4040).
+   */
+  private resolveCliAgentHookEndpointUrl(): string {
+    if (this.config.cliAgentHookEndpointUrl) {
+      return this.config.cliAgentHookEndpointUrl;
+    }
+    const port = Number(process.env.FUSION_DASHBOARD_PORT) || 4040;
+    return `http://127.0.0.1:${port}/api/cli-agent/hooks`;
+  }
+
   /**
    * Get the HeartbeatTriggerScheduler instance (if initialized).
    * Returns undefined when agent monitoring is not available.