feat(engine): wire cli-agent executor seam and task session lifecycle (U7)

Add `cli-agent` as a task-execute executor kind. A workflow node with
`config.executor === "cli-agent"` drives an engine-owned CLI coding agent
through the execute step via the new cli-agent/task-session.ts orchestration:
spawn in the worktree, issue the hook token + write hook scripts, inject the
prompt after readiness, subscribe to the state machine, and resolve on a
positive completion signal (R20 gating). Config is snapshotted at launch; the
PTY is reaped (completed) at the in-review handoff.

Executor seam: runGraphCustomNode gains a cli-agent branch delegating to
runCliAgentNode; the hard-cancel/abort path (awaitAbortInFlightTaskWork +
abortAllInFlight) claims and SIGKILLs the CLI session as a first-class surface,
marking it killed (never resume-eligible). Re-entry kills any prior live
session and launches fresh; follow-up resumes the recorded native session id
when supported. A PTY-pool ceiling surfaces as a typed task value, not a stall.

Node-config typing extended minimally (WorkflowNodeExecutorKind /
WorkflowNodeExecutorConfig in @fusion/core).

Tests: cli-agent/__tests__/task-session.test.ts (12) and
__tests__/cli-agent-executor.test.ts (9) cover AE1/AE5, hard cancel, re-entry,
follow-up, config snapshot, ceiling, and the generic-tier confirm-advance path,
using scripted adapters + a mock PTY seam. Engine typecheck clean; full
src/cli-agent suite + new tests green (144 passing).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-06-05 00:01:12 -07:00
parent a08e144530
commit c1c99a9de0
8 changed files with 1646 additions and 0 deletions

View File

@@ -0,0 +1,27 @@
---
"@runfusion/fusion": minor
---
Wire the CLI Agent Executor as a selectable executor kind for the task execute
path (U7). A workflow node with `config.executor === "cli-agent"` (plus
`cliAdapterId` and optional `cliAutonomy`/`cliNotify`) now drives an engine-owned
CLI coding agent (Claude Code / Codex / Droid / Pi / generic) through the execute
step inside the task worktree.
The new `cli-agent/task-session.ts` orchestrates the task↔session lifecycle:
spawn in the worktree, mint the per-session hook token and write the hook scripts,
inject the task prompt after readiness, subscribe to the authoritative state
machine, and resolve on a positive completion signal (origin R20 gating — a
native `done` advances the pipeline; the generic tier never auto-advances on idle
and exposes a `confirmAdvance()` affordance instead). The resolved executor config
is snapshotted at launch, so a mid-run node-config edit applies to the next run
only. The PTY is reaped (recorded `completed`) at the execute→in-review handoff.
Lifecycle semantics honor the existing contracts: a hard cancel
(`moveTask(in-progress→todo)` / column-exit abort) SIGKILLs the CLI session via
the same dispose/abort path API sessions use and marks it `killed` (never
resume-eligible); a re-plan/RETHINK re-entry kills any prior live session and
launches fresh; a follow-up to a done task resumes the recorded native session id
when the adapter supports resume, else launches fresh. A PTY-pool ceiling
(`CliConcurrencyLimitError`) surfaces as a clear queued/rejected task state rather
than a silent stall.

View File

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

View File

@@ -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;

View File

@@ -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);
});
});

View File

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

View File

@@ -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 });
}

View File

@@ -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> {

View File

@@ -646,3 +646,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";