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";
|
} from "@fusion/core";
|
||||||
import {
|
import {
|
||||||
createServer,
|
createServer,
|
||||||
|
AttachTicketStore,
|
||||||
|
CliInputAttributionLog,
|
||||||
|
CliConfirmAdvanceRegistry,
|
||||||
GitHubClient,
|
GitHubClient,
|
||||||
createSkillsAdapter,
|
createSkillsAdapter,
|
||||||
getCliPackageVersion,
|
getCliPackageVersion,
|
||||||
@@ -1741,9 +1744,33 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
|
|||||||
// to createServer — routes derived from getPluginRoutes() rely on it.
|
// to createServer — routes derived from getPluginRoutes() rely on it.
|
||||||
await phaseTime("pluginLoadingPromise (await)", () => pluginLoadingPromise);
|
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, {
|
app = createServer(store, {
|
||||||
engine: cwdEngine,
|
engine: cwdEngine,
|
||||||
engineManager,
|
engineManager,
|
||||||
|
cliAgentHubResolver,
|
||||||
|
cliSessionTransport,
|
||||||
hybridExecutor,
|
hybridExecutor,
|
||||||
centralCore: centralCoreForEngine,
|
centralCore: centralCoreForEngine,
|
||||||
authStorage: dashboardAuthStorage,
|
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,
|
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 {
|
private queueInFlightGenerationPersist(sessionId: string, snapshot: ChatInFlightGenerationState | null): void {
|
||||||
const existingTimer = this.inFlightPersistTimers.get(sessionId);
|
const existingTimer = this.inFlightPersistTimers.get(sessionId);
|
||||||
if (existingTimer) {
|
if (existingTimer) {
|
||||||
@@ -1392,6 +1415,26 @@ export class ChatManager {
|
|||||||
const broadcastOptions = { generationId };
|
const broadcastOptions = { generationId };
|
||||||
|
|
||||||
const session = this.chatStore.getSession(sessionId);
|
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 agentResult: AgentResult | undefined;
|
||||||
let accumulatedThinking = "";
|
let accumulatedThinking = "";
|
||||||
let accumulatedText = "";
|
let accumulatedText = "";
|
||||||
|
|||||||
@@ -77,3 +77,14 @@ export {
|
|||||||
MAX_CARRY_LENGTH,
|
MAX_CARRY_LENGTH,
|
||||||
type NeutralizeResult,
|
type NeutralizeResult,
|
||||||
} from "./cli-session-output-filter.js";
|
} 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,
|
rehydrateFromStore as rehydrateMilestoneSliceSessions,
|
||||||
} from "./milestone-slice-interview.js";
|
} from "./milestone-slice-interview.js";
|
||||||
import { ChatManager } from "./chat.js";
|
import { ChatManager } from "./chat.js";
|
||||||
|
import { CliChatSessionRunner } from "./cli-chat.js";
|
||||||
import { stopAllDevServers } from "./dev-server-routes.js";
|
import { stopAllDevServers } from "./dev-server-routes.js";
|
||||||
import type { SkillsAdapter } from "./skills-adapter.js";
|
import type { SkillsAdapter } from "./skills-adapter.js";
|
||||||
import { createAuthMiddleware, authenticateUpgradeRequest, getDaemonToken } from "./auth-middleware.js";
|
import { createAuthMiddleware, authenticateUpgradeRequest, getDaemonToken } from "./auth-middleware.js";
|
||||||
@@ -1136,6 +1137,73 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
|
|||||||
options?.engine?.getMessageStore(),
|
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 runAiSessionCleanup = (maxAgeMs: number, source: "initial" | "scheduled") => {
|
||||||
const result = aiSessionStore.cleanupStaleSessions(maxAgeMs);
|
const result = aiSessionStore.cleanupStaleSessions(maxAgeMs);
|
||||||
runtimeLogger.info("AI session cleanup summary", {
|
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 CliSessionManagerOptions,
|
||||||
type SpawnCliSessionOptions,
|
type SpawnCliSessionOptions,
|
||||||
} from "./cli-agent/session-manager.js";
|
} 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).
|
// CLI Agent Executor — resume coordinator + self-healing/stuck integration (U8).
|
||||||
export {
|
export {
|
||||||
CliResumeCoordinator,
|
CliResumeCoordinator,
|
||||||
|
|||||||
@@ -800,6 +800,16 @@ export class ProjectEngine {
|
|||||||
return this.runtime.getSelfHealingManager();
|
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. */
|
/** Get the project working directory. */
|
||||||
getWorkingDirectory(): string {
|
getWorkingDirectory(): string {
|
||||||
return this.config.workingDirectory;
|
return this.config.workingDirectory;
|
||||||
|
|||||||
@@ -55,6 +55,14 @@ export interface ProjectRuntimeConfig {
|
|||||||
* Useful when the caller (e.g. dashboard.ts) owns and watches the store.
|
* Useful when the caller (e.g. dashboard.ts) owns and watches the store.
|
||||||
*/
|
*/
|
||||||
externalTaskStore?: TaskStore;
|
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 { PrMonitor, PrComment } from "../pr-monitor.js";
|
||||||
import type { PrInfo } from "@fusion/core";
|
import type { PrInfo } from "@fusion/core";
|
||||||
import { TaskExecutor, type TaskExecutorOptions } from "../executor.js";
|
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 { WorktreePool, isGitRepository, type PoolInvariantViolation } from "../worktree-pool.js";
|
||||||
import { AgentSemaphore } from "../concurrency.js";
|
import { AgentSemaphore } from "../concurrency.js";
|
||||||
import { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "../agent-heartbeat.js";
|
import { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "../agent-heartbeat.js";
|
||||||
@@ -98,6 +100,13 @@ export class InProcessRuntime
|
|||||||
private worktreePool!: WorktreePool;
|
private worktreePool!: WorktreePool;
|
||||||
private globalSemaphore?: AgentSemaphore;
|
private globalSemaphore?: AgentSemaphore;
|
||||||
private stuckTaskDetector?: StuckTaskDetector;
|
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 usageLimitPauser?: UsageLimitPauser;
|
||||||
private selfHealingManager?: SelfHealingManager;
|
private selfHealingManager?: SelfHealingManager;
|
||||||
private leaseManager?: MeshLeaseManager;
|
private leaseManager?: MeshLeaseManager;
|
||||||
@@ -385,8 +394,29 @@ export class InProcessRuntime
|
|||||||
|
|
||||||
await yieldEventLoop();
|
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
|
// 5b. Initialize TaskExecutor
|
||||||
this.stuckTaskDetector = new StuckTaskDetector(this.taskStore, {
|
this.stuckTaskDetector = new StuckTaskDetector(this.taskStore, {
|
||||||
|
isCliSessionWaitingOnInput: this.cliAgentRuntime?.isCliSessionWaitingOnInput,
|
||||||
beforeRequeue: (taskId, reason, event) => this.selfHealingManager?.checkStuckBudget(taskId, reason, event) ?? Promise.resolve(true),
|
beforeRequeue: (taskId, reason, event) => this.selfHealingManager?.checkStuckBudget(taskId, reason, event) ?? Promise.resolve(true),
|
||||||
onLoopDetected: (event) => this.executor?.handleLoopDetected(event) ?? Promise.resolve(false),
|
onLoopDetected: (event) => this.executor?.handleLoopDetected(event) ?? Promise.resolve(false),
|
||||||
onStuck: (event) => {
|
onStuck: (event) => {
|
||||||
@@ -447,6 +477,7 @@ export class InProcessRuntime
|
|||||||
pool: this.worktreePool,
|
pool: this.worktreePool,
|
||||||
usageLimitPauser: this.usageLimitPauser,
|
usageLimitPauser: this.usageLimitPauser,
|
||||||
stuckTaskDetector: this.stuckTaskDetector,
|
stuckTaskDetector: this.stuckTaskDetector,
|
||||||
|
cliAgentRuntime: this.cliAgentRuntime?.bundle,
|
||||||
pluginRunner: this.pluginRunner,
|
pluginRunner: this.pluginRunner,
|
||||||
messageStore: this.messageStore,
|
messageStore: this.messageStore,
|
||||||
missionStore,
|
missionStore,
|
||||||
@@ -723,6 +754,7 @@ export class InProcessRuntime
|
|||||||
this.selfHealingManager = new SelfHealingManager(this.taskStore, {
|
this.selfHealingManager = new SelfHealingManager(this.taskStore, {
|
||||||
rootDir: this.config.workingDirectory,
|
rootDir: this.config.workingDirectory,
|
||||||
agentStore: this.agentStore,
|
agentStore: this.agentStore,
|
||||||
|
isWorktreeResumeReserved: this.cliAgentRuntime?.isWorktreeResumeReserved,
|
||||||
recoverCompletedTask: (task) => this.executor.recoverCompletedTask(task),
|
recoverCompletedTask: (task) => this.executor.recoverCompletedTask(task),
|
||||||
recoverFailedPreMergeStep: (task) => this.executor.recoverFailedPreMergeWorkflowStep(task),
|
recoverFailedPreMergeStep: (task) => this.executor.recoverFailedPreMergeWorkflowStep(task),
|
||||||
getExecutingTaskIds: () => this.executor.getExecutingTaskIds(),
|
getExecutingTaskIds: () => this.executor.getExecutingTaskIds(),
|
||||||
@@ -839,6 +871,25 @@ export class InProcessRuntime
|
|||||||
runtimeLog.warn(`Failed to stamp engineActiveSinceMs on runtime start: ${message}`);
|
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");
|
this.setStatus("active");
|
||||||
runtimeLog.log(`InProcessRuntime started for project ${this.config.projectId}`);
|
runtimeLog.log(`InProcessRuntime started for project ${this.config.projectId}`);
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
@@ -890,6 +941,21 @@ export class InProcessRuntime
|
|||||||
runtimeLog.log("SelfHealingManager stopped");
|
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)
|
// 2. Stop routine scheduler (stops new routine triggers; in-flight executions continue)
|
||||||
if (this.routineScheduler) {
|
if (this.routineScheduler) {
|
||||||
this.routineScheduler.stop();
|
this.routineScheduler.stop();
|
||||||
@@ -1208,6 +1274,28 @@ export class InProcessRuntime
|
|||||||
return this.selfHealingManager;
|
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).
|
* Get the HeartbeatTriggerScheduler instance (if initialized).
|
||||||
* Returns undefined when agent monitoring is not available.
|
* Returns undefined when agent monitoring is not available.
|
||||||
|
|||||||
Reference in New Issue
Block a user