diff --git a/.changeset/cli-agent-runtime-bootstrap.md b/.changeset/cli-agent-runtime-bootstrap.md new file mode 100644 index 0000000000..7136869088 --- /dev/null +++ b/.changeset/cli-agent-runtime-bootstrap.md @@ -0,0 +1,11 @@ +--- +"@runfusion/fusion": minor +--- + +Bootstrap the CLI Agent Executor runtime and wire it end-to-end. + +A new `createCliAgentRuntime` factory (engine) constructs the per-project bundle — a `CliSessionStore` over the project's existing core Database, a per-runtime adapter registry with all five bundled adapters, the `CliSessionManager` (PTY lifecycle), the `TelemetryHub` (per-session token registry rebuilt from live records), and the `CliResumeCoordinator` (relaunch re-mints a hook token + rewrites hook scripts) — returning the executor bundle, the `isWorktreeResumeReserved` / `isCliSessionWaitingOnInput` predicates, and a scoped `dispose`. + +The runtime is instantiated per project in `InProcessRuntime` behind the `experimentalFeatures.cliAgentExecutor` flag (opt-in, matching the `workflowGraphExecutor` precedent): the bundle threads into `TaskExecutorOptions.cliAgentRuntime`, the predicates feed the self-healing idle-worktree sweep and the stuck-task detector, and `resumeCoordinator.recoverOnStart()` runs non-blocking after engine start (errors logged, never thrown). The dashboard hook endpoint URL is derived from a server-threaded option, falling back to a localhost URL from `FUSION_DASHBOARD_PORT` (default 4040). + +The dashboard now resolves the project's `TelemetryHub` via `cliAgentHubResolver`, mounts the cli-sessions transport from the runtime's manager + store, and brokers cli-backed chat sends: a chat session with a `cliExecutorAdapterId` routes composer sends to a `CliChatSessionRunner` (instead of the model agent loop), and the hub's sanitized telemetry is routed per-session into the runner's transcript handler. diff --git a/packages/cli/src/commands/dashboard.ts b/packages/cli/src/commands/dashboard.ts index 0618905751..ac1c72597a 100644 --- a/packages/cli/src/commands/dashboard.ts +++ b/packages/cli/src/commands/dashboard.ts @@ -25,6 +25,9 @@ import { } from "@fusion/core"; import { createServer, + AttachTicketStore, + CliInputAttributionLog, + CliConfirmAdvanceRegistry, GitHubClient, createSkillsAdapter, getCliPackageVersion, @@ -1741,9 +1744,33 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?: // to createServer — routes derived from getPluginRoutes() rely on it. await phaseTime("pluginLoadingPromise (await)", () => pluginLoadingPromise); + // ── CLI Agent Executor: hub resolver + session transport ───────────── + // + // The hook route validates a per-session token against the project's live + // TelemetryHub; resolve it from that project's engine. The cli-sessions + // transport (REST + WS attach) is supplied from the cwd project's runtime + // (the canonical single-project surface) when the experimental flag is on. + // + const cliAgentHubResolver = (projectId: string | undefined, _sessionId: string) => { + const engine = projectId ? engineManager.getEngine(projectId) : cwdEngine; + return engine?.getCliAgentRuntime()?.bundle.hub; + }; + const cwdCliAgentRuntime = cwdEngine?.getCliAgentRuntime(); + const cliSessionTransport = cwdCliAgentRuntime + ? { + manager: cwdCliAgentRuntime.bundle.manager, + store: cwdCliAgentRuntime.bundle.store, + ticketStore: new AttachTicketStore(), + attributionLog: new CliInputAttributionLog(), + confirmAdvance: new CliConfirmAdvanceRegistry(), + } + : undefined; + app = createServer(store, { engine: cwdEngine, engineManager, + cliAgentHubResolver, + cliSessionTransport, hybridExecutor, centralCore: centralCoreForEngine, authStorage: dashboardAuthStorage, diff --git a/packages/dashboard/src/__tests__/chat-manager-cli-send.test.ts b/packages/dashboard/src/__tests__/chat-manager-cli-send.test.ts new file mode 100644 index 0000000000..3abf0fe1ab --- /dev/null +++ b/packages/dashboard/src/__tests__/chat-manager-cli-send.test.ts @@ -0,0 +1,71 @@ +/** + * ChatManager.sendMessage cli-agent send-branch (CLI Agent Executor integration). + * + * When a chat session selects a cli-agent executor (`cliExecutorAdapterId`), + * sendMessage must broker the composer text to the injected CliChatSessionRunner + * (ensureSession + send) rather than running the model agent loop. Narrow fakes: + * no real ChatStore, no pi-ai agent, no PTY, no network, no port 4040. + */ + +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { ChatManager } from "../chat.js"; + +const mockChatStore = { + getSession: vi.fn(), + createSession: vi.fn(), + addMessage: vi.fn(), + getMessages: vi.fn(), + updateSession: vi.fn(), + setCliSessionFile: vi.fn(), + setInFlightGeneration: vi.fn(), + getRoomMessages: vi.fn(), +}; + +function makeManager(): ChatManager { + return new ChatManager(mockChatStore as never, "/tmp/test"); +} + +describe("ChatManager.sendMessage — cli-agent send branch", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("routes a cli-executor chat session's composer send to runner.send", async () => { + mockChatStore.getSession.mockReturnValue({ + id: "chat-cli", + cliExecutorAdapterId: "claude-code", + projectId: "proj-1", + }); + + const ensureSession = vi.fn(async () => "cli-session-1"); + const send = vi.fn(async () => "sent" as const); + const manager = makeManager(); + manager.setCliChatRunner({ ensureSession, send }, "proj-1"); + + await manager.sendMessage("chat-cli", "hello agent"); + + expect(ensureSession).toHaveBeenCalledWith("chat-cli", { projectId: "proj-1" }); + expect(send).toHaveBeenCalledWith("chat-cli", "hello agent"); + // The model-agent path persists in-flight generation state; the cli branch + // must NOT touch it. + expect(mockChatStore.setInFlightGeneration).not.toHaveBeenCalled(); + }); + + it("uses the session's projectId when no explicit runner projectId is set", async () => { + mockChatStore.getSession.mockReturnValue({ + id: "chat-cli2", + cliExecutorAdapterId: "codex", + projectId: "proj-from-session", + }); + const ensureSession = vi.fn(async () => "cli-session-2"); + const send = vi.fn(async () => "queued" as const); + const manager = makeManager(); + // No projectId passed to setCliChatRunner → falls back to session.projectId. + manager.setCliChatRunner({ ensureSession, send }); + + await manager.sendMessage("chat-cli2", "queued please"); + + expect(ensureSession).toHaveBeenCalledWith("chat-cli2", { projectId: "proj-from-session" }); + expect(send).toHaveBeenCalledWith("chat-cli2", "queued please"); + }); +}); diff --git a/packages/dashboard/src/__tests__/cli-agent-runtime-wiring.test.ts b/packages/dashboard/src/__tests__/cli-agent-runtime-wiring.test.ts new file mode 100644 index 0000000000..d5bf274ded --- /dev/null +++ b/packages/dashboard/src/__tests__/cli-agent-runtime-wiring.test.ts @@ -0,0 +1,122 @@ +// @vitest-environment node + +/** + * CLI Agent Executor server-wiring contract (integration bootstrap). + * + * Proves that a real `createCliAgentRuntime` bundle (over a temp in-memory DB, + * PTY mocked at the loadPty seam) satisfies the shapes the dashboard ServerOptions + * consume: + * - `cliAgentHubResolver(projectId, sessionId)` resolves the project's live + * TelemetryHub from the runtime bundle. + * - `cliSessionTransport` accepts the runtime's manager + store and the + * transport-owned ticket/attribution/confirm singletons, and the + * cli-sessions router mounts against that dep without error. + * + * No real PTY, no network, no port 4040. + */ + +import express from "express"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { rm } from "node:fs/promises"; +import { Database } from "@fusion/core"; +import type { IPty } from "node-pty"; +import { createCliAgentRuntime, type BootstrappedCliAgentRuntime } from "@fusion/engine"; +import { + AttachTicketStore, + CliInputAttributionLog, + CliConfirmAdvanceRegistry, +} from "../cli-session-transport.js"; +import { createCliSessionsRouter } from "../routes/cli-sessions.js"; +import { request } from "../test-request.js"; + +function mockPty(): typeof import("node-pty") { + return { + spawn() { + return { + pid: 1, + onData: () => ({ dispose() {} }), + onExit: () => ({ dispose() {} }), + write() {}, + resize() {}, + pause() {}, + resume() {}, + kill() {}, + clear() {}, + } as unknown as IPty; + }, + } as unknown as typeof import("node-pty"); +} + +describe("cli-agent runtime server wiring", () => { + let runtime: BootstrappedCliAgentRuntime; + let db: Database; + let tmpDir: string; + + beforeEach(() => { + tmpDir = mkdtempSync(join(tmpdir(), "fn-cli-wiring-")); + const fusionDir = join(tmpDir, ".fusion"); + db = new Database(fusionDir, { inMemory: true }); + db.init(); + runtime = createCliAgentRuntime({ + fusionDir, + db, + projectId: "proj-a", + hookEndpointUrl: "http://127.0.0.1:4040/api/cli-agent/hooks", + managerOptions: { loadPty: async () => mockPty() }, + }); + }); + + afterEach(async () => { + runtime.dispose(); + db.close(); + await rm(tmpDir, { recursive: true, force: true }); + }); + + it("cliAgentHubResolver resolves the project's hub from the runtime bundle", () => { + const engines = new Map([["proj-a", { getCliAgentRuntime: () => runtime }]]); + const cliAgentHubResolver = (projectId: string | undefined, _sessionId: string) => { + const engine = projectId ? engines.get(projectId) : undefined; + return engine?.getCliAgentRuntime()?.bundle.hub; + }; + + expect(cliAgentHubResolver("proj-a", "cli-1")).toBe(runtime.bundle.hub); + expect(cliAgentHubResolver("missing", "cli-1")).toBeUndefined(); + expect(cliAgentHubResolver(undefined, "cli-1")).toBeUndefined(); + }); + + it("cliSessionTransport dep is satisfied by the runtime manager + store, and the router mounts", async () => { + // Seed a session so a transport-backed list route returns it. + runtime.bundle.store.createSession({ + adapterId: runtime.bundle.registry.ids()[0], + projectId: "proj-a", + purpose: "execute", + taskId: "FN-1", + worktreePath: "/tmp/wt", + agentState: "busy", + }); + + const transport = { + manager: runtime.bundle.manager, + store: runtime.bundle.store, + ticketStore: new AttachTicketStore(), + attributionLog: new CliInputAttributionLog(), + confirmAdvance: new CliConfirmAdvanceRegistry(), + }; + + const app = express(); + app.use(express.json()); + app.use("/api/cli-sessions", createCliSessionsRouter(transport)); + + const res = await request( + app as unknown as (req: import("http").IncomingMessage, res: import("http").ServerResponse) => void, + "GET", + "/api/cli-sessions?projectId=proj-a", + ); + expect(res.status).toBe(200); + const sessions = res.body.sessions as Array<{ taskId?: string }>; + expect(sessions.some((s) => s.taskId === "FN-1")).toBe(true); + }); +}); diff --git a/packages/dashboard/src/chat.ts b/packages/dashboard/src/chat.ts index 98f83e4f8a..d0e9948d95 100644 --- a/packages/dashboard/src/chat.ts +++ b/packages/dashboard/src/chat.ts @@ -735,6 +735,29 @@ export class ChatManager { private messageStore?: MessageStore, ) {} + /** + * Runner for CLI-agent-backed chat sessions (CLI Agent Executor). When a chat + * session selects a cli-agent executor (`cliExecutorAdapterId`), composer sends + * are brokered to the live PTY through this runner instead of the model agent + * loop. Injected post-construction (the runtime is built per-project at boot, + * after the ChatManager) so the positional ctor stays stable. + */ + private cliChatRunner?: { + ensureSession(chatSessionId: string, opts: { projectId: string; worktreePath?: string | null }): Promise; + send(chatSessionId: string, text: string): Promise<"sent" | "queued">; + }; + /** Project id used when the runner spawns a CLI session for a chat. */ + private cliChatProjectId?: string; + + /** Wire (or clear) the CLI-agent chat runner and its owning project id. */ + setCliChatRunner( + runner: ChatManager["cliChatRunner"] | undefined, + projectId?: string, + ): void { + this.cliChatRunner = runner; + this.cliChatProjectId = projectId; + } + private queueInFlightGenerationPersist(sessionId: string, snapshot: ChatInFlightGenerationState | null): void { const existingTimer = this.inFlightPersistTimers.get(sessionId); if (existingTimer) { @@ -1392,6 +1415,26 @@ export class ChatManager { const broadcastOptions = { generationId }; const session = this.chatStore.getSession(sessionId); + + // CLI-agent-backed chat: a session that selected a cli-agent executor brokers + // its composer sends to the live PTY (via the runner) rather than running the + // model agent loop. The runner persists the user message + the transcript. + if (session?.cliExecutorAdapterId && this.cliChatRunner) { + const runner = this.cliChatRunner; + try { + await runner.ensureSession(sessionId, { + projectId: this.cliChatProjectId ?? session.projectId ?? "", + }); + await runner.send(sessionId, content); + } finally { + const current = this.activeGenerations.get(sessionId); + if (current?.generationId === generationId) { + this.activeGenerations.delete(sessionId); + } + } + return; + } + let agentResult: AgentResult | undefined; let accumulatedThinking = ""; let accumulatedText = ""; diff --git a/packages/dashboard/src/index.ts b/packages/dashboard/src/index.ts index 5ba08b81a4..8d49616729 100644 --- a/packages/dashboard/src/index.ts +++ b/packages/dashboard/src/index.ts @@ -77,3 +77,14 @@ export { MAX_CARRY_LENGTH, type NeutralizeResult, } from "./cli-session-output-filter.js"; + +// CLI Agent Executor transport dependencies — re-exported so the CLI boot +// (packages/cli dashboard command) can construct the per-session attach-ticket +// store, input-attribution log, and confirm-advance registry that the +// cli-sessions transport routes require, then thread them into ServerOptions. +export { + AttachTicketStore, + CliInputAttributionLog, + CliConfirmAdvanceRegistry, + type CliSessionTransportDeps, +} from "./cli-session-transport.js"; diff --git a/packages/dashboard/src/server.ts b/packages/dashboard/src/server.ts index 9db182c8bb..cb62dd41c4 100644 --- a/packages/dashboard/src/server.ts +++ b/packages/dashboard/src/server.ts @@ -56,6 +56,7 @@ import { rehydrateFromStore as rehydrateMilestoneSliceSessions, } from "./milestone-slice-interview.js"; import { ChatManager } from "./chat.js"; +import { CliChatSessionRunner } from "./cli-chat.js"; import { stopAllDevServers } from "./dev-server-routes.js"; import type { SkillsAdapter } from "./skills-adapter.js"; import { createAuthMiddleware, authenticateUpgradeRequest, getDaemonToken } from "./auth-middleware.js"; @@ -1136,6 +1137,73 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT options?.engine?.getMessageStore(), ); + // CLI Agent Executor — chat surface wiring. When the cli-session transport is + // supplied (the runtime is live), broker cli-backed chat sends to the PTY and + // route the project hub's sanitized telemetry into the runner's transcript + // handler. The listener is keyed per-session inside one closure so it composes + // safely even if other taps exist. + if (options?.cliSessionTransport && options.cliAgentHubResolver) { + try { + const cliTransportStore = options.cliSessionTransport.store; + // The transport's `manager` is typed for the attach/inject transport slice; + // the chat runner additionally needs `spawn`. The concrete engine + // CliSessionManager provides both — widen via a structural cast to the + // spawn/inject slice the runner consumes. + const spawnInject = options.cliSessionTransport.manager as unknown as { + spawn: (opts: { + adapterId: string; + projectId: string; + purpose: "chat"; + chatSessionId: string; + worktreePath?: string | null; + resume?: { sessionId: string; nativeSessionId: string }; + }) => Promise<{ id: string; nativeSessionId: string | null; agentState: string }>; + inject: (sessionId: string, text: string) => Promise; + }; + // The runner needs spawn/inject (manager) + a fresh session record getter + // (store). Compose the slice the runner expects so flush decisions read + // authoritative records. + const cliChatRunner = new CliChatSessionRunner({ + store: chatStore, + manager: { + spawn: (opts) => spawnInject.spawn(opts), + inject: (sessionId, text) => spawnInject.inject(sessionId, text), + getSession: (sessionId) => { + const r = cliTransportStore.getSession(sessionId); + return r + ? { id: r.id, nativeSessionId: r.nativeSessionId, agentState: r.agentState } + : undefined; + }, + }, + }); + chatManager.setCliChatRunner(cliChatRunner, options.engine?.getProjectId?.()); + const hub = options.cliAgentHubResolver(undefined, ""); + if (hub) { + hub.setEventListener((cliSessionId, event) => { + // Per-session routing inside one listener: map the CLI session id to its + // owning chat session (only chat-purpose sessions carry chatSessionId); + // non-chat sessions (task/validator) are ignored here. + const record = cliTransportStore.getSession(cliSessionId); + const chatSessionId = record?.chatSessionId; + if (!chatSessionId) return; + void cliChatRunner + .handleTelemetry(chatSessionId, { + kind: event.kind, + text: event.text, + nativeSessionId: event.nativeSessionId, + }) + .catch(() => { + // best-effort: a transcript-handler throw must never break ingest. + }); + }); + } + } catch (err) { + runtimeLogger.warn?.("CLI-agent chat runner wiring failed", { + message: err instanceof Error ? err.message : String(err), + }); + } + } + const runAiSessionCleanup = (maxAgeMs: number, source: "initial" | "scheduled") => { const result = aiSessionStore.cleanupStaleSessions(maxAgeMs); runtimeLogger.info("AI session cleanup summary", { diff --git a/packages/engine/src/cli-agent/__tests__/runtime.test.ts b/packages/engine/src/cli-agent/__tests__/runtime.test.ts new file mode 100644 index 0000000000..c504d9c52f --- /dev/null +++ b/packages/engine/src/cli-agent/__tests__/runtime.test.ts @@ -0,0 +1,152 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { rm } from "node:fs/promises"; +import { Database } from "@fusion/core"; +import type { IPty } from "node-pty"; +import { createCliAgentRuntime, type BootstrappedCliAgentRuntime } from "../runtime.js"; +import { BUNDLED_CLI_ADAPTERS } from "../adapters/index.js"; + +// ── Mock PTY at the loadPtyModule seam (runtime construction must not touch a +// real PTY; spawning is not exercised here). ─────────────────────────────────── + +function makeMockPtyModule(): typeof import("node-pty") { + return { + spawn() { + const mock = { + pid: 4242, + cols: 80, + rows: 24, + process: "mock", + handleFlowControl: false, + onData: () => ({ dispose: () => {} }), + onExit: () => ({ dispose: () => {} }), + write: () => {}, + resize: () => {}, + pause: () => {}, + resume: () => {}, + kill: () => {}, + clear: () => {}, + }; + return mock as unknown as IPty; + }, + } as unknown as typeof import("node-pty"); +} + +interface Harness { + runtime: BootstrappedCliAgentRuntime; + db: Database; + tmpDir: string; + fusionDir: string; +} + +function makeHarness(): Harness { + const tmpDir = mkdtempSync(join(tmpdir(), "fn-cli-runtime-test-")); + const fusionDir = join(tmpDir, ".fusion"); + const db = new Database(fusionDir, { inMemory: true }); + db.init(); + const runtime = createCliAgentRuntime({ + fusionDir, + db, + projectId: "proj-1", + hookEndpointUrl: "http://127.0.0.1:4040/api/cli-agent/hooks", + managerOptions: { loadPty: async () => makeMockPtyModule() }, + }); + return { runtime, db, tmpDir, fusionDir }; +} + +describe("createCliAgentRuntime", () => { + let h: Harness; + + beforeEach(() => { + h = makeHarness(); + }); + + afterEach(async () => { + h.runtime.dispose(); + h.db.close(); + await rm(h.tmpDir, { recursive: true, force: true }); + }); + + it("constructs the full bundle with manager, hub, registry, and store", () => { + const { bundle } = h.runtime; + expect(bundle.manager).toBeDefined(); + expect(bundle.hub).toBeDefined(); + expect(bundle.registry).toBeDefined(); + expect(bundle.store).toBeDefined(); + expect(bundle.projectId).toBe("proj-1"); + expect(bundle.hookEndpointUrl).toBe("http://127.0.0.1:4040/api/cli-agent/hooks"); + }); + + it("registers all bundled adapters into a per-runtime registry", () => { + const ids = h.runtime.bundle.registry.ids().sort(); + const expected = BUNDLED_CLI_ADAPTERS.map((a) => a.id).sort(); + expect(ids).toEqual(expected); + expect(ids).toHaveLength(5); + }); + + it("does not pollute a second runtime's registry (no duplicate-registration)", () => { + // A second runtime over the SAME process registers the same adapters again; + // a per-runtime registry means no DuplicateCliAdapterError is thrown. + const second = createCliAgentRuntime({ + fusionDir: h.fusionDir, + db: h.db, + projectId: "proj-2", + hookEndpointUrl: "http://127.0.0.1:4040/api/cli-agent/hooks", + managerOptions: { loadPty: async () => makeMockPtyModule() }, + }); + expect(second.bundle.registry.ids()).toHaveLength(5); + second.dispose(); + }); + + it("isWorktreeResumeReserved reflects resume-eligible session records", () => { + const adapterId = BUNDLED_CLI_ADAPTERS[0].id; + // A live-on-restart record (busy) reserves its worktree. + h.runtime.bundle.store.createSession({ + adapterId, + projectId: "proj-1", + purpose: "execute", + taskId: "FN-1", + worktreePath: "/wt/reserved", + agentState: "busy", + }); + expect(h.runtime.isWorktreeResumeReserved("/wt/reserved")).toBe(true); + expect(h.runtime.isWorktreeResumeReserved("/wt/other")).toBe(false); + }); + + it("isCliSessionWaitingOnInput is true only when a task's session is waitingOnInput", () => { + const adapterId = BUNDLED_CLI_ADAPTERS[0].id; + h.runtime.bundle.store.createSession({ + adapterId, + projectId: "proj-1", + purpose: "execute", + taskId: "FN-busy", + worktreePath: "/wt/busy", + agentState: "busy", + }); + h.runtime.bundle.store.createSession({ + adapterId, + projectId: "proj-1", + purpose: "execute", + taskId: "FN-wait", + worktreePath: "/wt/wait", + agentState: "waitingOnInput", + }); + expect(h.runtime.isCliSessionWaitingOnInput("FN-wait")).toBe(true); + expect(h.runtime.isCliSessionWaitingOnInput("FN-busy")).toBe(false); + expect(h.runtime.isCliSessionWaitingOnInput("FN-unknown")).toBe(false); + }); + + it("exposes a resume coordinator whose recoverOnStart runs cleanly with no orphans", async () => { + const results = await h.runtime.resumeCoordinator.recoverOnStart(); + expect(Array.isArray(results)).toBe(true); + expect(results).toHaveLength(0); + }); + + it("dispose tears down the manager without throwing and is idempotent", () => { + expect(() => h.runtime.dispose()).not.toThrow(); + expect(() => h.runtime.dispose()).not.toThrow(); + // Re-dispose in afterEach is also safe. + }); +}); diff --git a/packages/engine/src/cli-agent/runtime.ts b/packages/engine/src/cli-agent/runtime.ts new file mode 100644 index 0000000000..c14900d3cf --- /dev/null +++ b/packages/engine/src/cli-agent/runtime.ts @@ -0,0 +1,176 @@ +/** + * createCliAgentRuntime — the per-project bootstrap that wires the CLI Agent + * Executor subsystem together (U-final integration). + * + * Every component (PTY session manager, telemetry hub, adapter registry, resume + * coordinator) is built with injection seams and tested in isolation; this + * factory is the single place that actually instantiates the live bundle and + * stitches the seams: + * + * - Builds a {@link CliSessionStore} over the project's EXISTING core Database + * (never opens a second connection — the store is a thin query layer). + * - Registers all bundled adapters into a fresh {@link CliAdapterRegistry} (a + * per-runtime registry, NOT the process-wide `defaultCliAdapterRegistry`, so + * multi-project boots never collide on duplicate-registration). + * - Constructs the {@link CliSessionManager} (PTY lifecycle) and + * {@link TelemetryHub} (per-session token registry, rebuilt from live store + * records on construction). + * - Constructs the {@link CliResumeCoordinator}, wiring `reattachTelemetry` to + * re-mint a hook token + rewrite the session's hook scripts on relaunch. + * + * It returns the {@link CliAgentRuntime} bundle the {@link TaskExecutor} + * consumes, plus the two narrow predicates the self-healing / stuck-task seams + * read, plus a `dispose` that tears the manager down cleanly (scoped SIGKILL of + * the runtime's own PTYs only — never the dashboard / port 4040). + */ + +import { CliSessionStore } from "@fusion/core"; +import type { Database } from "@fusion/core"; +import { CliAdapterRegistry } from "./adapter.js"; +import { BUNDLED_CLI_ADAPTERS } from "./adapters/index.js"; +import { CliSessionManager, type CliSessionManagerOptions } from "./session-manager.js"; +import { TelemetryHub, type TelemetryHubOptions } from "./telemetry-hub.js"; +import { CliResumeCoordinator } from "./resume-coordinator.js"; +import { writeSessionHookScripts } from "./hook-scripts.js"; +import type { CliAgentRuntime } from "../executor.js"; + +/** Options for {@link createCliAgentRuntime}. */ +export interface CreateCliAgentRuntimeOptions { + /** The project's `.fusion` dir (scratch root for hook scripts). */ + fusionDir: string; + /** The project's already-open core Database (reused, never re-opened). */ + db: Database; + /** Project this runtime drives (`cli_sessions.projectId`). */ + projectId: string; + /** + * Absolute URL of the dashboard hook ingestion endpoint the generated hook + * scripts POST to (e.g. `http://127.0.0.1:4040/api/cli-agent/hooks`). + */ + hookEndpointUrl: string; + /** Optional override for the hook scratch-dir root (tests). */ + hookDirRoot?: string; + /** Optional notification dispatch forwarded to the TelemetryHub. */ + onNotification?: TelemetryHubOptions["onNotification"]; + /** + * Test seams forwarded to the {@link CliSessionManager} (e.g. a mocked node-pty + * loader so runtime construction never touches a real PTY). + */ + managerOptions?: Pick< + CliSessionManagerOptions, + "loadPty" | "scrollbackBytes" | "concurrencyCeiling" | "highWatermark" | "injectionQuietWindowMs" + >; +} + +/** + * The full bootstrapped CLI-agent runtime: the executor bundle, the predicates + * the self-healing + stuck-task seams read, the resume coordinator, and dispose. + */ +export interface BootstrappedCliAgentRuntime { + /** The bundle threaded into {@link TaskExecutorOptions.cliAgentRuntime}. */ + bundle: CliAgentRuntime; + /** Engine-start orphan recovery sweep (call after engine start; errors logged). */ + resumeCoordinator: CliResumeCoordinator; + /** + * Self-healing seam: whether a worktree path backs a resume-eligible session + * record (so idle-worktree sweeps treat it as in-use). Delegates to the resume + * coordinator's reservation set. + */ + isWorktreeResumeReserved: (worktreePath: string) => boolean; + /** + * Stuck-task seam: whether a task's live CLI session is `waitingOnInput` + * (expected idleness — suppress stuck flagging). Reads the live store. + */ + isCliSessionWaitingOnInput: (taskId: string) => boolean; + /** Tear down the PTY manager (scoped SIGKILL of this runtime's PTYs only). */ + dispose: () => void; +} + +/** + * Construct the per-project CLI-agent runtime bundle. Pure construction — no IO + * beyond the store's reads against the supplied Database; spawning a PTY or + * running recovery is the caller's job (`resumeCoordinator.recoverOnStart()`). + */ +export function createCliAgentRuntime( + options: CreateCliAgentRuntimeOptions, +): BootstrappedCliAgentRuntime { + const { fusionDir, db, projectId, hookEndpointUrl } = options; + + // 1. Store over the project's existing Database (thin query layer; no new conn). + const store = new CliSessionStore(fusionDir, db); + + // 2. A per-runtime registry with every bundled adapter (not the process-wide + // singleton — avoids duplicate-registration across multi-project boots). + const registry = new CliAdapterRegistry(); + for (const adapter of BUNDLED_CLI_ADAPTERS) { + registry.register(adapter); + } + + // 3. PTY session manager. + const manager = new CliSessionManager({ + registry, + store, + ...options.managerOptions, + }); + + // 4. Telemetry hub — rebuilds its per-session token registry from live store + // records on construction. + const hub = new TelemetryHub({ + store, + onNotification: options.onNotification, + }); + + // 5. Resume coordinator. On relaunch, re-mint a hook token and rewrite the + // session's hook scripts so the resumed CLI POSTs with a fresh, valid token. + const resumeCoordinator = new CliResumeCoordinator({ + store, + manager, + registry, + reattachTelemetry: async (session) => { + const token = hub.issueToken(session.id); + await writeSessionHookScripts({ + sessionId: session.id, + token, + endpointUrl: hookEndpointUrl, + dir: hookScriptDir(options, session.id), + }); + }, + }); + + const bundle: CliAgentRuntime = { + manager, + hub, + registry, + store, + projectId, + hookEndpointUrl, + hookDirRoot: options.hookDirRoot, + }; + + return { + bundle, + resumeCoordinator, + isWorktreeResumeReserved: (worktreePath: string) => + resumeCoordinator.resumeReservedWorktrees().has(worktreePath), + isCliSessionWaitingOnInput: (taskId: string) => { + // A task's live session is "waiting on input" when any of its session + // records is in the waitingOnInput state. Defensive: a store error means + // "not waiting" (the stuck detector's own guard re-asserts this too). + try { + return store + .listByTask(taskId) + .some((s) => s.agentState === "waitingOnInput"); + } catch { + return false; + } + }, + dispose: () => { + manager.dispose(); + }, + }; +} + +/** Resolve the per-session hook scratch dir under the configured root. */ +function hookScriptDir(options: CreateCliAgentRuntimeOptions, sessionId: string): string { + const root = options.hookDirRoot ?? `${options.fusionDir}/cli-agent/hooks`; + return `${root}/${sessionId}`; +} diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 13c8fb59e2..b40dcefac8 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -637,6 +637,12 @@ export { type CliSessionManagerOptions, type SpawnCliSessionOptions, } from "./cli-agent/session-manager.js"; +// CLI Agent Executor — per-project runtime bootstrap (integration). +export { + createCliAgentRuntime, + type CreateCliAgentRuntimeOptions, + type BootstrappedCliAgentRuntime, +} from "./cli-agent/runtime.js"; // CLI Agent Executor — resume coordinator + self-healing/stuck integration (U8). export { CliResumeCoordinator, diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index a379a234fc..899e99a338 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -800,6 +800,16 @@ export class ProjectEngine { return this.runtime.getSelfHealingManager(); } + /** + * Get the bootstrapped CLI Agent Executor runtime (PTY manager + telemetry hub + * + adapter registry + resume coordinator), or undefined when the experimental + * flag is off. The dashboard reads this to resolve the project's TelemetryHub + * (hook route) and supply the cli-session transport dependency. + */ + getCliAgentRuntime() { + return this.runtime.getCliAgentRuntime(); + } + /** Get the project working directory. */ getWorkingDirectory(): string { return this.config.workingDirectory; diff --git a/packages/engine/src/project-runtime.ts b/packages/engine/src/project-runtime.ts index 1c09ee07c9..709a10ece2 100644 --- a/packages/engine/src/project-runtime.ts +++ b/packages/engine/src/project-runtime.ts @@ -55,6 +55,14 @@ export interface ProjectRuntimeConfig { * Useful when the caller (e.g. dashboard.ts) owns and watches the store. */ externalTaskStore?: TaskStore; + /** + * Absolute URL of the dashboard's CLI-agent hook ingestion endpoint that + * generated hook scripts POST to (e.g. `http://127.0.0.1:4040/api/cli-agent/hooks`). + * Threaded from the dashboard boot once the listening port is known. When + * absent, the runtime derives a localhost URL from `FUSION_DASHBOARD_PORT` + * (falling back to 4040). + */ + cliAgentHookEndpointUrl?: string; } /** diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index da1639bcff..ef7523fad4 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -17,6 +17,8 @@ import { Scheduler } from "../scheduler.js"; import type { PrMonitor, PrComment } from "../pr-monitor.js"; import type { PrInfo } from "@fusion/core"; import { TaskExecutor, type TaskExecutorOptions } from "../executor.js"; +import { isExperimentalFeatureEnabled } from "@fusion/core"; +import { createCliAgentRuntime, type BootstrappedCliAgentRuntime } from "../cli-agent/runtime.js"; import { WorktreePool, isGitRepository, type PoolInvariantViolation } from "../worktree-pool.js"; import { AgentSemaphore } from "../concurrency.js"; import { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "../agent-heartbeat.js"; @@ -98,6 +100,13 @@ export class InProcessRuntime private worktreePool!: WorktreePool; private globalSemaphore?: AgentSemaphore; private stuckTaskDetector?: StuckTaskDetector; + /** + * Per-project CLI Agent Executor runtime bundle (PTY manager + telemetry hub + + * adapter registry + resume coordinator). Built in `start()` when the + * `cliAgentExecutor` experimental flag is on; threaded into the executor + + * self-healing/stuck seams; disposed in `stop()`. + */ + private cliAgentRuntime?: BootstrappedCliAgentRuntime; private usageLimitPauser?: UsageLimitPauser; private selfHealingManager?: SelfHealingManager; private leaseManager?: MeshLeaseManager; @@ -385,8 +394,29 @@ export class InProcessRuntime await yieldEventLoop(); + // 5a-cli. Initialize the CLI Agent Executor runtime (behind the + // `cliAgentExecutor` experimental flag). Reuses the project's existing core + // Database; predicates feed the self-healing + stuck-task seams below. + if (isExperimentalFeatureEnabled(settings, "cliAgentExecutor")) { + try { + this.cliAgentRuntime = createCliAgentRuntime({ + fusionDir: this.taskStore.getFusionDir(), + db: this.taskStore.getDatabase(), + projectId: this.config.projectId, + hookEndpointUrl: this.resolveCliAgentHookEndpointUrl(), + }); + runtimeLog.log("CLI Agent Executor runtime initialized"); + } catch (cliErr) { + runtimeLog.warn( + `CLI Agent Executor runtime initialization failed (cli-agent nodes will report a config error):`, + cliErr instanceof Error ? cliErr.message : cliErr, + ); + } + } + // 5b. Initialize TaskExecutor this.stuckTaskDetector = new StuckTaskDetector(this.taskStore, { + isCliSessionWaitingOnInput: this.cliAgentRuntime?.isCliSessionWaitingOnInput, beforeRequeue: (taskId, reason, event) => this.selfHealingManager?.checkStuckBudget(taskId, reason, event) ?? Promise.resolve(true), onLoopDetected: (event) => this.executor?.handleLoopDetected(event) ?? Promise.resolve(false), onStuck: (event) => { @@ -447,6 +477,7 @@ export class InProcessRuntime pool: this.worktreePool, usageLimitPauser: this.usageLimitPauser, stuckTaskDetector: this.stuckTaskDetector, + cliAgentRuntime: this.cliAgentRuntime?.bundle, pluginRunner: this.pluginRunner, messageStore: this.messageStore, missionStore, @@ -723,6 +754,7 @@ export class InProcessRuntime this.selfHealingManager = new SelfHealingManager(this.taskStore, { rootDir: this.config.workingDirectory, agentStore: this.agentStore, + isWorktreeResumeReserved: this.cliAgentRuntime?.isWorktreeResumeReserved, recoverCompletedTask: (task) => this.executor.recoverCompletedTask(task), recoverFailedPreMergeStep: (task) => this.executor.recoverFailedPreMergeWorkflowStep(task), getExecutingTaskIds: () => this.executor.getExecutingTaskIds(), @@ -839,6 +871,25 @@ export class InProcessRuntime runtimeLog.warn(`Failed to stamp engineActiveSinceMs on runtime start: ${message}`); } + // 15. CLI Agent Executor: recover orphaned-live sessions left by a prior + // engine death. Non-blocking; errors are logged, never thrown. + if (this.cliAgentRuntime) { + const cliRuntime = this.cliAgentRuntime; + void cliRuntime.resumeCoordinator + .recoverOnStart() + .then((results) => { + if (results.length > 0) { + runtimeLog.log(`CLI Agent Executor recovered ${results.length} orphaned session(s) on start`); + } + }) + .catch((err) => { + runtimeLog.warn( + `CLI Agent Executor recoverOnStart failed (continuing):`, + err instanceof Error ? err.message : err, + ); + }); + } + this.setStatus("active"); runtimeLog.log(`InProcessRuntime started for project ${this.config.projectId}`); } catch (error) { @@ -890,6 +941,21 @@ export class InProcessRuntime runtimeLog.log("SelfHealingManager stopped"); } + // 2c. Dispose the CLI Agent Executor runtime (scoped SIGKILL of this + // runtime's own PTYs only — never the dashboard / port 4040). + if (this.cliAgentRuntime) { + try { + this.cliAgentRuntime.dispose(); + runtimeLog.log("CLI Agent Executor runtime disposed"); + } catch (cliErr) { + runtimeLog.warn( + `CLI Agent Executor dispose failed:`, + cliErr instanceof Error ? cliErr.message : cliErr, + ); + } + this.cliAgentRuntime = undefined; + } + // 2. Stop routine scheduler (stops new routine triggers; in-flight executions continue) if (this.routineScheduler) { this.routineScheduler.stop(); @@ -1208,6 +1274,28 @@ export class InProcessRuntime return this.selfHealingManager; } + /** + * Get the bootstrapped CLI Agent Executor runtime (if the experimental flag is + * on and construction succeeded). The dashboard reads this to resolve the + * project's TelemetryHub (hook route) and supply the cli-session transport. + */ + getCliAgentRuntime(): BootstrappedCliAgentRuntime | undefined { + return this.cliAgentRuntime; + } + + /** + * Resolve the dashboard CLI-agent hook ingestion endpoint URL. Prefers the + * value threaded from server boot (once the listening port is known); falls + * back to a localhost URL derived from `FUSION_DASHBOARD_PORT` (default 4040). + */ + private resolveCliAgentHookEndpointUrl(): string { + if (this.config.cliAgentHookEndpointUrl) { + return this.config.cliAgentHookEndpointUrl; + } + const port = Number(process.env.FUSION_DASHBOARD_PORT) || 4040; + return `http://127.0.0.1:${port}/api/cli-agent/hooks`; + } + /** * Get the HeartbeatTriggerScheduler instance (if initialized). * Returns undefined when agent monitoring is not available.