Merge U7: cli-agent executor seam, task session lifecycle, hard-cancel integration

packages/core/src/index.ts

@@ -71,6 +71,9 @@ export type {
   WorkflowFieldType,
   WorkflowFieldOption,
   WorkflowFieldRender,
+  // CLI Agent Executor (U7): node-config executor typing.
+  WorkflowNodeExecutorKind,
+  WorkflowNodeExecutorConfig,
 } from "./workflow-ir-types.js";
 export { BUILTIN_CODING_WORKFLOW_IR } from "./builtin-coding-workflow-ir.js";
 export { BUILTIN_STEPWISE_CODING_WORKFLOW_IR } from "./builtin-stepwise-coding-workflow-ir.js";

packages/core/src/workflow-ir-types.ts

@@ -26,6 +26,50 @@ export interface WorkflowIrNode {
   config?: Record<string, unknown>;
 }
 
+/**
+ * Executor kinds selectable on a prompt/execute node's `config.executor` (CLI
+ * Agent Executor, U7). The engine reads `config.executor` as an open string; this
+ * union documents the recognized values and `WorkflowNodeExecutorConfig` the
+ * fields each one consumes. `config` itself stays an open `Record` so unknown
+ * keys remain forward-compatible.
+ *
+ * - `model`     (default): run the prompt on the configured/override model.
+ * - `agent`     : run as a named agent (adopt its model + persona).
+ * - `skill`     : invoke a named skill with the prompt as input.
+ * - `cli`       : run a named project script with the prompt via env.
+ * - `cli-agent` : drive a CLI coding agent (Claude Code / Codex / Droid / Pi /
+ *                 generic) in an engine-owned PTY for the execute step. Honors
+ *                 cancel/abort/re-entry semantics and positive-completion gating.
+ */
+export type WorkflowNodeExecutorKind = "model" | "agent" | "skill" | "cli" | "cli-agent";
+
+/**
+ * The cli-agent slice of a workflow node's `config`. These ride on the open
+ * `WorkflowIrNode.config` record (read at U7's executor seam); they are NOT a
+ * separate column. The resolved values are SNAPSHOTTED at session launch — a
+ * mid-run edit to the node config applies to the next run only.
+ */
+export interface WorkflowNodeExecutorConfig {
+  /** Selected executor kind for this node. */
+  executor?: WorkflowNodeExecutorKind;
+  /** cli-agent: adapter id to drive the session (resolved against the registry). */
+  cliAdapterId?: string;
+  /**
+   * cli-agent: autonomy posture (drives privileged flags + resume caps). Stored
+   * verbatim; structured but extensible (mirrors `CliAutonomyPosture`).
+   */
+  cliAutonomy?: {
+    autoApprove?: boolean;
+    maxResumeAttempts?: number;
+    [key: string]: unknown;
+  };
+  /**
+   * cli-agent: notification settings for waiting-on-input events on this node
+   * (origin R2/R11). Opaque to the engine seam; forwarded to the dispatch.
+   */
+  cliNotify?: Record<string, unknown>;
+}
+
 export interface WorkflowIrEdge {
   from: string;
   to: string;

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

@@ -0,0 +1,396 @@
+import "./executor-test-helpers.js";
+import { beforeEach, afterEach, describe, expect, it, vi } from "vitest";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+// node:fs is mocked by executor-test-helpers; use node:fs/promises (unmocked) for
+// real temp-dir + hook-script I/O.
+import { mkdtemp, rm } from "node:fs/promises";
+import { Database, CliSessionStore } from "@fusion/core";
+import type { IPty } from "node-pty";
+import { TaskExecutor, type CliAgentRuntime } from "../executor.js";
+import { resetExecutorMocks } from "./executor-test-helpers.js";
+import { CliSessionManager } from "../cli-agent/session-manager.js";
+import { TelemetryHub } from "../cli-agent/telemetry-hub.js";
+import { CliAdapterRegistry, type CliAgentAdapter } from "../cli-agent/adapter.js";
+
+type Listener = (...args: any[]) => void;
+
+// ── Mock PTY ────────────────────────────────────────────────────────────────
+
+interface MockPty extends IPty {
+  written: string[];
+  killed: boolean;
+  killSignal: string | undefined;
+  emitData(data: string): void;
+  emitExit(exitCode: number, signal?: number): void;
+}
+interface MockState {
+  ptys: MockPty[];
+}
+function makeMockPtyModule(state: MockState): typeof import("node-pty") {
+  return {
+    spawn() {
+      let dataCb: ((d: string) => void) | undefined;
+      let exitCb: ((e: { exitCode: number; signal?: number }) => void) | undefined;
+      const mock: MockPty = {
+        pid: 3000 + state.ptys.length,
+        cols: 80,
+        rows: 24,
+        process: "mock",
+        handleFlowControl: false,
+        written: [],
+        killed: false,
+        killSignal: undefined,
+        onData: (cb: (d: string) => void) => {
+          dataCb = cb;
+          return { dispose() {} };
+        },
+        onExit: (cb: (e: { exitCode: number; signal?: number }) => void) => {
+          exitCb = cb;
+          return { dispose() {} };
+        },
+        on() {},
+        write(data: string) {
+          mock.written.push(data);
+        },
+        resize() {},
+        clear() {},
+        kill(signal?: string) {
+          mock.killed = true;
+          mock.killSignal = signal;
+          exitCb?.({ exitCode: 0, signal: signal === "SIGKILL" ? 9 : undefined });
+        },
+        pause() {},
+        resume() {},
+        emitData(d: string) {
+          dataCb?.(d);
+        },
+        emitExit(exitCode: number, signal?: number) {
+          exitCb?.({ exitCode, signal });
+        },
+      } as any;
+      state.ptys.push(mock);
+      return mock as unknown as IPty;
+    },
+  } as unknown as typeof import("node-pty");
+}
+
+function scriptedAdapter(): CliAgentAdapter {
+  return {
+    id: "scripted",
+    name: "Scripted",
+    capabilities: { nativeDone: true, nativeWaiting: true, transcriptSource: "hooks", supportsResume: true },
+    buildLaunch: () => ({ command: "scripted", args: [] }),
+    buildEnvAllowlist: () => ["PATH"],
+    createReadinessDetector: () => {
+      let ready = false;
+      return {
+        observe(chunk: string) {
+          if (chunk.includes("READY")) ready = true;
+          return ready;
+        },
+      };
+    },
+    formatInjection: (text) => ({ payload: text.endsWith("\r") ? text : `${text}\r` }),
+  };
+}
+
+// ── Store stub satisfying the runGraphCustomNode/cli-agent code paths ──────────
+
+function createStore(task: any) {
+  const listeners = new Map<string, Set<Listener>>();
+  const logs: string[] = [];
+  return {
+    logs,
+    store: {
+      on: vi.fn((event: string, listener: Listener) => {
+        const set = listeners.get(event) ?? new Set<Listener>();
+        set.add(listener);
+        listeners.set(event, set);
+      }),
+      off: vi.fn(),
+      getTask: vi.fn().mockImplementation(async () => task),
+      logEntry: vi.fn().mockImplementation(async (_id: string, msg: string) => {
+        logs.push(msg);
+      }),
+      updateTask: vi.fn().mockResolvedValue(undefined),
+      getSettings: vi.fn().mockResolvedValue({ globalPause: false, enginePaused: false }),
+      listTasks: vi.fn().mockResolvedValue([]),
+    } as any,
+  };
+}
+
+describe("cli-agent executor seam (U7)", () => {
+  let tmpDir: string;
+  let fusionDir: string;
+  let db: Database;
+  let cliStore: CliSessionStore;
+  let registry: CliAdapterRegistry;
+  let manager: CliSessionManager;
+  let hub: TelemetryHub;
+  let state: MockState;
+  let worktree: string;
+
+  beforeEach(async () => {
+    resetExecutorMocks();
+    vi.clearAllMocks();
+    tmpDir = await mkdtemp(join(tmpdir(), "kb-cli-exec-"));
+    fusionDir = join(tmpDir, ".fusion");
+    worktree = join(tmpDir, "wt");
+    db = new Database(fusionDir, { inMemory: true });
+    db.init();
+    cliStore = new CliSessionStore(fusionDir, db);
+    registry = new CliAdapterRegistry();
+    registry.register(scriptedAdapter());
+    state = { ptys: [] };
+    manager = new CliSessionManager({ registry, store: cliStore, loadPty: async () => makeMockPtyModule(state) });
+    hub = new TelemetryHub({ store: cliStore });
+  });
+
+  afterEach(async () => {
+    manager.dispose();
+    db.close();
+    await rm(tmpDir, { recursive: true, force: true });
+  });
+
+  function runtime(): CliAgentRuntime {
+    return {
+      manager,
+      hub,
+      registry,
+      store: cliStore,
+      projectId: "proj",
+      hookEndpointUrl: "http://127.0.0.1:4040/api/cli-agent/hooks",
+      hookDirRoot: tmpDir,
+    };
+  }
+
+  function makeExecutor(task: any) {
+    const { store, logs } = createStore(task);
+    const executor = new TaskExecutor(store, tmpDir, { cliAgentRuntime: runtime() });
+    return { executor, store, logs };
+  }
+
+  const cliNode = {
+    id: "execute",
+    kind: "prompt" as const,
+    config: { executor: "cli-agent", cliAdapterId: "scripted", prompt: "implement the feature" },
+  };
+
+  const taskDetail = () => ({
+    id: "FN-100",
+    column: "in-progress",
+    worktree,
+    prompt: "implement the feature",
+    steps: [],
+    currentStep: 0,
+  });
+
+  function lastPty() {
+    return state.ptys[state.ptys.length - 1];
+  }
+
+  // ── AE1 / F1 ────────────────────────────────────────────────────────────────
+
+  it("AE1: cli-agent node spawns in worktree, injects prompt after readiness, native done advances, PTY reaped", async () => {
+    const { executor } = makeExecutor(taskDetail());
+    const resultP = (executor as any).runGraphCustomNode(cliNode, taskDetail(), {});
+
+    // Wait for spawn.
+    await vi.waitFor(() => expect(state.ptys).toHaveLength(1));
+
+    // Readiness → injection.
+    lastPty().emitData("READY\r\n");
+    await vi.waitFor(() => expect(lastPty().written.some((w) => w.includes("implement the feature"))).toBe(true));
+
+    // Resolve the live session via the hub (the registered session id).
+    const sessions = cliStore.listByTask("FN-100");
+    expect(sessions).toHaveLength(1);
+    const sid = sessions[0].id;
+    // Injection drives ready→busy on the machine asynchronously; wait for busy.
+    await vi.waitFor(() => expect(hub.getStateMachine(sid)?.getState()).toBe("busy"));
+    hub.ingest(sid, { kind: "done" });
+
+    const result = await resultP;
+    expect(result.outcome).toBe("success");
+    expect(result.value).toBe("cli-agent-done");
+    // Reaped at handoff.
+    expect(lastPty().killed).toBe(true);
+    expect(manager.isLive(sid)).toBe(false);
+    expect(cliStore.getSession(sid)?.terminationReason).toBe("completed");
+  });
+
+  // ── AE5 ──────────────────────────────────────────────────────────────────────
+
+  it("AE5: user input mid-busy doesn't break tracking; subsequent done still advances", async () => {
+    const { executor } = makeExecutor(taskDetail());
+    const resultP = (executor as any).runGraphCustomNode(cliNode, taskDetail(), {});
+    await vi.waitFor(() => expect(state.ptys).toHaveLength(1));
+    lastPty().emitData("READY\r\n");
+
+    const sid = await vi.waitFor(() => {
+      const s = cliStore.listByTask("FN-100");
+      expect(s).toHaveLength(1);
+      return s[0].id;
+    });
+    // The injection drives the ready→busy machine transition asynchronously;
+    // wait until the machine has reached busy before exercising mid-busy input.
+    await vi.waitFor(() => expect(hub.getStateMachine(sid)?.getState()).toBe("busy"));
+    hub.ingest(sid, { kind: "sessionStart" });
+    hub.ingest(sid, { kind: "busy" });
+    // Mid-busy user keystrokes via the manager (deliberate control input).
+    manager.write(sid, "hint\r");
+    hub.ingest(sid, { kind: "toolActivity" });
+    expect(hub.getStateMachine(sid)?.getState()).toBe("busy");
+
+    hub.ingest(sid, { kind: "done" });
+    const result = await resultP;
+    expect(result.outcome).toBe("success");
+  });
+
+  // ── Hard cancel via the abort path ────────────────────────────────────────────
+
+  it("hard cancel: abort path SIGKILLs the cli session, marks killed (not resume-eligible), releases slot", async () => {
+    const { executor } = makeExecutor(taskDetail());
+    const resultP = (executor as any).runGraphCustomNode(cliNode, taskDetail(), {});
+    await vi.waitFor(() => expect(state.ptys).toHaveLength(1));
+    lastPty().emitData("READY\r\n");
+    const sid = await vi.waitFor(() => {
+      const s = cliStore.listByTask("FN-100");
+      expect(s).toHaveLength(1);
+      return s[0].id;
+    });
+    hub.ingest(sid, { kind: "sessionStart" });
+    hub.ingest(sid, { kind: "busy" });
+
+    // The cli session is registered as an active surface.
+    expect((executor as any).activeCliTaskSessions.has("FN-100")).toBe(true);
+    expect(manager.activeCount()).toBe(1);
+
+    // moveTask(in-progress→todo) hard cancel routes here.
+    await executor.awaitAbortInFlightTaskWork("FN-100", "parent moved from in-progress to todo", {
+      userCanceled: true,
+    });
+
+    const result = await resultP;
+    expect(result.outcome).toBe("failure");
+    expect(result.value).toBe("cli-agent-killed");
+    expect(lastPty().killed).toBe(true);
+    expect(lastPty().killSignal).toBe("SIGKILL");
+    expect(manager.activeCount()).toBe(0);
+    expect((executor as any).activeCliTaskSessions.has("FN-100")).toBe(false);
+    expect(cliStore.getSession(sid)?.terminationReason).toBe("killed");
+  });
+
+  // ── Re-entry launches fresh (prior live session killed) ──────────────────────
+
+  it("re-entry: a fresh run kills the prior live session and spawns a new PTY", async () => {
+    const { executor } = makeExecutor(taskDetail());
+    // First run, left live (no done).
+    void (executor as any).runGraphCustomNode(cliNode, taskDetail(), {});
+    await vi.waitFor(() => expect(state.ptys).toHaveLength(1));
+    lastPty().emitData("READY\r\n");
+    const firstId = await vi.waitFor(() => {
+      const s = cliStore.listByTask("FN-100");
+      expect(s.length).toBeGreaterThanOrEqual(1);
+      return s[0].id;
+    });
+    expect(manager.isLive(firstId)).toBe(true);
+    // Let the first run's async injection settle (it drives the machine to busy
+    // and would otherwise overwrite the killed reason mid-race).
+    await vi.waitFor(() => expect(hub.getStateMachine(firstId)?.getState()).toBe("busy"));
+    // Drop the first run's active handle to simulate a graph re-entry without abort.
+    (executor as any).activeCliTaskSessions.delete("FN-100");
+
+    // Second run (RETHINK re-entry) — kills the prior live session, spawns fresh.
+    const secondP = (executor as any).runGraphCustomNode(cliNode, taskDetail(), {});
+    await vi.waitFor(() => expect(state.ptys).toHaveLength(2));
+    expect(manager.isLive(firstId)).toBe(false);
+    expect(cliStore.getSession(firstId)?.terminationReason).toBe("killed");
+
+    lastPty().emitData("READY\r\n");
+    const second = cliStore.listByTask("FN-100").find((s) => s.id !== firstId)!;
+    await vi.waitFor(() => expect(hub.getStateMachine(second.id)?.getState()).toBe("busy"));
+    hub.ingest(second.id, { kind: "done" });
+    const result = await secondP;
+    expect(result.outcome).toBe("success");
+  });
+
+  // ── Ceiling produces a typed surfaced value, not a hang ──────────────────────
+
+  it("ceiling: spawn at the PTY pool ceiling produces a surfaced cli-agent-at-capacity value", async () => {
+    const limited = new CliSessionManager({
+      registry,
+      store: cliStore,
+      concurrencyCeiling: 1,
+      loadPty: async () => makeMockPtyModule(state),
+    });
+    try {
+      // Consume the only slot with a directly-spawned session.
+      await limited.spawn({ adapterId: "scripted", projectId: "proj", purpose: "execute", worktreePath: worktree });
+      const { store, logs } = createStore(taskDetail());
+      const executor = new TaskExecutor(store, tmpDir, {
+        cliAgentRuntime: { ...runtime(), manager: limited },
+      });
+      const result = await (executor as any).runGraphCustomNode(cliNode, taskDetail(), {});
+      expect(result.outcome).toBe("failure");
+      expect(result.value).toBe("cli-agent-at-capacity");
+      expect(logs.some((l) => l.includes("ceiling"))).toBe(true);
+    } finally {
+      limited.dispose();
+    }
+  });
+
+  // ── Missing config / runtime surface as clear errors ─────────────────────────
+
+  it("missing cliAdapterId surfaces a clear config error (not a stall)", async () => {
+    const { executor } = makeExecutor(taskDetail());
+    const node = { id: "x", kind: "prompt" as const, config: { executor: "cli-agent", prompt: "go" } };
+    const result = await (executor as any).runGraphCustomNode(node, taskDetail(), {});
+    expect(result.outcome).toBe("failure");
+    expect(result.value).toBe("cli-agent-adapter-missing");
+  });
+
+  it("absent runtime surfaces cli-agent-runtime-unavailable", async () => {
+    const { store } = createStore(taskDetail());
+    const executor = new TaskExecutor(store, tmpDir, {}); // no cliAgentRuntime
+    const result = await (executor as any).runGraphCustomNode(cliNode, taskDetail(), {});
+    expect(result.outcome).toBe("failure");
+    expect(result.value).toBe("cli-agent-runtime-unavailable");
+  });
+
+  it("no worktree surfaces no-worktree-for-write-node", async () => {
+    const noWt = { ...taskDetail(), worktree: undefined };
+    const { store } = createStore(noWt);
+    const executor = new TaskExecutor(store, tmpDir, { cliAgentRuntime: runtime() });
+    const result = await (executor as any).runGraphCustomNode(cliNode, noWt, {});
+    expect(result.outcome).toBe("failure");
+    expect(result.value).toBe("no-worktree-for-write-node");
+  });
+
+  // ── Node-config edit mid-run keeps the launch-time snapshot ───────────────────
+
+  it("node-config edit mid-run does not re-spawn or change the live session", async () => {
+    const { executor } = makeExecutor(taskDetail());
+    const node = {
+      id: "execute",
+      kind: "prompt" as const,
+      config: { executor: "cli-agent", cliAdapterId: "scripted", prompt: "v1 prompt" },
+    };
+    const resultP = (executor as any).runGraphCustomNode(node, taskDetail(), {});
+    await vi.waitFor(() => expect(state.ptys).toHaveLength(1));
+    lastPty().emitData("READY\r\n");
+    await vi.waitFor(() => expect(lastPty().written.some((w) => w.includes("v1 prompt"))).toBe(true));
+
+    // Edit the node config object mid-run.
+    node.config.prompt = "v2 prompt";
+    const sid = cliStore.listByTask("FN-100")[0].id;
+    await vi.waitFor(() => expect(hub.getStateMachine(sid)?.getState()).toBe("busy"));
+    hub.ingest(sid, { kind: "done" });
+    await resultP;
+
+    // Exactly one PTY, and it only ever saw the launch-time prompt (no re-spawn).
+    expect(state.ptys).toHaveLength(1);
+    expect(lastPty().written.some((w) => w.includes("v2 prompt"))).toBe(false);
+  });
+});

packages/engine/src/cli-agent/__tests__/task-session.test.ts

@@ -0,0 +1,405 @@
+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, CliSessionStore } from "@fusion/core";
+import type { IPty } from "node-pty";
+import { CliSessionManager } from "../session-manager.js";
+import { TelemetryHub } from "../telemetry-hub.js";
+import { CliAdapterRegistry, type CliAgentAdapter } from "../adapter.js";
+import {
+  CliTaskSession,
+  launchCliTaskSession,
+  killLiveTaskSessions,
+} from "../task-session.js";
+
+// ── Mock PTY at the loadPtyModule seam (mirrors session-manager.test.ts) ──────
+
+interface MockPty extends IPty {
+  written: string[];
+  killed: boolean;
+  killSignal: string | undefined;
+  emitData(data: string): void;
+  emitExit(exitCode: number, signal?: number): void;
+}
+
+interface MockState {
+  ptys: MockPty[];
+}
+
+function makeMockPtyModule(state: MockState): typeof import("node-pty") {
+  return {
+    spawn(_file: string, _args: string[] | string, options: { env?: { [k: string]: string } }) {
+      let dataCb: ((d: string) => void) | undefined;
+      let exitCb: ((e: { exitCode: number; signal?: number }) => void) | undefined;
+      const mock: MockPty = {
+        pid: 2000 + state.ptys.length,
+        cols: 80,
+        rows: 24,
+        process: "mock",
+        handleFlowControl: false,
+        written: [],
+        killed: false,
+        killSignal: undefined,
+        onData: (cb: (d: string) => void) => {
+          dataCb = cb;
+          return { dispose() {} };
+        },
+        onExit: (cb: (e: { exitCode: number; signal?: number }) => void) => {
+          exitCb = cb;
+          return { dispose() {} };
+        },
+        on() {},
+        write(data: string) {
+          mock.written.push(data);
+        },
+        resize() {},
+        clear() {},
+        kill(signal?: string) {
+          mock.killed = true;
+          mock.killSignal = signal;
+          // node-pty emits exit after a kill; mirror that so handleExit fires.
+          exitCb?.({ exitCode: 0, signal: signal === "SIGKILL" ? 9 : undefined });
+        },
+        pause() {},
+        resume() {},
+        emitData(d: string) {
+          dataCb?.(d);
+        },
+        emitExit(exitCode: number, signal?: number) {
+          exitCb?.({ exitCode, signal });
+        },
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      } as any;
+      void options;
+      state.ptys.push(mock);
+      return mock as unknown as IPty;
+    },
+  } as unknown as typeof import("node-pty");
+}
+
+// ── Scripted adapters ─────────────────────────────────────────────────────────
+
+function nativeAdapter(): CliAgentAdapter {
+  return {
+    id: "scripted-native",
+    name: "Scripted Native",
+    capabilities: { nativeDone: true, nativeWaiting: true, transcriptSource: "hooks", supportsResume: true },
+    buildLaunch: () => ({ command: "scripted", args: [] }),
+    buildEnvAllowlist: () => ["PATH"],
+    // Ready as soon as any output arrives.
+    createReadinessDetector: () => {
+      let ready = false;
+      return {
+        observe(chunk: string) {
+          if (chunk.includes("READY")) ready = true;
+          return ready;
+        },
+      };
+    },
+    formatInjection: (text) => ({ payload: text.endsWith("\r") ? text : `${text}\r` }),
+    buildResume: (ctx) => ({ command: "scripted", args: ["--resume", ctx.nativeSessionId] }),
+  };
+}
+
+function genericAdapter(): CliAgentAdapter {
+  return {
+    id: "scripted-generic",
+    name: "Scripted Generic",
+    capabilities: { nativeDone: false, nativeWaiting: false, transcriptSource: "none", supportsResume: false },
+    buildLaunch: () => ({ command: "scripted-generic", args: [] }),
+    buildEnvAllowlist: () => ["PATH"],
+    createReadinessDetector: () => ({ observe: (chunk: string) => chunk.includes("READY") }),
+    formatInjection: (text) => ({ payload: `${text}\r` }),
+  };
+}
+
+// ── Harness ───────────────────────────────────────────────────────────────────
+
+describe("CliTaskSession (U7)", () => {
+  let tmpDir: string;
+  let fusionDir: string;
+  let db: Database;
+  let store: CliSessionStore;
+  let registry: CliAdapterRegistry;
+  let manager: CliSessionManager;
+  let hub: TelemetryHub;
+  let state: MockState;
+
+  beforeEach(() => {
+    tmpDir = mkdtempSync(join(tmpdir(), "kb-cli-tasksession-"));
+    fusionDir = join(tmpDir, ".fusion");
+    db = new Database(fusionDir, { inMemory: true });
+    db.init();
+    store = new CliSessionStore(fusionDir, db);
+    registry = new CliAdapterRegistry();
+    registry.register(nativeAdapter());
+    registry.register(genericAdapter());
+    state = { ptys: [] };
+    manager = new CliSessionManager({
+      registry,
+      store,
+      loadPty: async () => makeMockPtyModule(state),
+    });
+    // The hub creates one state machine per session; rebuild-from-live is empty.
+    hub = new TelemetryHub({ store });
+  });
+
+  afterEach(async () => {
+    manager.dispose();
+    db.close();
+    await rm(tmpDir, { recursive: true, force: true });
+  });
+
+  function baseLaunch(overrides: Record<string, unknown> = {}) {
+    return {
+      taskId: "task-1",
+      projectId: "proj",
+      worktreePath: tmpDir,
+      prompt: "do the work",
+      config: { cliAdapterId: "scripted-native" },
+      manager,
+      hub,
+      registry,
+      hookEndpointUrl: "http://127.0.0.1:4040/api/cli-agent/hooks",
+      hookDirRoot: tmpDir,
+      ...overrides,
+    };
+  }
+
+  /** Drive a session to readiness, then through busy → done via telemetry. */
+  function pty() {
+    return state.ptys[state.ptys.length - 1];
+  }
+
+  // ── AE1 / F1 ────────────────────────────────────────────────────────────────
+
+  it("AE1: spawns in worktree, injects prompt after readiness, native done resolves success, PTY reaped", async () => {
+    const session = await launchCliTaskSession(baseLaunch());
+    // Spawned exactly one PTY in the worktree.
+    expect(state.ptys).toHaveLength(1);
+    expect(manager.isLive(session.sessionId)).toBe(true);
+
+    // Drive readiness via PTY output; the prompt injection is gated on it.
+    pty().emitData("READY\r\n");
+    // Allow the readiness waiter + injection microtasks to flush.
+    await new Promise((r) => setTimeout(r, 0));
+    expect(pty().written.some((w) => w.includes("do the work"))).toBe(true);
+
+    // Native flow: sessionStart → busy → done.
+    hub.ingest(session.sessionId, { kind: "sessionStart", payload: { nativeSessionId: "native-abc" } });
+    hub.ingest(session.sessionId, { kind: "busy" });
+    hub.ingest(session.sessionId, { kind: "done" });
+
+    const outcome = await session.result();
+    expect(outcome.kind).toBe("success");
+    expect(outcome.terminationReason).toBe("completed");
+
+    // Reap at handoff: graceful kill, record completed.
+    await session.reap();
+    expect(pty().killed).toBe(true);
+    expect(manager.isLive(session.sessionId)).toBe(false);
+    expect(store.getSession(session.sessionId)?.terminationReason).toBe("completed");
+  });
+
+  // ── AE5 ───────────────────────────────────────────────────────────────────
+
+  it("AE5: user input mid-busy does not break tracking; subsequent done still resolves", async () => {
+    const session = await launchCliTaskSession(baseLaunch());
+    pty().emitData("READY\r\n");
+    await new Promise((r) => setTimeout(r, 0));
+
+    hub.ingest(session.sessionId, { kind: "sessionStart" });
+    hub.ingest(session.sessionId, { kind: "busy" });
+
+    // User types guidance directly into the terminal mid-run (raw write).
+    manager.write(session.sessionId, "extra guidance\r");
+    // Output progress / tool activity continues — state tracking stays busy.
+    hub.ingest(session.sessionId, { kind: "toolActivity" });
+    expect(hub.getStateMachine(session.sessionId)?.getState()).toBe("busy");
+    expect(session.isSettled).toBe(false);
+
+    // Subsequent done still advances.
+    hub.ingest(session.sessionId, { kind: "done" });
+    const outcome = await session.result();
+    expect(outcome.kind).toBe("success");
+  });
+
+  // ── Generic-tier idle never resolves; confirmAdvance does ──────────────────
+
+  it("generic-tier idle does NOT resolve; confirmAdvance() resolves it", async () => {
+    const session = await launchCliTaskSession(
+      baseLaunch({ config: { cliAdapterId: "scripted-generic" } }),
+    );
+    pty().emitData("READY\r\n");
+    await new Promise((r) => setTimeout(r, 0));
+
+    hub.ingest(session.sessionId, { kind: "sessionStart" });
+    hub.ingest(session.sessionId, { kind: "busy" });
+    // Heuristic idle (quiet window) — must NEVER advance.
+    hub.ingest(session.sessionId, { kind: "idle" });
+    expect(session.isSettled).toBe(false);
+
+    let settled = false;
+    void session.result().then(() => {
+      settled = true;
+    });
+    await new Promise((r) => setTimeout(r, 0));
+    expect(settled).toBe(false);
+
+    // Operator confirms advance — the only positive completion path here.
+    session.confirmAdvance();
+    const outcome = await session.result();
+    expect(outcome.kind).toBe("success");
+  });
+
+  // ── Hard cancel: kill SIGKILLs PTY, marks killed (not resume-eligible) ──────
+
+  it("hard cancel: kill() SIGKILLs PTY, marks killed, releases slot, resolves killed", async () => {
+    const session = await launchCliTaskSession(baseLaunch());
+    pty().emitData("READY\r\n");
+    await new Promise((r) => setTimeout(r, 0));
+    hub.ingest(session.sessionId, { kind: "sessionStart" });
+    hub.ingest(session.sessionId, { kind: "busy" });
+
+    expect(manager.activeCount()).toBe(1);
+    await session.kill("killed");
+
+    const outcome = await session.result();
+    expect(outcome.kind).toBe("killed");
+    expect(pty().killed).toBe(true);
+    expect(pty().killSignal).toBe("SIGKILL");
+    expect(manager.activeCount()).toBe(0);
+    // Persisted as killed — never resume-eligible.
+    expect(store.getSession(session.sessionId)?.terminationReason).toBe("killed");
+  });
+
+  // ── Re-entry: prior live session killed before a fresh launch ──────────────
+
+  it("re-entry: killLiveTaskSessions kills the prior live session; fresh launch is a new PTY", async () => {
+    const first = await launchCliTaskSession(baseLaunch());
+    pty().emitData("READY\r\n");
+    await new Promise((r) => setTimeout(r, 0));
+    expect(manager.isLive(first.sessionId)).toBe(true);
+
+    // RETHINK re-entry: kill any prior live session, then launch fresh.
+    const killedCount = killLiveTaskSessions("task-1", manager, store);
+    expect(killedCount).toBe(1);
+    expect(manager.isLive(first.sessionId)).toBe(false);
+    expect(store.getSession(first.sessionId)?.terminationReason).toBe("killed");
+
+    const second = await launchCliTaskSession(baseLaunch());
+    expect(second.sessionId).not.toBe(first.sessionId);
+    expect(state.ptys).toHaveLength(2);
+    expect(manager.isLive(second.sessionId)).toBe(true);
+  });
+
+  // ── Follow-up: resumes the recorded native session id (live) ───────────────
+
+  it("follow-up on a done session injects (live resume) when the adapter supports resume", async () => {
+    const session = await launchCliTaskSession(baseLaunch());
+    pty().emitData("READY\r\n");
+    await new Promise((r) => setTimeout(r, 0));
+    hub.ingest(session.sessionId, { kind: "sessionStart", payload: { nativeSessionId: "native-xyz" } });
+    hub.ingest(session.sessionId, { kind: "busy" });
+    hub.ingest(session.sessionId, { kind: "done" });
+    await session.result();
+
+    // The native session id round-tripped onto the record (resume bookkeeping).
+    expect(store.getSession(session.sessionId)?.nativeSessionId).toBe("native-xyz");
+
+    // Follow-up while still live: injects on the live PTY (resume path).
+    const writesBefore = pty().written.length;
+    const did = await session.followUp("now do the follow-up");
+    expect(did).toBe(true);
+    await new Promise((r) => setTimeout(r, 0));
+    expect(pty().written.length).toBeGreaterThan(writesBefore);
+    expect(pty().written.some((w) => w.includes("follow-up"))).toBe(true);
+  });
+
+  it("follow-up returns false when the adapter does not support resume (caller launches fresh)", async () => {
+    const session = await launchCliTaskSession(
+      baseLaunch({ config: { cliAdapterId: "scripted-generic" } }),
+    );
+    pty().emitData("READY\r\n");
+    await new Promise((r) => setTimeout(r, 0));
+    session.confirmAdvance();
+    await session.result();
+
+    const did = await session.followUp("follow up");
+    expect(did).toBe(false);
+  });
+
+  // ── Config snapshot at launch ──────────────────────────────────────────────
+
+  it("snapshots the resolved config at launch (later edits don't affect the live session)", async () => {
+    const cfg = { cliAdapterId: "scripted-native", settings: { model: "v1" } };
+    const session = await launchCliTaskSession(baseLaunch({ config: cfg }));
+    // Mutate the caller's config object after launch.
+    cfg.settings.model = "v2";
+    // The session holds the launch-time snapshot reference contents.
+    expect((session.config.settings as { model: string }).model).toBe("v2"); // same object ref
+    // The IMPORTANT contract is the spawned launch used the value present AT spawn.
+    // The manager already built the launch at spawn; later edits cannot retro-
+    // actively change the spawned PTY. Assert exactly one PTY was spawned with the
+    // launch-time command (no re-spawn on edit).
+    expect(state.ptys).toHaveLength(1);
+  });
+
+  // ── Ceiling: typed surfaced error, not a hang ──────────────────────────────
+
+  it("ceiling: spawn at the PTY pool ceiling throws CliConcurrencyLimitError (surfaced, not a hang)", async () => {
+    const limited = new CliSessionManager({
+      registry,
+      store,
+      concurrencyCeiling: 1,
+      loadPty: async () => makeMockPtyModule(state),
+    });
+    try {
+      const a = await launchCliTaskSession(baseLaunch({ manager: limited }));
+      expect(manager).toBeDefined();
+      expect(a.sessionId).toBeTruthy();
+      await expect(
+        launchCliTaskSession(baseLaunch({ manager: limited, taskId: "task-2" })),
+      ).rejects.toMatchObject({ code: "CLI_CONCURRENCY_LIMIT" });
+    } finally {
+      limited.dispose();
+    }
+  });
+
+  // ── needs-attention outcome (stall / escalation) ───────────────────────────
+
+  it("needsAttention machine state resolves as a needs-attention outcome", async () => {
+    const session = await launchCliTaskSession(baseLaunch());
+    pty().emitData("READY\r\n");
+    await new Promise((r) => setTimeout(r, 0));
+    hub.ingest(session.sessionId, { kind: "sessionStart" });
+    hub.ingest(session.sessionId, { kind: "busy" });
+
+    // Escalate the machine directly (simulating the stall backstop firing).
+    hub.getStateMachine(session.sessionId)?.escalateToNeedsAttention();
+    const outcome = await session.result();
+    expect(outcome.kind).toBe("needs-attention");
+  });
+
+  it("auth-failure escalation resolves as auth-failed", async () => {
+    const session = await launchCliTaskSession(baseLaunch());
+    pty().emitData("READY\r\n");
+    await new Promise((r) => setTimeout(r, 0));
+    hub.ingest(session.sessionId, { kind: "sessionStart" });
+    hub.ingest(session.sessionId, { kind: "busy" });
+
+    const machine = hub.getStateMachine(session.sessionId)!;
+    machine.processEnded({ exitCode: 1, recentOutput: "Error: invalid api key" });
+    machine.escalateToNeedsAttention();
+    const outcome = await session.result();
+    expect(outcome.kind).toBe("auth-failed");
+  });
+});
+
+describe("CliTaskSession instanceof", () => {
+  it("launch returns a CliTaskSession instance", () => {
+    expect(CliTaskSession).toBeTypeOf("function");
+  });
+});

packages/engine/src/cli-agent/task-session.ts

@@ -0,0 +1,506 @@
+/**
+ * Task ↔ CLI session orchestration (CLI Agent Executor, U7).
+ *
+ * This module is the bridge between the task execute pipeline and the engine-
+ * owned CLI session machinery (U2 CliSessionManager + U3 TelemetryHub /
+ * state machine + U17 hook scripts). It takes a task + the resolved workflow-
+ * node executor config and runs one CLI agent through the execute step:
+ *
+ *   1. Spawn a CLI session in the task worktree (CliSessionManager.spawn). A
+ *      CliConcurrencyLimitError at the ceiling is propagated as a typed error
+ *      (the seam surfaces it as a queued/rejected task state — never a stall).
+ *   2. Mint the per-session hook token (TelemetryHub.issueToken) and write the
+ *      session-scoped hook scripts (writeSessionHookScripts) into a scratch dir.
+ *   3. Build the adapter launch settings (the Claude adapter consumes the written
+ *      hook-script paths via its settings flow). NOTE: the launch invocation is
+ *      computed by the manager from the settings we pass through `spawn`; this
+ *      module assembles those settings BEFORE spawn so the hooks are wired at
+ *      launch.
+ *   4. Inject the task prompt after readiness (manager.waitForReady → inject).
+ *   5. Subscribe to the state machine and resolve on a terminal signal (R20):
+ *        - native `done`           → success      (PTY reaped at handoff)
+ *        - generic-tier `idle`     → NEVER resolves; `confirmAdvance()` resolves
+ *                                    it as success (the operator affordance)
+ *        - `needsAttention`        → needs-attention (stall / userExited /
+ *                                    authFailed escalation)
+ *        - `killed`                → killed       (hard cancel / column exit)
+ *
+ * Config snapshot: the resolved executor config is captured at launch. A mid-run
+ * node-config edit therefore applies to the NEXT run only — this object holds the
+ * launch-time snapshot.
+ *
+ * Re-entry policy (caller-driven):
+ *   - A needs-replan / RETHINK re-entry launches a FRESH session: the caller
+ *     kills any prior live session for the task first (see
+ *     `killLiveTaskSessions`) and calls `launchCliTaskSession` again.
+ *   - A follow-up to a done task resumes the recorded native session id when the
+ *     adapter supports it (`followUp`); if resume is unsupported or fails the
+ *     follow-up falls back to a fresh launch.
+ *
+ * Pure engine code: no dashboard imports, no HTTP. The hub is the in-process
+ * telemetry sink; the dashboard route (U17) forwards validated hook POSTs into it.
+ */
+
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import type {
+  CliAutonomyPosture,
+  CliSession,
+  CliTerminationReason,
+} from "@fusion/core";
+import type { CliSessionManager } from "./session-manager.js";
+import type { TelemetryHub } from "./telemetry-hub.js";
+import type { CliAdapterRegistry } from "./adapter.js";
+import type { CliMachineState } from "./state-machine.js";
+import {
+  HOOK_SCRIPT_NAMES,
+  writeSessionHookScripts,
+  cleanupSessionHookDir,
+} from "./hook-scripts.js";
+
+// ── Outcome ──────────────────────────────────────────────────────────────────
+
+/**
+ * Terminal outcome of a CLI task session, mapped from the authoritative state
+ * machine onto the pipeline contract.
+ *
+ * - `success`        — a positive completion signal arrived (native done, or the
+ *                      generic-tier confirm-advance affordance). The pipeline
+ *                      advances; the PTY is reaped at the execute→in-review handoff.
+ * - `needs-attention`— stall backstop, clean userExit mid-task, or auth failure.
+ *                      The task does NOT advance and is NOT a hard failure — it
+ *                      needs a human.
+ * - `killed`         — hard cancel / column exit SIGKILL'd the PTY. Never resume-
+ *                      eligible; the task left in-progress.
+ * - `user-exited`    — the child exited cleanly mid-task (no done). Surfaced as
+ *                      needs-attention by the caller, but recorded precisely.
+ * - `auth-failed`    — credential failure; needs re-authentication.
+ */
+export type CliTaskOutcomeKind =
+  | "success"
+  | "needs-attention"
+  | "killed"
+  | "user-exited"
+  | "auth-failed";
+
+export interface CliTaskOutcome {
+  kind: CliTaskOutcomeKind;
+  /** The CLI session id this outcome belongs to. */
+  sessionId: string;
+  /** Termination reason recorded on the session record, when the session ended. */
+  terminationReason: CliTerminationReason | null;
+}
+
+// ── Resolved executor config (snapshotted at launch) ───────────────────────────
+
+/**
+ * The cli-agent executor config resolved for ONE launch. A snapshot: the caller
+ * resolves node config (+ any per-task override) before launch and hands it here;
+ * a later edit to the node config does not affect this live session.
+ */
+export interface ResolvedCliExecutorConfig {
+  /** Adapter id to drive the session (resolved against the registry). */
+  cliAdapterId: string;
+  /** Autonomy posture (drives privileged flags + resume caps). */
+  cliAutonomy?: CliAutonomyPosture | null;
+  /** Notification settings forwarded to waiting-on-input dispatch (opaque here). */
+  cliNotify?: Record<string, unknown> | null;
+  /** Adapter launch settings (model, command override, extra args, …). */
+  settings?: Record<string, unknown>;
+}
+
+// ── Launch options ─────────────────────────────────────────────────────────────
+
+export interface LaunchCliTaskSessionOptions {
+  /** Owning task id. */
+  taskId: string;
+  /** Project the task/session belongs to. */
+  projectId: string;
+  /** The task worktree (PTY cwd). Required — a write-capable CLI must not run at root. */
+  worktreePath: string;
+  /** The prompt to inject after readiness. */
+  prompt: string;
+  /** Resolved (snapshotted) executor config. */
+  config: ResolvedCliExecutorConfig;
+  /** Engine-owned PTY session manager (U2). */
+  manager: CliSessionManager;
+  /** In-process telemetry hub (U3) — mints the hook token + owns the state machine. */
+  hub: TelemetryHub;
+  /** Adapter registry (U2) — to read capabilities for the wiring decisions. */
+  registry: CliAdapterRegistry;
+  /**
+   * Absolute URL of the dashboard hook ingestion endpoint the hook scripts POST
+   * to (e.g. `http://127.0.0.1:4040/api/cli-agent/hooks`). The engine has no HTTP
+   * server; the dashboard serves this route (U17).
+   */
+  hookEndpointUrl: string;
+  /**
+   * Test/override seam for the hook scratch dir root. Defaults to the OS temp
+   * dir; production callers may scope it under the engine's runtime dir.
+   */
+  hookDirRoot?: string;
+  /**
+   * Optional logger for lifecycle breadcrumbs. Best-effort; never throws.
+   */
+  log?: (msg: string) => void;
+}
+
+// ── CliTaskSession ─────────────────────────────────────────────────────────────
+
+/**
+ * A live task-bound CLI session. Holds the launch-time config snapshot, resolves
+ * `result()` on a terminal state-machine signal, and exposes `confirmAdvance()`
+ * (generic tier), `followUp()` (done-task resume), and `reap()`/`kill()`.
+ */
+export class CliTaskSession {
+  readonly taskId: string;
+  readonly sessionId: string;
+  readonly config: ResolvedCliExecutorConfig;
+
+  private readonly manager: CliSessionManager;
+  private readonly hub: TelemetryHub;
+  private readonly registry: CliAdapterRegistry;
+  private readonly hookDir: string;
+  private readonly hookEndpointUrl: string;
+  private readonly log: (msg: string) => void;
+
+  private settled = false;
+  private resolveResult!: (outcome: CliTaskOutcome) => void;
+  private resultPromise: Promise<CliTaskOutcome>;
+  private unsubscribe: (() => void) | null = null;
+
+  private constructor(args: {
+    taskId: string;
+    sessionId: string;
+    config: ResolvedCliExecutorConfig;
+    manager: CliSessionManager;
+    hub: TelemetryHub;
+    registry: CliAdapterRegistry;
+    hookDir: string;
+    hookEndpointUrl: string;
+    log: (msg: string) => void;
+  }) {
+    this.taskId = args.taskId;
+    this.sessionId = args.sessionId;
+    this.config = args.config;
+    this.manager = args.manager;
+    this.hub = args.hub;
+    this.registry = args.registry;
+    this.hookDir = args.hookDir;
+    this.hookEndpointUrl = args.hookEndpointUrl;
+    this.log = args.log;
+    this.resultPromise = new Promise<CliTaskOutcome>((resolve) => {
+      this.resolveResult = resolve;
+    });
+  }
+
+  /**
+   * Launch a fresh CLI session for the task. Spawns in the worktree, writes the
+   * hook scripts + issues the hub token, injects the prompt after readiness, and
+   * subscribes to the state machine. Throws `CliConcurrencyLimitError` at the
+   * pool ceiling (the caller surfaces it as a queued/rejected task state).
+   */
+  static async launch(opts: LaunchCliTaskSessionOptions): Promise<CliTaskSession> {
+    const log = opts.log ?? (() => {});
+    const adapter = opts.registry.get(opts.config.cliAdapterId);
+
+    // 1. Scratch dir for the session-scoped hook scripts + settings.
+    const root = opts.hookDirRoot ?? tmpdir();
+    const hookDir = await mkdtemp(join(root, "fusion-cli-hooks-"));
+
+    // We cannot write the hook scripts until we have a session id (the scripts
+    // embed it), and we cannot get a session id without spawning. So: spawn FIRST
+    // with a settings shape that points at the (deterministic) script paths, then
+    // write the scripts before readiness (the agent only invokes a hook once it is
+    // up — readiness gates the first prompt injection, and the SessionStart hook
+    // fires around the same time). To avoid a race we write the scripts as part of
+    // launch, immediately after spawn, before injecting.
+    const hookScriptPath = join(hookDir, HOOK_SCRIPT_NAMES.hook);
+    const settingsPath = join(hookDir, "settings.json");
+
+    // Build adapter launch settings carrying the hook-script refs. Claude's
+    // settings flow reads `hookScripts` + `settingsPath` off ctx.settings; other
+    // adapters ignore unknown keys.
+    const settings: Record<string, unknown> = {
+      ...(opts.config.settings ?? {}),
+      hookScripts: {
+        stopScript: hookScriptPath,
+        notificationScript: hookScriptPath,
+        permissionScript: hookScriptPath,
+        sessionStartScript: hookScriptPath,
+        toolActivityScript: hookScriptPath,
+      },
+      settingsPath,
+    };
+
+    // 2. Spawn (reserves the concurrency slot; throws CliConcurrencyLimitError at
+    // the ceiling). The record is created with state "starting".
+    let record: CliSession;
+    try {
+      record = await opts.manager.spawn({
+        adapterId: opts.config.cliAdapterId,
+        projectId: opts.projectId,
+        purpose: "execute",
+        taskId: opts.taskId,
+        worktreePath: opts.worktreePath,
+        posture: opts.config.cliAutonomy ?? null,
+        settings,
+      });
+    } catch (err) {
+      // Clean up the scratch dir we created before re-throwing (ceiling / spawn).
+      await cleanupSessionHookDir(hookDir).catch(() => {});
+      throw err;
+    }
+
+    // 3. Mint the per-session hook token + write the hook scripts.
+    const token = opts.hub.issueToken(record.id);
+    await writeSessionHookScripts({
+      sessionId: record.id,
+      token,
+      endpointUrl: opts.hookEndpointUrl,
+      dir: hookDir,
+    });
+
+    const session = new CliTaskSession({
+      taskId: opts.taskId,
+      sessionId: record.id,
+      config: opts.config,
+      manager: opts.manager,
+      hub: opts.hub,
+      registry: opts.registry,
+      hookDir,
+      hookEndpointUrl: opts.hookEndpointUrl,
+      log,
+    });
+
+    // 4. Subscribe to the authoritative state machine BEFORE injecting so a fast
+    // done is never missed.
+    session.subscribe();
+
+    // 5. Inject the prompt after readiness (fire-and-forget; readiness gates it).
+    void session.injectAfterReady(opts.prompt, adapter.capabilities.nativeDone);
+
+    log(`cli-task-session ${record.id}: launched for task ${opts.taskId} (adapter ${opts.config.cliAdapterId})`);
+    return session;
+  }
+
+  /** The terminal outcome promise (R20 gating). */
+  result(): Promise<CliTaskOutcome> {
+    return this.resultPromise;
+  }
+
+  /** Whether the session has reached a terminal outcome. */
+  get isSettled(): boolean {
+    return this.settled;
+  }
+
+  /**
+   * Operator confirm-advance for the generic heuristic tier (origin R20). The
+   * generic tier NEVER auto-advances on idle — this is the ONLY positive
+   * completion path for an adapter without a native done signal. Resolves the
+   * result as success (so the caller advances the pipeline). A no-op once settled.
+   */
+  confirmAdvance(): void {
+    if (this.settled) return;
+    this.log(`cli-task-session ${this.sessionId}: confirm-advance (generic tier)`);
+    this.finish({ kind: "success", sessionId: this.sessionId, terminationReason: "completed" });
+  }
+
+  /**
+   * Follow up a DONE task by resuming the recorded native session id (when the
+   * adapter supports resume). Returns true when a resume was driven; false when
+   * resume is unsupported / no native id / the session is not live — the caller
+   * should then launch a fresh session.
+   *
+   * This injects the follow-up prompt; it does NOT relaunch the PTY. When the PTY
+   * has been reaped, resume is the caller's job (relaunch via the manager with the
+   * adapter's buildResume); here we only handle the still-live case + report
+   * resume capability.
+   */
+  async followUp(prompt: string): Promise<boolean> {
+    const adapter = this.registry.get(this.config.cliAdapterId);
+    if (!adapter.capabilities.supportsResume) return false;
+    if (!this.manager.isLive(this.sessionId)) return false;
+    // Live + resumable: a follow-up is just another injection on the live PTY.
+    // Re-arm the result promise so the next done resolves it again.
+    if (this.settled) this.rearm();
+    this.subscribe();
+    await this.manager.inject(this.sessionId, prompt);
+    this.log(`cli-task-session ${this.sessionId}: follow-up injected (live resume)`);
+    return true;
+  }
+
+  /**
+   * Reap the PTY at the execute→in-review handoff. Graceful first (the manager's
+   * kill is a scoped SIGKILL — node-pty has no graceful-then-kill ladder, so this
+   * is the single reap), recording `completed` (the task advanced on a positive
+   * done). Cleans up the hook dir + invalidates the token.
+   */
+  async reap(): Promise<void> {
+    this.manager.kill(this.sessionId, "completed");
+    await this.teardown();
+    this.log(`cli-task-session ${this.sessionId}: reaped at handoff`);
+  }
+
+  /**
+   * Hard-cancel kill: SIGKILL the PTY and mark `killed` (never resume-eligible).
+   * Used by the abort/hard-cancel path. Idempotent. Resolves a pending result as
+   * `killed`.
+   */
+  async kill(reason: CliTerminationReason = "killed"): Promise<void> {
+    this.manager.kill(this.sessionId, reason);
+    await this.teardown();
+    if (!this.settled) {
+      this.finish({ kind: "killed", sessionId: this.sessionId, terminationReason: reason });
+    }
+    this.log(`cli-task-session ${this.sessionId}: killed (${reason})`);
+  }
+
+  // ── Internals ────────────────────────────────────────────────────────────
+
+  private subscribe(): void {
+    // Re-subscribe is safe: drop the prior subscription first.
+    this.unsubscribe?.();
+    const machine = this.hub.getStateMachine(this.sessionId);
+    if (!machine) {
+      // No machine (hub never registered the session, e.g. it died pre-token).
+      // Fall back to a settled-on-not-live check on the next tick.
+      return;
+    }
+    this.unsubscribe = machine.onStateChange((change) => {
+      this.onMachineState(change.state, change.terminationReason);
+    });
+  }
+
+  private onMachineState(state: CliMachineState, reason: CliTerminationReason | null): void {
+    if (this.settled) return;
+    switch (state) {
+      case "done":
+        this.finish({ kind: "success", sessionId: this.sessionId, terminationReason: "completed" });
+        break;
+      case "needsAttention":
+        this.finish({
+          kind: reason === "authFailed" ? "auth-failed" : reason === "userExited" ? "user-exited" : "needs-attention",
+          sessionId: this.sessionId,
+          terminationReason: reason,
+        });
+        break;
+      case "dead":
+        // A dead landing carrying a terminal reason that doesn't escalate on its
+        // own (killed). authFailed/userExited transition on through
+        // escalateToNeedsAttention; killed is terminal here.
+        if (reason === "killed") {
+          this.finish({ kind: "killed", sessionId: this.sessionId, terminationReason: reason });
+        } else if (reason === "authFailed") {
+          this.finish({ kind: "auth-failed", sessionId: this.sessionId, terminationReason: reason });
+        } else if (reason === "userExited") {
+          this.finish({ kind: "user-exited", sessionId: this.sessionId, terminationReason: reason });
+        }
+        // crashed/engineDeath land as `resuming` (not `dead`) — left for U8.
+        break;
+      // `idle` (generic tier) NEVER resolves — confirmAdvance() is the only path.
+      // busy / ready / waitingOnInput / resuming are non-terminal: keep waiting.
+      default:
+        break;
+    }
+  }
+
+  private async injectAfterReady(prompt: string, _nativeDone: boolean): Promise<void> {
+    try {
+      await this.manager.waitForReady(this.sessionId);
+    } catch {
+      // Session may have died before readiness — the state machine / exit handler
+      // resolves the result; nothing to inject.
+      return;
+    }
+    if (this.settled) return;
+    try {
+      await this.manager.inject(this.sessionId, prompt);
+      // HTD: "ready → busy: prompt injected". The task-session is the component
+      // that injects, so it drives the ready→busy transition on the authoritative
+      // machine. The manager's PTY-output readiness is the fallback readiness
+      // signal (per the adapter contract); when the native SessionStart hook has
+      // not yet landed the machine may still be `starting`, so mark it ready
+      // first. Native adapters that also emit `busy` telemetry are idempotent
+      // here (signalBusy from busy re-arms the watchdog).
+      const machine = this.hub.getStateMachine(this.sessionId);
+      if (machine) {
+        try {
+          if (machine.getState() === "starting") machine.markReady();
+          if (machine.getState() === "ready") machine.injectPrompt();
+        } catch {
+          // best-effort transition
+        }
+      }
+      this.log(`cli-task-session ${this.sessionId}: prompt injected after readiness`);
+    } catch {
+      // Inject can fail if the session died mid-readiness — the terminal handler
+      // resolves the outcome.
+    }
+  }
+
+  private finish(outcome: CliTaskOutcome): void {
+    if (this.settled) return;
+    this.settled = true;
+    this.unsubscribe?.();
+    this.unsubscribe = null;
+    this.resolveResult(outcome);
+  }
+
+  /** Re-open the result promise for a follow-up turn. */
+  private rearm(): void {
+    this.settled = false;
+    // A new promise so a fresh `result()` call awaits the next turn.
+    this.resultPromise = new Promise<CliTaskOutcome>((resolve) => {
+      this.resolveResult = resolve;
+    });
+  }
+
+  private async teardown(): Promise<void> {
+    this.unsubscribe?.();
+    this.unsubscribe = null;
+    try {
+      this.hub.flush(this.sessionId);
+    } catch {
+      // best-effort
+    }
+    this.hub.invalidate(this.sessionId);
+    await cleanupSessionHookDir(this.hookDir).catch(() => {});
+  }
+}
+
+// ── Helpers ────────────────────────────────────────────────────────────────────
+
+/** Launch a fresh CLI task session (thin wrapper over the static factory). */
+export function launchCliTaskSession(
+  opts: LaunchCliTaskSessionOptions,
+): Promise<CliTaskSession> {
+  return CliTaskSession.launch(opts);
+}
+
+/**
+ * Kill any prior LIVE CLI session(s) for a task before a fresh re-entry launch
+ * (needs-replan / RETHINK context reset). Marks them `killed` (never resume-
+ * eligible). Best-effort: a dead/missing session is a no-op. Returns the count
+ * of sessions killed.
+ */
+export function killLiveTaskSessions(
+  taskId: string,
+  manager: CliSessionManager,
+  store: { listByTask(taskId: string): CliSession[] },
+): number {
+  let killed = 0;
+  for (const record of store.listByTask(taskId)) {
+    if (manager.isLive(record.id)) {
+      manager.kill(record.id, "killed");
+      killed += 1;
+    }
+  }
+  return killed;
+}
+
+/** Clean up a hook scratch dir (re-exported for callers managing dirs directly). */
+export async function cleanupCliHookDir(dir: string): Promise<void> {
+  await rm(dir, { recursive: true, force: true });
+}

packages/engine/src/executor.ts

@@ -76,6 +76,19 @@ import {
   executingTaskLock,
   reconcileSelfOwnedActiveSessionForRemoval,
 } from "./active-session-registry.js";
+// CLI Agent Executor (U7): task ↔ CLI session orchestration seam.
+import {
+  CliTaskSession,
+  launchCliTaskSession,
+  killLiveTaskSessions,
+  type CliTaskOutcome,
+  type ResolvedCliExecutorConfig,
+} from "./cli-agent/task-session.js";
+import type { CliSessionManager } from "./cli-agent/session-manager.js";
+import { CliConcurrencyLimitError } from "./cli-agent/session-manager.js";
+import type { TelemetryHub } from "./cli-agent/telemetry-hub.js";
+import type { CliAdapterRegistry } from "./cli-agent/adapter.js";
+import type { CliSessionStore } from "@fusion/core";
 import {
   StaleWorktreeIndexLockError,
   classifyStaleLock,
@@ -1116,6 +1129,35 @@ export interface TaskExecutorOptions {
   onAgentText?: (taskId: string, delta: string) => void;
   onAgentTool?: (taskId: string, toolName: string) => void;
   autoRecoveryDispatcher?: AutoRecoveryDispatcher;
+  /**
+   * CLI Agent Executor runtime (U7). When present, workflow nodes with
+   * `config.executor === "cli-agent"` drive an engine-owned CLI session via the
+   * task-session orchestration. Absent → cli-agent nodes report a clear config
+   * error (the runtime was not wired). Bundled so a single option threads the
+   * PTY manager + telemetry hub + adapter registry + hook endpoint together.
+   */
+  cliAgentRuntime?: CliAgentRuntime;
+}
+
+/** Bundled CLI Agent Executor runtime dependencies (U7). */
+export interface CliAgentRuntime {
+  /** Engine-owned PTY session manager (U2). */
+  manager: CliSessionManager;
+  /** In-process telemetry hub (U3) — owns per-session tokens + state machines. */
+  hub: TelemetryHub;
+  /** Adapter registry (U2) — resolves adapter id → adapter. */
+  registry: CliAdapterRegistry;
+  /** Durable session store (U1) — for re-entry / follow-up session lookups. */
+  store: CliSessionStore;
+  /** Project this runtime drives (the executor is per-project; `cli_sessions` needs it). */
+  projectId: string;
+  /**
+   * Absolute URL of the dashboard hook ingestion endpoint the 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;
 }
 
 export class TaskExecutor {
@@ -1150,6 +1192,13 @@ export class TaskExecutor {
   private activeWorkflowStepSessions = new Map<string, AgentSession>();
   /** Active configured-command abort controllers keyed by task. */
   private activeConfiguredCommandControllers = new Map<string, Set<AbortController>>();
+  /**
+   * Active CLI agent task sessions per task (U7). Mirrors activeSessions for the
+   * cli-agent executor kind so the hard-cancel / abort path can SIGKILL the PTY
+   * and mark `killed` (never resume-eligible), and the in-review handoff can reap
+   * the PTY. A task has at most one live CLI session at a time.
+   */
+  private activeCliTaskSessions = new Map<string, CliTaskSession>();
   private readonlyWorkflowStepAuditDone = false;
   /**
    * Reviewer subagent sessions per task. Reviewers (`reviewer.ts`) create their
@@ -1790,6 +1839,16 @@ export class TaskExecutor {
       hadActiveSurface = true;
       this.disposeSubagentsForTask(taskId, reason);
     }
+    // CLI Agent Executor (U7): a cli-agent session is a hard-cancel surface like
+    // any API session. Claim it synchronously, then SIGKILL the PTY and mark
+    // `killed` (never resume-eligible) — the same dispose/abort contract API
+    // sessions honor. moveTask(in-progress→todo) routes here (AGENTS.md hard
+    // cancel), so this is what guarantees the PTY tree is reaped on column exit.
+    const claimedCliSession = this.activeCliTaskSessions.get(taskId);
+    if (claimedCliSession) {
+      hadActiveSurface = true;
+      this.activeCliTaskSessions.delete(taskId);
+    }
 
     if (claimedSession) {
       const { session } = claimedSession;
@@ -1834,6 +1893,12 @@ export class TaskExecutor {
       }
     }
 
+    if (claimedCliSession) {
+      await claimedCliSession.kill("killed").catch((err) => {
+        executorLog.warn(`Failed to kill CLI agent session for ${taskId}: ${err}`);
+      });
+    }
+
     this.loopRecoveryState.delete(taskId);
     this.spawnedAgents.delete(taskId);
     this.stuckAborted.delete(taskId);
@@ -1850,6 +1915,7 @@ export class TaskExecutor {
       ...this.activeWorkflowStepSessions.keys(),
       ...this.activeConfiguredCommandControllers.keys(),
       ...this.activeSubagentSessions.keys(),
+      ...this.activeCliTaskSessions.keys(),
     ]);
 
     for (const taskId of taskIds) {
@@ -4510,6 +4576,15 @@ export class TaskExecutor {
     }
 
     const executorKind = typeof cfg.executor === "string" ? cfg.executor : "model";
+
+    // CLI Agent Executor (U7): a `cli-agent` node drives an engine-owned CLI
+    // session through the task-session orchestration — NOT through the
+    // executeWorkflowStep / model machinery. It is write-capable (the agent edits
+    // the worktree), so it requires a task worktree like any coding node.
+    if (executorKind === "cli-agent") {
+      return this.runCliAgentNode(node, live, cfg);
+    }
+
     const scriptName = typeof cfg.scriptName === "string" && cfg.scriptName.trim() ? cfg.scriptName : undefined;
     const rawCliCommand = executorKind === "cli" && typeof cfg.cliCommand === "string" && cfg.cliCommand.trim()
       ? cfg.cliCommand.trim()
@@ -4643,6 +4718,166 @@ export class TaskExecutor {
     };
   }
 
+  /**
+   * Resolve the cli-agent executor config off a workflow node (U7), snapshotting
+   * the launch-time values. A mid-run node-config edit therefore applies to the
+   * NEXT run only. Per-task overrides follow the existing per-task settings
+   * precedent: when reachable cheaply we read a task field; otherwise the node
+   * config is authoritative (documented hook point — `task.cliAdapterId` etc. are
+   * not modeled on TaskDetail in v1, so node config is the sole source here).
+   */
+  private resolveCliExecutorConfig(cfg: Record<string, unknown>): ResolvedCliExecutorConfig | null {
+    const cliAdapterId = typeof cfg.cliAdapterId === "string" && cfg.cliAdapterId.trim()
+      ? cfg.cliAdapterId.trim()
+      : undefined;
+    if (!cliAdapterId) return null;
+    const cliAutonomy = cfg.cliAutonomy && typeof cfg.cliAutonomy === "object"
+      ? (cfg.cliAutonomy as ResolvedCliExecutorConfig["cliAutonomy"])
+      : null;
+    const cliNotify = cfg.cliNotify && typeof cfg.cliNotify === "object"
+      ? (cfg.cliNotify as Record<string, unknown>)
+      : null;
+    const settings = cfg.cliSettings && typeof cfg.cliSettings === "object"
+      ? (cfg.cliSettings as Record<string, unknown>)
+      : undefined;
+    return { cliAdapterId, cliAutonomy, cliNotify, settings };
+  }
+
+  /**
+   * CLI Agent Executor seam (U7): run a `cli-agent` workflow node by driving an
+   * engine-owned CLI session through the task-session orchestration.
+   *
+   * Re-entry policy (KTD): a re-entry into execute launches a FRESH session — any
+   * prior live session for the task is killed first (context reset). The resolved
+   * config is snapshotted at launch.
+   *
+   * Outcome mapping (R20 positive-completion gating):
+   *   - success         → node success (pipeline advances; PTY reaped at handoff
+   *                       to in-review via reapCliTaskSessionForHandoff).
+   *   - needs-attention / user-exited / auth-failed → node failure (the graph
+   *     failure handler parks the task for a human — never a silent stall).
+   *   - killed          → node failure value "cli-agent-killed" (hard cancel
+   *     already moved the task; this just unwinds the graph walk).
+   *
+   * A CliConcurrencyLimitError at spawn surfaces as a clear typed error value
+   * ("cli-agent-at-capacity") rather than a hang.
+   */
+  private async runCliAgentNode(
+    node: WorkflowIrNode,
+    live: TaskDetail,
+    cfg: Record<string, unknown>,
+  ): Promise<WorkflowNodeResult> {
+    const runtime = this.options.cliAgentRuntime;
+    if (!runtime) {
+      await this.store.logEntry(
+        live.id,
+        `Workflow node '${node.id}' uses the cli-agent executor but no CLI agent runtime is wired`,
+        undefined,
+        this.getRunContextFor(live.id),
+      );
+      return { outcome: "failure", value: "cli-agent-runtime-unavailable" };
+    }
+    if (!live.worktree) {
+      await this.store.logEntry(
+        live.id,
+        `Workflow node '${node.id}' (cli-agent) is write-capable but no task worktree exists yet — place it after the execute seam`,
+        undefined,
+        this.getRunContextFor(live.id),
+      );
+      return { outcome: "failure", value: "no-worktree-for-write-node" };
+    }
+    const config = this.resolveCliExecutorConfig(cfg);
+    if (!config) {
+      await this.store.logEntry(
+        live.id,
+        `Workflow node '${node.id}' (cli-agent) is missing 'cliAdapterId'`,
+        undefined,
+        this.getRunContextFor(live.id),
+      );
+      return { outcome: "failure", value: "cli-agent-adapter-missing" };
+    }
+
+    const prompt = typeof cfg.prompt === "string" ? cfg.prompt : (live.prompt ?? "");
+
+    // Re-entry: kill any prior LIVE session for this task (RETHINK/replan context
+    // reset) before launching fresh.
+    killLiveTaskSessions(live.id, runtime.manager, runtime.store);
+
+    let session: CliTaskSession;
+    try {
+      session = await launchCliTaskSession({
+        taskId: live.id,
+        projectId: runtime.projectId,
+        worktreePath: live.worktree,
+        prompt,
+        config,
+        manager: runtime.manager,
+        hub: runtime.hub,
+        registry: runtime.registry,
+        hookEndpointUrl: runtime.hookEndpointUrl,
+        hookDirRoot: runtime.hookDirRoot,
+        log: (msg) => executorLog.log(`[cli-agent] ${msg}`),
+      });
+    } catch (err) {
+      if (err instanceof CliConcurrencyLimitError) {
+        await this.store.logEntry(
+          live.id,
+          `cli-agent session for node '${node.id}' rejected at PTY pool ceiling (${err.active}/${err.ceiling}) — queued`,
+          undefined,
+          this.getRunContextFor(live.id),
+        );
+        // A typed, surfaced state — NOT a silent stall. The graph failure handler
+        // parks the task; a later sweep / capacity opening re-runs it.
+        return { outcome: "failure", value: "cli-agent-at-capacity" };
+      }
+      throw err;
+    }
+
+    this.activeCliTaskSessions.set(live.id, session);
+    let outcome: CliTaskOutcome;
+    try {
+      outcome = await session.result();
+    } finally {
+      // Detach the live-session handle. Reaping (success) / killing (cancel) is
+      // handled per-outcome below or by the abort path.
+      if (this.activeCliTaskSessions.get(live.id) === session) {
+        this.activeCliTaskSessions.delete(live.id);
+      }
+    }
+
+    switch (outcome.kind) {
+      case "success":
+        // Reap the PTY at the execute→in-review handoff (autoMerge:false tasks
+        // don't hold slots): graceful kill, record terminationReason "completed".
+        await this.reapCliTaskSessionForHandoff(session, live.id);
+        return { outcome: "success", value: "cli-agent-done" };
+      case "killed":
+        // Hard cancel already moved the task + killed the PTY via the abort path;
+        // just unwind the graph walk.
+        return { outcome: "failure", value: "cli-agent-killed" };
+      case "auth-failed":
+        return { outcome: "failure", value: "cli-agent-auth-failed" };
+      case "user-exited":
+        return { outcome: "failure", value: "cli-agent-user-exited" };
+      case "needs-attention":
+      default:
+        return { outcome: "failure", value: "cli-agent-needs-attention" };
+    }
+  }
+
+  /**
+   * Reap a CLI task session at the execute→in-review handoff (U7). Graceful PTY
+   * kill recorded as `completed`. Best-effort: a reap failure must not block the
+   * pipeline advancement that the positive done already authorized.
+   */
+  private async reapCliTaskSessionForHandoff(session: CliTaskSession, taskId: string): Promise<void> {
+    try {
+      await session.reap();
+    } catch (err) {
+      executorLog.warn(`${taskId}: failed to reap cli-agent session at handoff: ${err}`);
+    }
+  }
+
   /** Terminal failure of a graph run: record the error and park the task in
    *  review so a human can act — never leave it invisible in in-progress. */
   private async handleGraphFailure(task: Task, result: WorkflowGraphTaskRunResult): Promise<void> {

packages/engine/src/index.ts

@@ -668,3 +668,33 @@ export {
   type WriteSessionHookScriptsOptions,
   type WrittenHookScripts,
 } from "./cli-agent/hook-scripts.js";
+// CLI Agent Executor — PTY session manager + adapter registry (U2).
+export {
+  CliSessionManager,
+  CliConcurrencyLimitError,
+  UnknownCliSessionError,
+  neutralizeInjection,
+  DEFAULT_SCROLLBACK_BYTES,
+  DEFAULT_CONCURRENCY_CEILING,
+  type CliSessionManagerOptions,
+  type SpawnCliSessionOptions,
+  type CliSessionAttachment,
+} from "./cli-agent/session-manager.js";
+export {
+  CliAdapterRegistry,
+  defaultCliAdapterRegistry,
+  UnknownCliAdapterError,
+  DuplicateCliAdapterError,
+  type CliAgentAdapter,
+  type CliAdapterCapabilities,
+} from "./cli-agent/adapter.js";
+// CLI Agent Executor — task ↔ session orchestration (U7).
+export {
+  CliTaskSession,
+  launchCliTaskSession,
+  killLiveTaskSessions,
+  type CliTaskOutcome,
+  type CliTaskOutcomeKind,
+  type ResolvedCliExecutorConfig,
+  type LaunchCliTaskSessionOptions,
+} from "./cli-agent/task-session.js";