diff --git a/.changeset/cli-agent-one-shot-sessions.md b/.changeset/cli-agent-one-shot-sessions.md new file mode 100644 index 0000000000..6c8b466079 --- /dev/null +++ b/.changeset/cli-agent-one-shot-sessions.md @@ -0,0 +1,21 @@ +--- +"@runfusion/fusion": minor +--- + +Add CLI-agent one-shot sessions for the validator, planning, and CE plugin +surfaces (U9). A one-shot session runs an adapter's non-interactive invocation +(`claude -p`, `codex exec --json`, `droid exec --output-format json`, +`pi --print`) to completion in a working directory, streams output to a +read-only terminal (input disabled server-side via the durable +`autonomyPosture.readOnly` flag the transport's `isReadOnlySession` honors), +parses the adapter's structured JSON result, and reaps the PTY on exit. + +The new `cli-agent/one-shot-session.ts` returns a typed result: a success with +the parsed payload, or a typed failure (`nonzero-exit` / `unparseable` / +`spawn-failed`) carrying a bounded output tail. The validator integration +(`cli-agent-validator.ts`) maps results into the existing +pass/fail/blocked/error verdict contract — a malformed or unparseable result +maps to `error`, NEVER a silent pass. A planning seam (`runCliAgentPlanning`) +maps one-shot output into the same `PlanningResponse` shape a model run +produces, and the CE plugin's orchestrator threads an `executor` option +(`model` | `cli-agent`) end-to-end to its resolver. diff --git a/packages/dashboard/src/cli-session-transport.ts b/packages/dashboard/src/cli-session-transport.ts index 9c12b102ec..933422c7ec 100644 --- a/packages/dashboard/src/cli-session-transport.ts +++ b/packages/dashboard/src/cli-session-transport.ts @@ -213,8 +213,9 @@ export class CliConfirmAdvanceRegistry { * Whether a session is read-only (one-shot validator/planning, U9). Server-side * enforcement (not just client) — input/inject is rejected for these. * - * U9's one-shot sessions are not yet wired, so the current signal is the - * autonomy posture `readOnly` flag (forward-compatible) plus validator/planning + * U9's one-shot sessions (validator/planning/CE) are wired via the engine's + * `runOneShotSession`, which persists the autonomy posture `readOnly` flag on + * the session record. This check honors that flag plus validator/planning * purposes which are inherently read-only. */ export function isReadOnlySession(session: CliSession): boolean { diff --git a/packages/engine/src/__tests__/cli-agent-validator.test.ts b/packages/engine/src/__tests__/cli-agent-validator.test.ts new file mode 100644 index 0000000000..de2a4b1a8e --- /dev/null +++ b/packages/engine/src/__tests__/cli-agent-validator.test.ts @@ -0,0 +1,153 @@ +import { describe, it, expect } from "vitest"; +import { + mapParsedToVerdict, + oneShotResultToVerdict, + normalizeVerdictToken, + inferVerdictFromProse, + runCliAgentValidation, +} from "../cli-agent-validator.js"; +import type { + OneShotResult, + RunOneShotOptions, +} from "../cli-agent/one-shot-session.js"; + +function success(parsed: Record, text = ""): OneShotResult { + return { ok: true, sessionId: "s1", parsed, text, rawOutput: JSON.stringify(parsed) }; +} + +describe("verdict token normalization", () => { + it("maps synonyms to the contract set", () => { + expect(normalizeVerdictToken("APPROVE")).toBe("pass"); + expect(normalizeVerdictToken("passed")).toBe("pass"); + expect(normalizeVerdictToken("REVISE")).toBe("fail"); + expect(normalizeVerdictToken("blocked")).toBe("blocked"); + expect(normalizeVerdictToken("nonsense")).toBeNull(); + }); +}); + +describe("mapParsedToVerdict — per-adapter shapes → verdicts", () => { + it("claude-shaped pass (is_error:false + verdict)", () => { + const v = mapParsedToVerdict({ type: "result", verdict: "pass", is_error: false }, ""); + expect(v.status).toBe("pass"); + }); + + it("claude-shaped error flag is authoritative", () => { + const v = mapParsedToVerdict({ is_error: true, result: "crashed" }, ""); + expect(v.status).toBe("error"); + }); + + it("boolean passed:false → fail", () => { + const v = mapParsedToVerdict({ passed: false, summary: "missing X" }, ""); + expect(v.status).toBe("fail"); + expect(v.summary).toBe("missing X"); + }); + + it("explicit blocked flag → blocked with reason", () => { + const v = mapParsedToVerdict({ blocked: true, reason: "needs creds" }, ""); + expect(v.status).toBe("blocked"); + expect(v.blockedReason).toBe("needs creds"); + }); + + it("status token + assertions array", () => { + const v = mapParsedToVerdict( + { + status: "fail", + assertions: [ + { assertionId: "a1", passed: true }, + { id: "a2", passed: false, message: "nope" }, + ], + }, + "", + ); + expect(v.status).toBe("fail"); + expect(v.assertions).toHaveLength(2); + expect(v.assertions[1]).toEqual({ assertionId: "a2", passed: false, message: "nope" }); + }); + + it("prose-only pass inference", () => { + expect(inferVerdictFromProse("All assertions pass.")).toBe("pass"); + const v = mapParsedToVerdict({}, "All assertions pass."); + expect(v.status).toBe("pass"); + }); + + it("MALFORMED / undecidable → error, NEVER pass", () => { + const v = mapParsedToVerdict({ irrelevant: 1 }, "the agent rambled without a verdict"); + expect(v.status).toBe("error"); + }); +}); + +describe("oneShotResultToVerdict — failures map to error", () => { + it("nonzero exit → error with stderr in summary", () => { + const v = oneShotResultToVerdict({ + ok: false, + reason: "nonzero-exit", + sessionId: "s1", + exitCode: 1, + stderr: "segfault", + message: "exited with code 1", + }); + expect(v.status).toBe("error"); + expect(v.summary).toContain("segfault"); + }); + + it("unparseable → error (never silent pass)", () => { + const v = oneShotResultToVerdict({ + ok: false, + reason: "unparseable", + sessionId: "s1", + exitCode: 0, + stderr: "garbage", + message: "no decodable result", + }); + expect(v.status).toBe("error"); + }); + + it("success with pass verdict → pass", () => { + expect(oneShotResultToVerdict(success({ verdict: "pass" })).status).toBe("pass"); + }); +}); + +describe("runCliAgentValidation — seam threads purpose:validator and maps verdict", () => { + it("invokes runner with validator purpose and returns the verdict", async () => { + let seenPurpose: string | undefined; + const fakeRun = async (opts: RunOneShotOptions): Promise => { + seenPurpose = opts.purpose; + return success({ verdict: "pass", summary: "looks good" }, "looks good"); + }; + const verdict = await runCliAgentValidation( + { + manager: {} as RunOneShotOptions["manager"], + adapterId: "claude-code", + projectId: "p", + prompt: "validate", + cwd: "/tmp", + }, + fakeRun as never, + ); + expect(seenPurpose).toBe("validator"); + expect(verdict.status).toBe("pass"); + expect(verdict.summary).toBe("looks good"); + }); + + it("runner failure surfaces as error verdict", async () => { + const fakeRun = async (): Promise => ({ + ok: false, + reason: "spawn-failed", + sessionId: null, + exitCode: null, + stderr: "", + message: "ENOENT claude", + }); + const verdict = await runCliAgentValidation( + { + manager: {} as RunOneShotOptions["manager"], + adapterId: "claude-code", + projectId: "p", + prompt: "validate", + cwd: "/tmp", + }, + fakeRun as never, + ); + expect(verdict.status).toBe("error"); + }); +}); diff --git a/packages/engine/src/__tests__/interactive-ai-session.test.ts b/packages/engine/src/__tests__/interactive-ai-session.test.ts index af876cfba7..86cf97e3f5 100644 --- a/packages/engine/src/__tests__/interactive-ai-session.test.ts +++ b/packages/engine/src/__tests__/interactive-ai-session.test.ts @@ -2,9 +2,61 @@ import { describe, expect, it, vi } from "vitest"; import type { PlanningQuestion, PlanningResponse } from "@fusion/core"; import { createInteractiveAiSessionWith, + runCliAgentPlanning, type InteractiveAgentResult, type InteractiveAgentSession, } from "../interactive-ai-session.js"; +import type { + OneShotResult, + RunOneShotOptions, +} from "../cli-agent/one-shot-session.js"; + +describe("runCliAgentPlanning (U9 one-shot planning seam)", () => { + const baseOpts = { + manager: {} as RunOneShotOptions["manager"], + adapterId: "claude-code", + projectId: "p", + prompt: "plan it", + cwd: "/tmp", + }; + + it("maps one-shot output to the SAME PlanningResponse shape a model run produces", async () => { + let seenPurpose: string | undefined; + const fakeRun = async (opts: RunOneShotOptions): Promise => { + seenPurpose = opts.purpose; + const summary = { + title: "Do X", + description: "Plan to do X", + suggestedSize: "M", + suggestedDependencies: [], + keyDeliverables: ["X"], + }; + return { + ok: true, + sessionId: "s1", + parsed: {}, + text: JSON.stringify({ type: "complete", data: summary }), + rawOutput: "", + }; + }; + const resp: PlanningResponse = await runCliAgentPlanning(baseOpts, fakeRun as never); + expect(seenPurpose).toBe("planning"); + expect(resp.type).toBe("complete"); + if (resp.type === "complete") expect(resp.data.title).toBe("Do X"); + }); + + it("throws on a failed one-shot (never returns a fabricated plan)", async () => { + const fakeRun = async (): Promise => ({ + ok: false, + reason: "unparseable", + sessionId: "s1", + exitCode: 0, + stderr: "", + message: "no result", + }); + await expect(runCliAgentPlanning(baseOpts, fakeRun as never)).rejects.toThrow(/planning/i); + }); +}); /** * A scripted fake agent: each `prompt()` advances through a queue of canned diff --git a/packages/engine/src/cli-agent-validator.ts b/packages/engine/src/cli-agent-validator.ts new file mode 100644 index 0000000000..8f950f2dfb --- /dev/null +++ b/packages/engine/src/cli-agent-validator.ts @@ -0,0 +1,183 @@ +/** + * CLI-agent validator integration (CLI Agent Executor, U9). + * + * Bridges a one-shot CLI agent session into the existing validator verdict + * contract (`ValidationResult` in mission-execution-loop.ts: + * `status: "pass" | "fail" | "blocked" | "error"`). + * + * The cardinal rule: a malformed / unparseable / nonzero-exit one-shot maps to + * `error`, NEVER a silent `pass`. The verdict must be indistinguishable + * downstream from a model-executed validation run. + */ + +import type { + OneShotResult, + RunOneShotOptions, + runOneShotSession as RunOneShotFn, +} from "./cli-agent/one-shot-session.js"; + +/** The validator verdict contract shared with model-executed runs. */ +export interface ValidatorVerdict { + status: "pass" | "fail" | "blocked" | "error"; + /** Per-assertion results (empty when the adapter reports a bare verdict). */ + assertions: Array<{ assertionId: string; passed: boolean; message?: string }>; + summary: string; + blockedReason?: string; +} + +/** + * A structured verdict an adapter may emit in its one-shot JSON result. We look + * for these fields on the parsed payload (in priority order) before falling + * back to prose inference on the `text` field. + */ +interface ParsedVerdictShape { + verdict?: unknown; + status?: unknown; + result?: unknown; + passed?: unknown; + blocked?: unknown; + is_error?: unknown; + summary?: unknown; + reason?: unknown; + assertions?: unknown; +} + +/** Normalize a free-form verdict token to the contract's status set. */ +export function normalizeVerdictToken(token: string): ValidatorVerdict["status"] | null { + const t = token.trim().toLowerCase(); + if (["pass", "passed", "approve", "approved", "ok", "success"].includes(t)) return "pass"; + if (["fail", "failed", "revise", "reject", "rejected", "failure"].includes(t)) return "fail"; + if (["blocked", "block", "unavailable"].includes(t)) return "blocked"; + if (["error", "errored"].includes(t)) return "error"; + return null; +} + +/** + * Map a parsed one-shot result payload into a validator verdict. + * + * Precedence: + * 1. explicit `verdict`/`status` string token (normalized) + * 2. boolean `passed` (true→pass, false→fail) and `blocked === true` + * 3. `is_error === true` → error (claude-shaped) + * 4. prose inference from the result text + * 5. nothing decodable → error (NEVER pass) + */ +export function mapParsedToVerdict( + parsed: Record, + text: string, +): ValidatorVerdict { + const p = parsed as ParsedVerdictShape; + const summary = + (typeof p.summary === "string" && p.summary) || + (typeof p.reason === "string" && p.reason) || + text || + ""; + const assertions = parseAssertions(p.assertions); + + // 3 (early): an adapter error flag is authoritative. + if (p.is_error === true) { + return { status: "error", assertions, summary: summary || "Adapter reported an error" }; + } + + // 2: explicit blocked flag. + if (p.blocked === true) { + return { + status: "blocked", + assertions, + summary: summary || "Validation blocked", + blockedReason: typeof p.reason === "string" ? p.reason : summary || "blocked", + }; + } + + // 1: explicit verdict / status token. + for (const candidate of [p.verdict, p.status, p.result]) { + if (typeof candidate === "string") { + const status = normalizeVerdictToken(candidate); + if (status) { + return status === "blocked" + ? { status, assertions, summary: summary || "Validation blocked", blockedReason: summary } + : { status, assertions, summary }; + } + } + } + + // 2 (boolean): passed flag. + if (typeof p.passed === "boolean") { + return { status: p.passed ? "pass" : "fail", assertions, summary }; + } + + // 4: prose inference. + const inferred = inferVerdictFromProse(text); + if (inferred) return { status: inferred, assertions, summary: summary || text }; + + // 5: undecidable → error, never a silent pass. + return { + status: "error", + assertions, + summary: summary || "Validator produced no decodable verdict", + }; +} + +function parseAssertions(raw: unknown): ValidatorVerdict["assertions"] { + if (!Array.isArray(raw)) return []; + const out: ValidatorVerdict["assertions"] = []; + for (const item of raw) { + if (!item || typeof item !== "object") continue; + const a = item as Record; + const assertionId = + typeof a.assertionId === "string" + ? a.assertionId + : typeof a.id === "string" + ? a.id + : null; + if (!assertionId) continue; + out.push({ + assertionId, + passed: a.passed === true, + message: typeof a.message === "string" ? a.message : undefined, + }); + } + return out; +} + +/** Conservative prose inference. Only confident phrasings map to a verdict. */ +export function inferVerdictFromProse(text: string): ValidatorVerdict["status"] | null { + const t = text.toLowerCase(); + if (/\bblocked\b/.test(t)) return "blocked"; + if (/\b(revise|revision requested|does not (pass|meet)|fail(s|ed)?\b)/.test(t)) return "fail"; + if (/\b(all (assertions|checks) pass|validation pass(ed)?|approve(d)?)\b/.test(t)) return "pass"; + return null; +} + +/** + * Convert any one-shot result (success or failure) into a validator verdict. + * Failures (nonzero exit, unparseable, spawn-failed) → `error` with the bounded + * stderr folded into the summary. + */ +export function oneShotResultToVerdict(result: OneShotResult): ValidatorVerdict { + if (!result.ok) { + return { + status: "error", + assertions: [], + summary: `${result.message}${result.stderr ? `\n--- output tail ---\n${result.stderr}` : ""}`, + }; + } + return mapParsedToVerdict(result.parsed, result.text); +} + +/** + * Run a CLI-agent one-shot validation and map it to the verdict contract. + * + * This is the integration seam mission-execution-loop's `runValidation` branches + * to when the resolved validator executor is a CLI agent (adapter-backed) rather + * than a model. The `run` parameter is injected so this is unit-testable without + * a live PTY (tests pass a stubbed runner; production passes + * `runOneShotSession`). + */ +export async function runCliAgentValidation( + opts: Omit, + run: typeof RunOneShotFn, +): Promise { + const result = await run({ ...opts, purpose: "validator" }); + return oneShotResultToVerdict(result); +} diff --git a/packages/engine/src/cli-agent/__tests__/one-shot-session.test.ts b/packages/engine/src/cli-agent/__tests__/one-shot-session.test.ts new file mode 100644 index 0000000000..9577d7c149 --- /dev/null +++ b/packages/engine/src/cli-agent/__tests__/one-shot-session.test.ts @@ -0,0 +1,283 @@ +import { describe, it, expect, 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, type CliSession } from "@fusion/core"; +import type { IPty } from "node-pty"; +import { CliSessionManager } from "../session-manager.js"; +import { CliAdapterRegistry, type CliAgentAdapter } from "../adapter.js"; +import { + runOneShotSession, + parseOneShotOutput, + extractJsonObjects, + buildOneShotSettings, + boundedStderrTail, + ONE_SHOT_STDERR_CAP_BYTES, +} from "../one-shot-session.js"; + +/** + * Mirror of the dashboard transport's isReadOnlySession contract (asserted here + * without an engine→dashboard dependency): validator/planning are inherently + * read-only, and `autonomyPosture.readOnly === true` is an explicit flag. + */ +function isReadOnlySession(session: CliSession): boolean { + if (session.autonomyPosture && session.autonomyPosture.readOnly === true) return true; + return session.purpose === "validator" || session.purpose === "planning"; +} + +// ── Mock PTY at the loadPtyModule seam ───────────────────────────────────── + +interface MockPty extends IPty { + written: string[]; + killed: boolean; + 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, + spawnEnv: (options.env ?? {}) as { [k: string]: string }, + 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() { + mock.killed = true; + }, + pause() {}, + resume() {}, + emitData(d: string) { + dataCb?.(d); + }, + emitExit(exitCode: number, signal?: number) { + exitCb?.({ exitCode, signal }); + }, + } as unknown as MockPty; + state.ptys.push(mock); + return mock as unknown as IPty; + }, + } as unknown as typeof import("node-pty"); +} + +// ── Test adapter: one-shot forms exit immediately (no readiness gate). ─────── + +function makeAdapter(id: string): CliAgentAdapter { + return { + id, + name: `Test ${id}`, + capabilities: { + nativeDone: true, + nativeWaiting: false, + transcriptSource: "event-stream", + supportsResume: false, + }, + buildLaunch: (ctx) => ({ + command: id, + args: (ctx.settings.oneShotArgs as string[] | undefined) ?? [], + }), + buildEnvAllowlist: () => ["PATH"], + // One-shot output is non-interactive; readiness is immediately true so the + // generic injection fallback (if any) doesn't hang. + createReadinessDetector: () => ({ observe: () => true }), + formatInjection: (text) => ({ payload: `${text}\r` }), + }; +} + +interface Harness { + manager: CliSessionManager; + store: CliSessionStore; + state: MockState; + db: Database; + tmpDir: string; +} + +function makeHarness(adapterIds: string[]): Harness { + const tmpDir = mkdtempSync(join(tmpdir(), "kb-oneshot-test-")); + const fusionDir = join(tmpDir, ".fusion"); + const db = new Database(fusionDir, { inMemory: true }); + db.init(); + const store = new CliSessionStore(fusionDir, db); + const registry = new CliAdapterRegistry(); + for (const id of adapterIds) registry.register(makeAdapter(id)); + const state: MockState = { ptys: [] }; + const manager = new CliSessionManager({ + registry, + store, + loadPty: async () => makeMockPtyModule(state), + }); + return { manager, store, state, db, tmpDir }; +} + +/** + * Run a one-shot, then once the PTY exists drive its output + exit. Returns the + * resolved one-shot result. + */ +async function runWith( + h: Harness, + adapterId: string, + purpose: "validator" | "planning" | "ce", + output: string, + exitCode: number, +) { + const promise = runOneShotSession({ + manager: h.manager, + adapterId, + projectId: "proj-1", + purpose, + prompt: "do the thing", + cwd: h.tmpDir, + taskId: "FN-1", + }); + // Wait a tick for spawn + attach to settle, then drive the mock PTY. + await new Promise((r) => setTimeout(r, 5)); + const pty = h.state.ptys[h.state.ptys.length - 1]; + if (output) pty.emitData(output); + pty.emitExit(exitCode); + return promise; +} + +describe("one-shot session output parsing", () => { + it("buildOneShotSettings carries each adapter's documented non-interactive args", () => { + expect(buildOneShotSettings("claude-code", "P").oneShotArgs).toEqual(["-p", "P"]); + expect(buildOneShotSettings("codex", "P").oneShotArgs).toEqual(["exec", "--json", "P"]); + expect(buildOneShotSettings("droid", "P").oneShotArgs).toEqual([ + "exec", + "--output-format", + "json", + "P", + ]); + expect(buildOneShotSettings("pi", "P").oneShotArgs).toEqual(["--print", "P"]); + }); + + it("extractJsonObjects handles JSONL and embedded pretty JSON", () => { + expect(extractJsonObjects('{"a":1}\n{"b":2}\n')).toHaveLength(2); + expect(extractJsonObjects('banner\n{\n "x": 5\n}\ntrailer')).toEqual([{ x: 5 }]); + expect(extractJsonObjects("no json here")).toEqual([]); + }); + + it("parseOneShotOutput picks the claude result frame", () => { + const out = '{"type":"system"}\n{"type":"result","result":"done","is_error":false}'; + const parsed = parseOneShotOutput("claude-code", out); + expect(parsed?.text).toBe("done"); + expect(parsed?.parsed.type).toBe("result"); + }); + + it("boundedStderrTail caps very long output", () => { + const big = "x".repeat(ONE_SHOT_STDERR_CAP_BYTES + 100); + expect(Buffer.byteLength(boundedStderrTail(big))).toBeLessThanOrEqual( + ONE_SHOT_STDERR_CAP_BYTES, + ); + }); +}); + +describe("one-shot session lifecycle", () => { + let harnesses: Harness[] = []; + afterEach(async () => { + for (const h of harnesses) { + h.manager.dispose(); + h.db.close(); + await rm(h.tmpDir, { recursive: true, force: true }); + } + harnesses = []; + }); + function newHarness(ids: string[]): Harness { + const h = makeHarness(ids); + harnesses.push(h); + return h; + } + + it("creates a read-only session record, streams terminal output, reaps on completion", async () => { + const h = newHarness(["claude-code"]); + let captured: CliSession | null = null; + const result = await (async () => { + const promise = runOneShotSession({ + manager: h.manager, + adapterId: "claude-code", + projectId: "proj-1", + purpose: "validator", + prompt: "p", + cwd: h.tmpDir, + }); + await new Promise((r) => setTimeout(r, 5)); + const pty = h.state.ptys[0]; + // While live, the session record exists and is read-only, terminal streams. + const sessions = h.store.listSessions({ projectId: "proj-1" }); + captured = sessions[0] ?? null; + pty.emitData('{"type":"result","result":"ok","is_error":false}'); + pty.emitExit(0); + return promise; + })(); + + expect(captured).not.toBeNull(); + expect(isReadOnlySession(captured!)).toBe(true); + expect(result.ok).toBe(true); + // Reaped: no longer live. + expect(h.manager.isLive(captured!.id)).toBe(false); + const after = h.store.getSession(captured!.id); + expect(after?.agentState).toBe("dead"); + }); + + it("nonzero exit → failure with bounded stderr tail", async () => { + const h = newHarness(["codex"]); + const result = await runWith(h, "codex", "validator", "boom: fatal error\n", 1); + expect(result.ok).toBe(false); + if (!result.ok) { + expect(result.reason).toBe("nonzero-exit"); + expect(result.exitCode).toBe(1); + expect(result.stderr).toContain("boom: fatal error"); + } + }); + + it("unparseable output → typed unparseable failure (never silent success)", async () => { + const h = newHarness(["droid"]); + const result = await runWith(h, "droid", "validator", "not json at all", 0); + expect(result.ok).toBe(false); + if (!result.ok) expect(result.reason).toBe("unparseable"); + }); + + it("ce purpose is read-only via posture flag", async () => { + const h = newHarness(["pi"]); + const promise = runOneShotSession({ + manager: h.manager, + adapterId: "pi", + projectId: "proj-1", + purpose: "ce", + prompt: "p", + cwd: h.tmpDir, + }); + await new Promise((r) => setTimeout(r, 5)); + const session = h.store.listSessions({ projectId: "proj-1" })[0]; + expect(session.purpose).toBe("ce"); + expect(isReadOnlySession(session)).toBe(true); + const pty = h.state.ptys[0]; + pty.emitData('{"text":"hi"}'); + pty.emitExit(0); + await promise; + }); +}); diff --git a/packages/engine/src/cli-agent/one-shot-session.ts b/packages/engine/src/cli-agent/one-shot-session.ts new file mode 100644 index 0000000000..7d9c58270f --- /dev/null +++ b/packages/engine/src/cli-agent/one-shot-session.ts @@ -0,0 +1,378 @@ +/** + * One-shot CLI agent sessions (CLI Agent Executor, U9). + * + * A *one-shot* session runs an adapter's NON-INTERACTIVE invocation + * (`claude -p`, `codex exec --json`, `droid exec --output-format json`, + * `pi --print`) to completion in a working directory, streams its output to a + * read-only terminal (so U10's attach surface works exactly as for interactive + * sessions — but with input disabled server-side), collects the output, parses + * the adapter's structured (JSON) result, and returns a typed result. + * + * Read-only enforcement: the dashboard transport's `isReadOnlySession` treats + * `validator`/`planning` purposes as inherently read-only, and ALSO honors an + * `autonomyPosture.readOnly === true` flag. One-shot sessions persist that flag + * unconditionally so a `ce` (or any future) purpose is read-only too — the flag + * is the durable, transport-readable signal, not a transient client hint. + * + * Design notes: + * - The non-interactive command is built by `buildOneShotLaunch`, keyed off the + * adapter id. The interactive adapter launch builders intentionally don't + * model the `-p`/`exec` forms (those drive a REPL); one-shot is a distinct + * invocation that produces a single machine-readable result and exits. + * - Output is merged stdout+stderr (a PTY has a single stream). The structured + * result is parsed from that stream per adapter. On a nonzero exit or an + * unparseable result we return a typed failure carrying a BOUNDED tail of the + * output as `stderr` (the best available diagnostic on a PTY). + * - The PTY is reaped on completion: spawn → wait-for-exit → the session + * manager has already removed the live session and closed streams by the time + * `waitForExit` resolves. + */ + +import type { CliSessionPurpose } from "@fusion/core"; + +import type { CliSessionManager } from "./session-manager.js"; + +// ── Bounds ──────────────────────────────────────────────────────────────── + +/** Maximum bytes of output retained for diagnostics on a failed one-shot. */ +export const ONE_SHOT_STDERR_CAP_BYTES = 8 * 1024; + +// ── One-shot launch (non-interactive command builder) ─────────────────────── + +/** A non-interactive invocation: command + args (the prompt is passed inline). */ +export interface OneShotLaunchSpec { + /** Adapter-specific extra args appended after the adapter's one-shot base. */ + extraArgs?: readonly string[]; + /** Adapter-specific settings forwarded to the session manager's spawn. */ + settings: Record; +} + +/** + * Per-adapter one-shot settings. Each adapter's interactive `buildLaunch` + * consults `settings`; one-shot mode flags the non-interactive form there. + * + * The flag (`oneShot: true`) plus the documented per-adapter args are forwarded + * verbatim as launch settings. Adapters that don't yet branch on `oneShot` + * still receive the prompt via injection fallback (see runOneShotSession). + */ +export function buildOneShotSettings( + adapterId: string, + prompt: string, + base: Record = {}, +): Record { + const settings: Record = { ...base, oneShot: true, oneShotPrompt: prompt }; + // The non-interactive arg sets are documented in each adapter file. We carry + // them as explicit extraArgs so the session manager forwards them to spawn. + switch (adapterId) { + case "claude-code": + settings.oneShotArgs = ["-p", prompt]; + break; + case "codex": + settings.oneShotArgs = ["exec", "--json", prompt]; + break; + case "droid": + settings.oneShotArgs = ["exec", "--output-format", "json", prompt]; + break; + case "pi": + settings.oneShotArgs = ["--print", prompt]; + break; + default: + // generic / unknown: no structured form; prompt is injected. + settings.oneShotArgs = []; + break; + } + return settings; +} + +// ── Structured one-shot result ────────────────────────────────────────────── + +/** A successfully parsed one-shot result. */ +export interface OneShotSuccess { + ok: true; + /** The session record id (for terminal attach / audit). */ + sessionId: string; + /** Free-form structured payload parsed from the adapter's JSON output. */ + parsed: Record; + /** The text/result field the adapter surfaced (best-effort). */ + text: string; + /** Full captured output (bounded by scrollback). */ + rawOutput: string; +} + +/** Reason a one-shot failed. */ +export type OneShotFailureReason = "nonzero-exit" | "unparseable" | "spawn-failed"; + +/** A typed one-shot failure — NEVER mistaken for a pass downstream. */ +export interface OneShotFailure { + ok: false; + reason: OneShotFailureReason; + /** The session record id, when a session was created. */ + sessionId: string | null; + /** Process exit code, when the process ran. */ + exitCode: number | null; + /** Bounded diagnostic tail of the merged PTY output. */ + stderr: string; + /** Human-readable message. */ + message: string; +} + +export type OneShotResult = OneShotSuccess | OneShotFailure; + +// ── Per-adapter result parsing ────────────────────────────────────────────── + +/** + * Parse the structured result out of an adapter's one-shot output. Each adapter + * emits JSON on its non-interactive path; the PTY merges it with any banner + * lines, so we scan for the LAST decodable JSON object/array on the stream and + * normalize a `text` field per adapter shape. + * + * Returns null when no decodable structured result is present (→ unparseable). + */ +export function parseOneShotOutput( + adapterId: string, + output: string, +): { parsed: Record; text: string } | null { + const objects = extractJsonObjects(output); + if (objects.length === 0) return null; + + switch (adapterId) { + case "claude-code": { + // `claude -p --output-format json` (or stream-json) → a result object + // with `{ type: "result", result | text, is_error }`. Prefer the final + // result frame. + const result = + objects.find((o) => o.type === "result") ?? objects[objects.length - 1]; + const text = + pickString(result, ["result", "text", "content", "message"]) ?? ""; + return { parsed: result, text }; + } + case "codex": { + // `codex exec --json` emits a stream of JSON events; the final + // agent/assistant message carries the answer. + const last = objects[objects.length - 1]; + const text = pickString(last, ["text", "message", "content", "result"]) ?? ""; + return { parsed: last, text }; + } + case "droid": { + // `droid exec --output-format json` → a single result object. + const last = objects[objects.length - 1]; + const text = pickString(last, ["result", "text", "message", "output"]) ?? ""; + return { parsed: last, text }; + } + case "pi": { + const last = objects[objects.length - 1]; + const text = pickString(last, ["text", "result", "message", "content"]) ?? ""; + return { parsed: last, text }; + } + default: { + const last = objects[objects.length - 1]; + const text = pickString(last, ["text", "result", "message"]) ?? ""; + return { parsed: last, text }; + } + } +} + +/** Pull the first present string field from a parsed object. */ +function pickString(obj: Record, keys: string[]): string | null { + for (const k of keys) { + const v = obj[k]; + if (typeof v === "string" && v.length > 0) return v; + } + return null; +} + +/** + * Extract decodable top-level JSON objects from a (possibly noisy, possibly + * line-delimited) output stream. Handles both JSONL (one object per line) and a + * single pretty-printed object embedded in banner text. + */ +export function extractJsonObjects(output: string): Record[] { + const objects: Record[] = []; + // First try JSONL: each non-empty line that decodes to an object. + let sawLineJson = false; + for (const line of output.split(/\r?\n/)) { + const trimmed = line.trim(); + if (!trimmed || (trimmed[0] !== "{" && trimmed[0] !== "[")) continue; + try { + const v = JSON.parse(trimmed); + if (v && typeof v === "object") { + objects.push(v as Record); + sawLineJson = true; + } + } catch { + // not a standalone JSON line; fall through to brace scanning below. + } + } + if (sawLineJson) return objects; + + // Fallback: brace-balanced scan for a single embedded JSON object. + const start = output.indexOf("{"); + const end = output.lastIndexOf("}"); + if (start >= 0 && end > start) { + try { + const v = JSON.parse(output.slice(start, end + 1)); + if (v && typeof v === "object") objects.push(v as Record); + } catch { + // unparseable + } + } + return objects; +} + +// ── Runner ────────────────────────────────────────────────────────────────── + +export interface RunOneShotOptions { + manager: CliSessionManager; + adapterId: string; + projectId: string; + /** validator | planning | ce (chat/execute are interactive, not one-shot). */ + purpose: Extract; + prompt: string; + /** Working directory the CLI runs in (also the PTY cwd). */ + cwd: string; + taskId?: string | null; + chatSessionId?: string | null; + /** Extra adapter launch settings (model, profile…). Merged under one-shot. */ + settings?: Record; + /** Optional overall timeout (ms). On timeout the session is killed → failure. */ + timeoutMs?: number; +} + +/** + * Run an adapter's one-shot invocation to completion and return a typed result. + * + * Lifecycle: spawn (read-only record) → attach (collect output for the + * read-only terminal + parsing) → wait for exit → parse → reap. On nonzero exit + * or unparseable output, returns a typed {@link OneShotFailure} with a bounded + * output tail — NEVER a silent success. + */ +export async function runOneShotSession(opts: RunOneShotOptions): Promise { + const { + manager, + adapterId, + projectId, + purpose, + prompt, + cwd, + taskId = null, + chatSessionId = null, + timeoutMs, + } = opts; + + const settings = buildOneShotSettings(adapterId, prompt, opts.settings ?? {}); + + let sessionId: string | null = null; + try { + const record = await manager.spawn({ + adapterId, + projectId, + purpose, + taskId, + chatSessionId, + worktreePath: cwd, + // Durable, transport-readable read-only flag. validator/planning are + // already inherently read-only; this makes `ce` (and any future purpose) + // read-only too. See isReadOnlySession in the dashboard transport. + posture: { readOnly: true }, + settings, + }); + sessionId = record.id; + } catch (err) { + return { + ok: false, + reason: "spawn-failed", + sessionId, + exitCode: null, + stderr: "", + message: err instanceof Error ? err.message : String(err), + }; + } + + // Attach to collect output (also exercises the read-only terminal stream). + const attachment = manager.attach(sessionId); + const chunks: Buffer[] = []; + // Replay scrollback captured at attach (usually empty for a fresh spawn). + if (attachment.scrollback.byteLength > 0) { + chunks.push(Buffer.from(attachment.scrollback)); + } + const drainPromise = (async () => { + for await (const bytes of attachment.stream) { + chunks.push(Buffer.from(bytes)); + } + })(); + + // For adapters whose one-shot form is NOT carried in args (generic fallback), + // inject the prompt once ready. Adapters that consume the prompt via args + // (claude/codex/droid/pi) ignore this — they already have it on argv. + if (adapterId === "generic") { + try { + await manager.inject(sessionId, prompt); + } catch { + // session may exit immediately for non-interactive forms; ignore. + } + } + + let timer: NodeJS.Timeout | undefined; + let timedOut = false; + const exitPromise = manager.waitForExit(sessionId); + const exit = await (timeoutMs && timeoutMs > 0 + ? Promise.race([ + exitPromise, + new Promise<{ exitCode: number; signal: number | undefined }>((resolve) => { + timer = setTimeout(() => { + timedOut = true; + manager.kill(sessionId as string, "killed"); + resolve({ exitCode: -1, signal: 9 }); + }, timeoutMs); + }), + ]) + : exitPromise); + if (timer) clearTimeout(timer); + + // Ensure the stream is fully drained before reading output. + attachment.detach(); + await drainPromise.catch(() => undefined); + + const rawOutput = Buffer.concat(chunks).toString("utf8"); + const boundedTail = boundedStderrTail(rawOutput); + + if (exit.exitCode !== 0 || timedOut) { + return { + ok: false, + reason: "nonzero-exit", + sessionId, + exitCode: exit.exitCode, + stderr: boundedTail, + message: timedOut + ? `one-shot ${adapterId} session timed out after ${timeoutMs}ms` + : `one-shot ${adapterId} session exited with code ${exit.exitCode}`, + }; + } + + const parsedResult = parseOneShotOutput(adapterId, rawOutput); + if (!parsedResult) { + return { + ok: false, + reason: "unparseable", + sessionId, + exitCode: exit.exitCode, + stderr: boundedTail, + message: `one-shot ${adapterId} produced no decodable structured result`, + }; + } + + return { + ok: true, + sessionId, + parsed: parsedResult.parsed, + text: parsedResult.text, + rawOutput, + }; +} + +/** Bounded tail of merged output, for failure diagnostics. */ +export function boundedStderrTail(output: string): string { + const buf = Buffer.from(output, "utf8"); + if (buf.byteLength <= ONE_SHOT_STDERR_CAP_BYTES) return output; + return buf.subarray(buf.byteLength - ONE_SHOT_STDERR_CAP_BYTES).toString("utf8"); +} diff --git a/packages/engine/src/cli-agent/session-manager.ts b/packages/engine/src/cli-agent/session-manager.ts index ba3b50cd90..194835645a 100644 --- a/packages/engine/src/cli-agent/session-manager.ts +++ b/packages/engine/src/cli-agent/session-manager.ts @@ -320,6 +320,10 @@ interface LiveSession { terminated: boolean; /** Bytes buffered toward the high watermark since last drain to consumers. */ inflightBytes: number; + /** Captured exit result (set once on exit/kill), for one-shot waiters. */ + exitResult: { exitCode: number; signal: number | undefined } | null; + /** Resolvers waiting on process exit (one-shot sessions). */ + exitWaiters: ((result: { exitCode: number; signal: number | undefined }) => void)[]; } // ── Manager options ────────────────────────────────────────────────────────── @@ -456,6 +460,8 @@ export class CliSessionManager { paused: false, terminated: false, inflightBytes: 0, + exitResult: null, + exitWaiters: [], }; this.sessions.set(record.id, live); @@ -535,10 +541,19 @@ export class CliSessionManager { } } + /** Settle one-shot exit waiters exactly once with the captured result. */ + private settleExit(live: LiveSession, exitCode: number, signal: number | undefined): void { + if (live.exitResult) return; + live.exitResult = { exitCode, signal }; + const waiters = live.exitWaiters.splice(0); + for (const w of waiters) w(live.exitResult); + } + private handleExit(live: LiveSession, exitCode: number, signal?: number): void { if (live.terminated) return; live.terminated = true; this.sessions.delete(live.id); + this.settleExit(live, exitCode, signal); for (const stream of live.streams) stream.close(); live.streams.clear(); @@ -578,6 +593,19 @@ export class CliSessionManager { return new Promise((resolve) => live.readyWaiters.push(resolve)); } + /** + * Resolve once the session's PTY has exited (or been killed), yielding the + * captured exit result. Powers one-shot (validator/planning/CE) sessions that + * run a non-interactive invocation to completion. A killed session resolves + * with `{ exitCode: -1, signal: 9 }`. Throws if the session id is unknown AND + * not already exited within this manager's memory. + */ + waitForExit(sessionId: string): Promise<{ exitCode: number; signal: number | undefined }> { + const live = this.require(sessionId); + if (live.exitResult) return Promise.resolve(live.exitResult); + return new Promise((resolve) => live.exitWaiters.push(resolve)); + } + // ── Injection ────────────────────────────────────────────────────────── /** @@ -763,6 +791,8 @@ export class CliSessionManager { } live.terminated = true; this.sessions.delete(live.id); + // A killed PTY exited via signal — surface a nonzero result to one-shot waiters. + this.settleExit(live, -1, 9); for (const stream of live.streams) stream.close(); live.streams.clear(); diff --git a/packages/engine/src/interactive-ai-session.ts b/packages/engine/src/interactive-ai-session.ts index f175309b76..890c5b78da 100644 --- a/packages/engine/src/interactive-ai-session.ts +++ b/packages/engine/src/interactive-ai-session.ts @@ -187,6 +187,40 @@ export function parseAgentResponse(text: string): PlanningResponse { throw new Error("AI returned an invalid response structure."); } +/** + * CLI-agent planning one-shot seam (U9). + * + * Runs a CLI-agent one-shot in `planning` purpose and maps its output into the + * SAME `PlanningResponse` shape a model-backed planning run produces — so the + * downstream planning flow cannot tell a CLI-backed run from a model run. The + * one-shot runner is injected (`run`) so this is testable without a live PTY. + * + * NOTE (deviation, see report): the full planning *loop* (multi-turn + * question/answer over a resumable interactive session) is interactive, not + * one-shot — a one-shot planning run produces a single terminal response. This + * seam covers the single-shot "produce a plan" case and proves output-shape + * compatibility (`parseAgentResponse`). Wiring it into the resumable planning + * loop's executor resolution remains TODO when planning gains a CLI executor + * selector. + */ +export async function runCliAgentPlanning( + opts: Omit< + import("./cli-agent/one-shot-session.js").RunOneShotOptions, + "purpose" + >, + run: typeof import("./cli-agent/one-shot-session.js").runOneShotSession, +): Promise { + const result = await run({ ...opts, purpose: "planning" }); + if (!result.ok) { + throw new Error( + `CLI-agent planning one-shot failed (${result.reason}): ${result.message}`, + ); + } + // Map to the planning flow's shape exactly as a model run would: parse the + // adapter's textual result through the canonical planning parser. + return parseAgentResponse(result.text || result.rawOutput); +} + /** Extract text from the last assistant message (string | text blocks | thinking fallback). */ function extractLastAssistantText(session: InteractiveAgentSession): string { const lastMessage = session.state.messages.filter((m) => m.role === "assistant").pop(); diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-executor-seam.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-executor-seam.test.ts new file mode 100644 index 0000000000..fe5c7b98f2 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-executor-seam.test.ts @@ -0,0 +1,40 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { CeOrchestrator, type CeSessionExecutor } from "../session/orchestrator.js"; +import { makeHarness, type TestHarness } from "./_harness.js"; + +/** + * U9 CE executor seam contract. + * + * Proves the `executor` option threads end-to-end on a REAL orchestrator (not a + * scripted fake): the option set on deps is what `resolveExecutor()` returns, + * and it defaults to the model backend. The cli-agent one-shot wiring itself is + * engine-side (`@fusion/engine` runOneShotSession); this asserts the plugin + * carries the choice through, per the plugin-skills option-threading learning. + */ +let h: TestHarness; +beforeEach(() => { + h = makeHarness(); +}); +afterEach(() => { + h.close(); +}); + +function makeOrch(executor?: CeSessionExecutor) { + return new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: vi.fn(async () => ({ session: {} as never })), + projectRoot: h.projectRoot, + executor, + }); +} + +describe("CE executor seam (U9)", () => { + it("defaults to the model backend when no executor option is supplied", () => { + expect(makeOrch().resolveExecutor()).toEqual({ kind: "model" }); + }); + + it("threads a cli-agent executor selection through deps → resolver", () => { + const orch = makeOrch({ kind: "cli-agent", adapterId: "claude-code" }); + expect(orch.resolveExecutor()).toEqual({ kind: "cli-agent", adapterId: "claude-code" }); + }); +}); diff --git a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts index 7f943cbd02..05040c17ee 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts @@ -85,6 +85,20 @@ export class CeTurnTimeoutError extends Error { } } +/** + * Which session backend a CE stage runs on (U9 seam). + * + * - `model` (default): the model-backed interactive AI session (existing path). + * - `cli-agent`: a CLI agent adapter, run as a read-only one-shot per stage. + * + * The CE plugin threads this choice end-to-end (deps → resolver) per the + * plugin-skills option-threading learning; the cli-agent one-shot wiring itself + * lives engine-side in `@fusion/engine` (`runOneShotSession`). + */ +export type CeSessionExecutor = + | { kind: "model" } + | { kind: "cli-agent"; adapterId: string }; + export interface OrchestratorDeps { ctx: PluginContext; /** @@ -96,6 +110,19 @@ export interface OrchestratorDeps { projectRoot?: string; /** Override the per-turn timeout (ms). */ turnTimeoutMs?: number; + /** + * Session backend selector (U9). Defaults to `{ kind: "model" }`. When set to + * `{ kind: "cli-agent", adapterId }`, CE sessions select the CLI-agent + * one-shot executor. Threaded through to `resolveExecutor()` so callers can + * route a CE stage onto a CLI adapter. + * + * DEVIATION (see U9 report): the cli-agent branch of the *live* CE stage loop + * (replacing the interactive factory with one-shot turns inside + * `startStage`/`continueStage`) is not yet wired — only the option seam and + * its resolver contract land here. The engine-side one-shot runner is ready; + * the remaining work is invoking it from the stage loop. + */ + executor?: CeSessionExecutor; } /** @@ -176,6 +203,7 @@ export class CeOrchestrator { private readonly factory: CreateInteractiveAiSessionFactory | undefined; private readonly projectRoot: string; private readonly turnTimeoutMs: number; + private readonly executor: CeSessionExecutor; /** Live in-memory session handles keyed by ce_session id. */ private readonly live = new Map(); /** Mid-turn working output per session (transient; flushed to history on settle). */ @@ -194,6 +222,17 @@ export class CeOrchestrator { this.factory = deps.createInteractiveAiSession ?? deps.ctx.createInteractiveAiSession; this.projectRoot = deps.projectRoot ?? deps.ctx.taskStore.getRootDir(); this.turnTimeoutMs = deps.turnTimeoutMs ?? DEFAULT_TURN_TIMEOUT_MS; + this.executor = deps.executor ?? { kind: "model" }; + } + + /** + * Resolve the session backend for a CE stage (U9 seam). Threaded from + * `OrchestratorDeps.executor`; defaults to the model-backed interactive + * session. Exposed so the seam contract is directly assertable in tests and so + * future stage-loop wiring has a single resolution point. + */ + resolveExecutor(): CeSessionExecutor { + return this.executor; } /**