diff --git a/.changeset/cli-agent-executor-seam.md b/.changeset/cli-agent-executor-seam.md new file mode 100644 index 0000000000..98aed6d5cd --- /dev/null +++ b/.changeset/cli-agent-executor-seam.md @@ -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. diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 42df615f80..d5bf1b5134 100644 --- a/packages/core/src/index.ts +++ b/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"; diff --git a/packages/core/src/workflow-ir-types.ts b/packages/core/src/workflow-ir-types.ts index b96e538fb5..9a40aa199f 100644 --- a/packages/core/src/workflow-ir-types.ts +++ b/packages/core/src/workflow-ir-types.ts @@ -26,6 +26,50 @@ export interface WorkflowIrNode { config?: Record; } +/** + * 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; +} + export interface WorkflowIrEdge { from: string; to: string; diff --git a/packages/engine/src/__tests__/cli-agent-executor.test.ts b/packages/engine/src/__tests__/cli-agent-executor.test.ts new file mode 100644 index 0000000000..d3f9609a0d --- /dev/null +++ b/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>(); + const logs: string[] = []; + return { + logs, + store: { + on: vi.fn((event: string, listener: Listener) => { + const set = listeners.get(event) ?? new Set(); + 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); + }); +}); diff --git a/packages/engine/src/cli-agent/__tests__/task-session.test.ts b/packages/engine/src/cli-agent/__tests__/task-session.test.ts new file mode 100644 index 0000000000..8e8b12f1ae --- /dev/null +++ b/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 = {}) { + 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"); + }); +}); diff --git a/packages/engine/src/cli-agent/task-session.ts b/packages/engine/src/cli-agent/task-session.ts new file mode 100644 index 0000000000..9ad9f32477 --- /dev/null +++ b/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 | null; + /** Adapter launch settings (model, command override, extra args, …). */ + settings?: Record; +} + +// ── 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; + 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((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 { + 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 = { + ...(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 { + 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 { + 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 { + 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 { + 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 { + 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((resolve) => { + this.resolveResult = resolve; + }); + } + + private async teardown(): Promise { + 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 { + 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 { + await rm(dir, { recursive: true, force: true }); +} diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 3a29414187..38bfc7a04c 100644 --- a/packages/engine/src/executor.ts +++ b/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(); /** Active configured-command abort controllers keyed by task. */ private activeConfiguredCommandControllers = new Map>(); + /** + * 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(); 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): 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) + : null; + const settings = cfg.cliSettings && typeof cfg.cliSettings === "object" + ? (cfg.cliSettings as Record) + : 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, + ): Promise { + 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 { + 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 { diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index bca73df4bb..0eba1e94ff 100644 --- a/packages/engine/src/index.ts +++ b/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";