feat: bootstrap cli-agent runtime and wire executor, transport, chat, and recovery seams
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
11
.changeset/cli-agent-runtime-bootstrap.md
Normal file
11
.changeset/cli-agent-runtime-bootstrap.md
Normal file
@@ -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.
|
||||
@@ -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,
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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<string>;
|
||||
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 = "";
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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<void>;
|
||||
};
|
||||
// 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", {
|
||||
|
||||
152
packages/engine/src/cli-agent/__tests__/runtime.test.ts
Normal file
152
packages/engine/src/cli-agent/__tests__/runtime.test.ts
Normal file
@@ -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.
|
||||
});
|
||||
});
|
||||
176
packages/engine/src/cli-agent/runtime.ts
Normal file
176
packages/engine/src/cli-agent/runtime.ts
Normal file
@@ -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}`;
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user