feat(KB-616): add ProjectRuntime abstraction for task execution isolation
- Define ProjectRuntime interface with unified task execution contract\n- Implement IPC protocol for host-worker communication (messages, streaming, heartbeats)\n- Add InProcessRuntime for synchronous in-process task execution\n- Add ChildProcessRuntime with isolated worker processes for sandboxed execution\n- Implement ProjectManager to coordinate runtime selection and task lifecycle\n- Update engine exports to expose runtime APIs\n- Add comprehensive tests for all runtime implementations and IPC protocol
This commit is contained in:
gsxdsm
@@ -14,3 +14,62 @@ export { PrCommentHandler } from "./pr-comment-handler.js";
|
||||
export { NtfyNotifier, type NtfyNotifierOptions } from "./notifier.js";
|
||||
export { CronRunner, type CronRunnerOptions } from "./cron-runner.js";
|
||||
export { StuckTaskDetector, type StuckTaskDetectorOptions, type DisposableSession } from "./stuck-task-detector.js";
|
||||
|
||||
// ── Project Runtime (Multi-Project Support) ────────────────────────────────
|
||||
|
||||
export {
|
||||
type ProjectRuntime,
|
||||
type ProjectRuntimeConfig,
|
||||
type RuntimeStatus,
|
||||
type RuntimeMetrics,
|
||||
type ProjectRuntimeEvents,
|
||||
type GlobalMetrics,
|
||||
} from "./project-runtime.js";
|
||||
|
||||
export { InProcessRuntime } from "./runtimes/in-process-runtime.js";
|
||||
export { ChildProcessRuntime } from "./runtimes/child-process-runtime.js";
|
||||
export { ProjectManager, type ProjectManagerEvents } from "./project-manager.js";
|
||||
|
||||
// ── IPC Protocol ───────────────────────────────────────────────────────
|
||||
|
||||
export {
|
||||
type IpcMessage,
|
||||
type IpcCommandType,
|
||||
type IpcResponseType,
|
||||
type IpcEventType,
|
||||
START_RUNTIME,
|
||||
STOP_RUNTIME,
|
||||
GET_STATUS,
|
||||
GET_METRICS,
|
||||
GET_TASK_STORE,
|
||||
GET_SCHEDULER,
|
||||
PING,
|
||||
OK,
|
||||
ERROR,
|
||||
PONG,
|
||||
TASK_CREATED,
|
||||
TASK_MOVED,
|
||||
TASK_UPDATED,
|
||||
ERROR_EVENT,
|
||||
HEALTH_CHANGED,
|
||||
type StartRuntimePayload,
|
||||
type StopRuntimePayload,
|
||||
type OkPayload,
|
||||
type ErrorPayload,
|
||||
type PongPayload,
|
||||
type TaskCreatedPayload,
|
||||
type TaskMovedPayload,
|
||||
type TaskUpdatedPayload,
|
||||
type ErrorEventPayload,
|
||||
type HealthChangedPayload,
|
||||
isIpcCommand,
|
||||
isIpcResponse,
|
||||
isIpcEvent,
|
||||
createCommand,
|
||||
createResponse,
|
||||
createEvent,
|
||||
generateCorrelationId,
|
||||
} from "./ipc/ipc-protocol.js";
|
||||
|
||||
export { IpcHost } from "./ipc/ipc-host.js";
|
||||
export { IpcWorker } from "./ipc/ipc-worker.js";
|
||||
|
||||
279
packages/engine/src/ipc/ipc-host.ts
Normal file
279
packages/engine/src/ipc/ipc-host.ts
Normal file
@@ -0,0 +1,279 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { ChildProcess } from "node:child_process";
|
||||
import type { IpcMessage, IpcCommandType, IpcResponseType } from "./ipc-protocol.js";
|
||||
import { OK, ERROR, PONG, generateCorrelationId } from "./ipc-protocol.js";
|
||||
import { ipcLog } from "../logger.js";
|
||||
|
||||
/**
|
||||
* Pending command waiting for a response.
|
||||
*/
|
||||
interface PendingCommand {
|
||||
resolve: (value: unknown) => void;
|
||||
reject: (error: Error) => void;
|
||||
timeout: ReturnType<typeof setTimeout>;
|
||||
type: IpcCommandType;
|
||||
}
|
||||
|
||||
/**
|
||||
* IPC Host handler for managing communication with a child process worker.
|
||||
*
|
||||
* Handles:
|
||||
* - Sending commands and correlating responses
|
||||
* - Forwarding events from worker to host listeners
|
||||
* - Detecting and handling IPC disconnections
|
||||
* - Timeout handling for commands
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* const child = fork(workerPath);
|
||||
* const ipcHost = new IpcHost(child);
|
||||
*
|
||||
* // Send command and await response
|
||||
* const response = await ipcHost.sendCommand("GET_STATUS", {});
|
||||
*
|
||||
* // Listen for events
|
||||
* ipcHost.on("TASK_CREATED", (payload) => {
|
||||
* console.log("Task created:", payload.task.id);
|
||||
* });
|
||||
* ```
|
||||
*/
|
||||
export class IpcHost extends EventEmitter {
|
||||
private pendingCommands = new Map<string, PendingCommand>();
|
||||
private commandTimeoutMs = 30000; // 30 second default timeout
|
||||
private disconnected = false;
|
||||
|
||||
/**
|
||||
* @param childProcess - The forked child process to communicate with
|
||||
* @param options - Optional configuration
|
||||
*/
|
||||
constructor(
|
||||
private childProcess: ChildProcess,
|
||||
options?: { commandTimeoutMs?: number }
|
||||
) {
|
||||
super();
|
||||
this.setMaxListeners(100);
|
||||
this.commandTimeoutMs = options?.commandTimeoutMs ?? 30000;
|
||||
this.setupListeners();
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up message and error listeners on the child process.
|
||||
*/
|
||||
private setupListeners(): void {
|
||||
// Handle incoming messages from child
|
||||
this.childProcess.on("message", (message: IpcMessage) => {
|
||||
this.handleMessage(message);
|
||||
});
|
||||
|
||||
// Handle child process errors
|
||||
this.childProcess.on("error", (error) => {
|
||||
ipcLog.error(`Child process error: ${error.message}`);
|
||||
this.handleDisconnection(error);
|
||||
});
|
||||
|
||||
// Handle child process exit
|
||||
this.childProcess.on("exit", (code, signal) => {
|
||||
const reason = signal
|
||||
? `Child process exited with signal ${signal}`
|
||||
: `Child process exited with code ${code}`;
|
||||
ipcLog.warn(reason);
|
||||
this.handleDisconnection(new Error(reason));
|
||||
});
|
||||
|
||||
// Handle IPC channel disconnection
|
||||
this.childProcess.on("disconnect", () => {
|
||||
ipcLog.warn("Child process IPC channel disconnected");
|
||||
this.handleDisconnection(new Error("IPC channel disconnected"));
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle an incoming message from the child process.
|
||||
*/
|
||||
private handleMessage(message: IpcMessage): void {
|
||||
// Validate message structure
|
||||
if (!this.isValidMessage(message)) {
|
||||
ipcLog.warn(`Received malformed IPC message: ${JSON.stringify(message)}`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Handle responses to pending commands
|
||||
if (message.type === OK || message.type === ERROR || message.type === PONG) {
|
||||
const pending = this.pendingCommands.get(message.id);
|
||||
if (pending) {
|
||||
this.handleResponse(message, pending);
|
||||
} else {
|
||||
ipcLog.warn(`Received response for unknown command ID: ${message.id}`);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Handle events (forward to listeners)
|
||||
this.emit(message.type, message.payload);
|
||||
this.emit("message", message); // Generic message event
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle a response message for a pending command.
|
||||
*/
|
||||
private handleResponse(
|
||||
message: IpcMessage,
|
||||
pending: PendingCommand
|
||||
): void {
|
||||
// Clear the timeout
|
||||
clearTimeout(pending.timeout);
|
||||
this.pendingCommands.delete(message.id);
|
||||
|
||||
if (message.type === OK) {
|
||||
pending.resolve((message.payload as { data?: unknown }).data);
|
||||
} else if (message.type === ERROR) {
|
||||
const errorPayload = message.payload as {
|
||||
message: string;
|
||||
code?: string;
|
||||
stack?: string;
|
||||
};
|
||||
const error = new Error(errorPayload.message);
|
||||
if (errorPayload.code) {
|
||||
(error as Error & { code: string }).code = errorPayload.code;
|
||||
}
|
||||
pending.reject(error);
|
||||
} else if (message.type === PONG) {
|
||||
pending.resolve(message.payload);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle disconnection from the child process.
|
||||
* Rejects all pending commands.
|
||||
*/
|
||||
private handleDisconnection(error: Error): void {
|
||||
if (this.disconnected) return;
|
||||
this.disconnected = true;
|
||||
|
||||
// Reject all pending commands
|
||||
for (const [id, pending] of this.pendingCommands) {
|
||||
clearTimeout(pending.timeout);
|
||||
pending.reject(new Error(`IPC disconnected: ${error.message}`));
|
||||
}
|
||||
this.pendingCommands.clear();
|
||||
|
||||
this.emit("disconnect", error);
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a command to the child process and await a response.
|
||||
*
|
||||
* @param type - Command type
|
||||
* @param payload - Command payload
|
||||
* @param timeoutMs - Optional timeout override
|
||||
* @returns Promise that resolves with the response data
|
||||
* @throws Error if the command times out or IPC disconnects
|
||||
*/
|
||||
async sendCommand<T = unknown>(
|
||||
type: IpcCommandType,
|
||||
payload: unknown,
|
||||
timeoutMs?: number
|
||||
): Promise<T> {
|
||||
if (this.disconnected) {
|
||||
throw new Error("Cannot send command: IPC channel disconnected");
|
||||
}
|
||||
|
||||
const id = generateCorrelationId();
|
||||
const message: IpcMessage = { type, id, payload };
|
||||
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
// Set up timeout
|
||||
const timeout = setTimeout(() => {
|
||||
this.pendingCommands.delete(id);
|
||||
reject(new Error(`Command ${type} timed out after ${timeoutMs ?? this.commandTimeoutMs}ms`));
|
||||
}, timeoutMs ?? this.commandTimeoutMs);
|
||||
|
||||
// Store pending command
|
||||
this.pendingCommands.set(id, {
|
||||
resolve: resolve as (value: unknown) => void,
|
||||
reject,
|
||||
timeout,
|
||||
type,
|
||||
});
|
||||
|
||||
// Send message to child
|
||||
try {
|
||||
if (!this.childProcess.send) {
|
||||
throw new Error("Child process does not have IPC channel");
|
||||
}
|
||||
this.childProcess.send(message, (err) => {
|
||||
if (err) {
|
||||
clearTimeout(timeout);
|
||||
this.pendingCommands.delete(id);
|
||||
reject(new Error(`Failed to send command: ${err.message}`));
|
||||
}
|
||||
});
|
||||
} catch (err) {
|
||||
clearTimeout(timeout);
|
||||
this.pendingCommands.delete(id);
|
||||
reject(err);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a ping to check if the child process is responsive.
|
||||
*
|
||||
* @param timeoutMs - Timeout for pong response (default: 5000)
|
||||
* @returns Promise that resolves with pong payload or rejects on timeout
|
||||
*/
|
||||
async ping(timeoutMs = 5000): Promise<{ timestamp: string }> {
|
||||
return this.sendCommand("PING", {}, timeoutMs);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if the IPC channel is connected.
|
||||
*/
|
||||
isConnected(): boolean {
|
||||
return !this.disconnected && this.childProcess.connected;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the underlying child process.
|
||||
*/
|
||||
getChildProcess(): ChildProcess {
|
||||
return this.childProcess;
|
||||
}
|
||||
|
||||
/**
|
||||
* Disconnect the IPC channel and clean up.
|
||||
*/
|
||||
disconnect(): void {
|
||||
this.handleDisconnection(new Error("Host initiated disconnect"));
|
||||
|
||||
if (this.childProcess.connected) {
|
||||
this.childProcess.disconnect();
|
||||
}
|
||||
|
||||
this.removeAllListeners();
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate that a message has the expected structure.
|
||||
*/
|
||||
private isValidMessage(message: unknown): message is IpcMessage {
|
||||
if (typeof message !== "object" || message === null) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const msg = message as Record<string, unknown>;
|
||||
|
||||
return (
|
||||
typeof msg.type === "string" &&
|
||||
typeof msg.id === "string" &&
|
||||
"payload" in msg
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of pending commands.
|
||||
*/
|
||||
getPendingCommandCount(): number {
|
||||
return this.pendingCommands.size;
|
||||
}
|
||||
}
|
||||
175
packages/engine/src/ipc/ipc-protocol.test.ts
Normal file
175
packages/engine/src/ipc/ipc-protocol.test.ts
Normal file
@@ -0,0 +1,175 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import {
|
||||
START_RUNTIME,
|
||||
STOP_RUNTIME,
|
||||
GET_STATUS,
|
||||
GET_METRICS,
|
||||
GET_TASK_STORE,
|
||||
GET_SCHEDULER,
|
||||
PING,
|
||||
OK,
|
||||
ERROR,
|
||||
PONG,
|
||||
TASK_CREATED,
|
||||
TASK_MOVED,
|
||||
TASK_UPDATED,
|
||||
ERROR_EVENT,
|
||||
HEALTH_CHANGED,
|
||||
isIpcCommand,
|
||||
isIpcResponse,
|
||||
isIpcEvent,
|
||||
createCommand,
|
||||
createResponse,
|
||||
createEvent,
|
||||
generateCorrelationId,
|
||||
} from "./ipc-protocol.js";
|
||||
|
||||
describe("IPC Protocol", () => {
|
||||
describe("constants", () => {
|
||||
it("should export all command types", () => {
|
||||
expect(START_RUNTIME).toBe("START_RUNTIME");
|
||||
expect(STOP_RUNTIME).toBe("STOP_RUNTIME");
|
||||
expect(GET_STATUS).toBe("GET_STATUS");
|
||||
expect(GET_METRICS).toBe("GET_METRICS");
|
||||
expect(GET_TASK_STORE).toBe("GET_TASK_STORE");
|
||||
expect(GET_SCHEDULER).toBe("GET_SCHEDULER");
|
||||
expect(PING).toBe("PING");
|
||||
});
|
||||
|
||||
it("should export all response types", () => {
|
||||
expect(OK).toBe("OK");
|
||||
expect(ERROR).toBe("ERROR");
|
||||
expect(PONG).toBe("PONG");
|
||||
});
|
||||
|
||||
it("should export all event types", () => {
|
||||
expect(TASK_CREATED).toBe("TASK_CREATED");
|
||||
expect(TASK_MOVED).toBe("TASK_MOVED");
|
||||
expect(TASK_UPDATED).toBe("TASK_UPDATED");
|
||||
expect(ERROR_EVENT).toBe("ERROR_EVENT");
|
||||
expect(HEALTH_CHANGED).toBe("HEALTH_CHANGED");
|
||||
});
|
||||
|
||||
it("should have distinct ERROR and ERROR_EVENT values", () => {
|
||||
expect(ERROR).toBe("ERROR");
|
||||
expect(ERROR_EVENT).toBe("ERROR_EVENT");
|
||||
expect(ERROR).not.toBe(ERROR_EVENT);
|
||||
});
|
||||
});
|
||||
|
||||
describe("isIpcCommand", () => {
|
||||
it("should return true for command types", () => {
|
||||
expect(isIpcCommand({ type: START_RUNTIME, id: "1", payload: {} })).toBe(true);
|
||||
expect(isIpcCommand({ type: STOP_RUNTIME, id: "1", payload: {} })).toBe(true);
|
||||
expect(isIpcCommand({ type: GET_STATUS, id: "1", payload: {} })).toBe(true);
|
||||
expect(isIpcCommand({ type: PING, id: "1", payload: {} })).toBe(true);
|
||||
});
|
||||
|
||||
it("should return false for response types", () => {
|
||||
expect(isIpcCommand({ type: OK, id: "1", payload: {} })).toBe(false);
|
||||
expect(isIpcCommand({ type: ERROR, id: "1", payload: {} })).toBe(false);
|
||||
expect(isIpcCommand({ type: PONG, id: "1", payload: {} })).toBe(false);
|
||||
});
|
||||
|
||||
it("should return false for event types", () => {
|
||||
expect(isIpcCommand({ type: TASK_CREATED, id: "1", payload: {} })).toBe(false);
|
||||
expect(isIpcCommand({ type: HEALTH_CHANGED, id: "1", payload: {} })).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("isIpcResponse", () => {
|
||||
it("should return true for response types", () => {
|
||||
expect(isIpcResponse({ type: OK, id: "1", payload: {} })).toBe(true);
|
||||
expect(isIpcResponse({ type: ERROR, id: "1", payload: {} })).toBe(true);
|
||||
expect(isIpcResponse({ type: PONG, id: "1", payload: {} })).toBe(true);
|
||||
});
|
||||
|
||||
it("should return false for command types", () => {
|
||||
expect(isIpcResponse({ type: START_RUNTIME, id: "1", payload: {} })).toBe(false);
|
||||
expect(isIpcResponse({ type: PING, id: "1", payload: {} })).toBe(false);
|
||||
});
|
||||
|
||||
it("should return false for event types", () => {
|
||||
expect(isIpcResponse({ type: TASK_CREATED, id: "1", payload: {} })).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("isIpcEvent", () => {
|
||||
it("should return true for event types", () => {
|
||||
expect(isIpcEvent({ type: TASK_CREATED, id: "1", payload: {} })).toBe(true);
|
||||
expect(isIpcEvent({ type: TASK_MOVED, id: "1", payload: {} })).toBe(true);
|
||||
expect(isIpcEvent({ type: TASK_UPDATED, id: "1", payload: {} })).toBe(true);
|
||||
expect(isIpcEvent({ type: ERROR_EVENT, id: "1", payload: {} })).toBe(true);
|
||||
expect(isIpcEvent({ type: HEALTH_CHANGED, id: "1", payload: {} })).toBe(true);
|
||||
});
|
||||
|
||||
it("should return false for command types", () => {
|
||||
expect(isIpcEvent({ type: START_RUNTIME, id: "1", payload: {} })).toBe(false);
|
||||
expect(isIpcEvent({ type: PING, id: "1", payload: {} })).toBe(false);
|
||||
});
|
||||
|
||||
it("should return false for response types", () => {
|
||||
expect(isIpcEvent({ type: OK, id: "1", payload: {} })).toBe(false);
|
||||
expect(isIpcEvent({ type: ERROR, id: "1", payload: {} })).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("createCommand", () => {
|
||||
it("should create a command message", () => {
|
||||
const payload = { config: { projectId: "test" } };
|
||||
const message = createCommand(START_RUNTIME, "cmd-1", payload);
|
||||
|
||||
expect(message).toEqual({
|
||||
type: START_RUNTIME,
|
||||
id: "cmd-1",
|
||||
payload,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("createResponse", () => {
|
||||
it("should create a response message", () => {
|
||||
const payload = { data: { status: "active" } };
|
||||
const message = createResponse(OK, "cmd-1", payload);
|
||||
|
||||
expect(message).toEqual({
|
||||
type: OK,
|
||||
id: "cmd-1",
|
||||
payload,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("createEvent", () => {
|
||||
it("should create an event message", () => {
|
||||
const payload = { task: { id: "KB-001" } };
|
||||
const message = createEvent(TASK_CREATED, "evt-1", payload);
|
||||
|
||||
expect(message).toEqual({
|
||||
type: TASK_CREATED,
|
||||
id: "evt-1",
|
||||
payload,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("generateCorrelationId", () => {
|
||||
it("should generate unique IDs", () => {
|
||||
const id1 = generateCorrelationId();
|
||||
const id2 = generateCorrelationId();
|
||||
|
||||
expect(id1).toBeDefined();
|
||||
expect(id2).toBeDefined();
|
||||
expect(id1).not.toBe(id2);
|
||||
});
|
||||
|
||||
it("should generate string IDs with timestamp and random parts", () => {
|
||||
const id = generateCorrelationId();
|
||||
const parts = id.split("-");
|
||||
|
||||
expect(parts.length).toBeGreaterThanOrEqual(2);
|
||||
// First part should be a timestamp (number)
|
||||
expect(Number.parseInt(parts[0], 10)).not.toBeNaN();
|
||||
});
|
||||
});
|
||||
});
|
||||
277
packages/engine/src/ipc/ipc-protocol.ts
Normal file
277
packages/engine/src/ipc/ipc-protocol.ts
Normal file
@@ -0,0 +1,277 @@
|
||||
/**
|
||||
* IPC Protocol for child-process isolation mode.
|
||||
*
|
||||
* Defines message types for communication between the host (ProjectManager)
|
||||
* and worker (child process running InProcessRuntime internally).
|
||||
*
|
||||
* Message Flow:
|
||||
* 1. Host sends commands with unique correlation IDs
|
||||
* 2. Worker processes commands and sends responses with matching IDs
|
||||
* 3. Worker can also send events unsolicited (task events, health changes)
|
||||
*/
|
||||
|
||||
import type { RuntimeStatus, RuntimeMetrics, ProjectRuntimeConfig } from "../project-runtime.js";
|
||||
import type { Task, TaskStore } from "@fusion/core";
|
||||
import type { Scheduler } from "../scheduler.js";
|
||||
|
||||
// ── Base Message Types ────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Base interface for all IPC messages.
|
||||
*/
|
||||
export interface IpcMessage {
|
||||
/** Message type discriminator */
|
||||
type: string;
|
||||
/** Unique correlation ID for request/response matching */
|
||||
id: string;
|
||||
/** Message payload */
|
||||
payload: unknown;
|
||||
}
|
||||
|
||||
// ── Command Types (Host → Worker) ───────────────────────────────────────────
|
||||
|
||||
/** Command type: Start the runtime */
|
||||
export const START_RUNTIME = "START_RUNTIME" as const;
|
||||
/** Command type: Stop the runtime */
|
||||
export const STOP_RUNTIME = "STOP_RUNTIME" as const;
|
||||
/** Command type: Get current status */
|
||||
export const GET_STATUS = "GET_STATUS" as const;
|
||||
/** Command type: Get runtime metrics */
|
||||
export const GET_METRICS = "GET_METRICS" as const;
|
||||
/** Command type: Get TaskStore reference (returns serialized state) */
|
||||
export const GET_TASK_STORE = "GET_TASK_STORE" as const;
|
||||
/** Command type: Get Scheduler reference (returns serialized state) */
|
||||
export const GET_SCHEDULER = "GET_SCHEDULER" as const;
|
||||
/** Command type: Ping for health check */
|
||||
export const PING = "PING" as const;
|
||||
|
||||
/**
|
||||
* Union of all command types.
|
||||
*/
|
||||
export type IpcCommandType =
|
||||
| typeof START_RUNTIME
|
||||
| typeof STOP_RUNTIME
|
||||
| typeof GET_STATUS
|
||||
| typeof GET_METRICS
|
||||
| typeof GET_TASK_STORE
|
||||
| typeof GET_SCHEDULER
|
||||
| typeof PING;
|
||||
|
||||
/**
|
||||
* Payload for START_RUNTIME command.
|
||||
*/
|
||||
export interface StartRuntimePayload {
|
||||
config: ProjectRuntimeConfig;
|
||||
}
|
||||
|
||||
/**
|
||||
* Payload for STOP_RUNTIME command.
|
||||
*/
|
||||
export interface StopRuntimePayload {
|
||||
/** Timeout in milliseconds for graceful shutdown (default: 30000) */
|
||||
timeoutMs?: number;
|
||||
}
|
||||
|
||||
// ── Response Types (Worker → Host) ──────────────────────────────────────
|
||||
|
||||
/** Response type: Success */
|
||||
export const OK = "OK" as const;
|
||||
/** Response type: Error */
|
||||
export const ERROR = "ERROR" as const;
|
||||
/** Response type: Pong (ping reply) */
|
||||
export const PONG = "PONG" as const;
|
||||
|
||||
/**
|
||||
* Union of all response types.
|
||||
*/
|
||||
export type IpcResponseType = typeof OK | typeof ERROR | typeof PONG;
|
||||
|
||||
/**
|
||||
* Successful response payload.
|
||||
*/
|
||||
export interface OkPayload {
|
||||
/** Response data (type depends on the command) */
|
||||
data?: unknown;
|
||||
}
|
||||
|
||||
/**
|
||||
* Error response payload.
|
||||
*/
|
||||
export interface ErrorPayload {
|
||||
/** Error message */
|
||||
message: string;
|
||||
/** Error code for programmatic handling */
|
||||
code?: string;
|
||||
/** Stack trace (only in development) */
|
||||
stack?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pong response payload for health checks.
|
||||
*/
|
||||
export interface PongPayload {
|
||||
/** ISO-8601 timestamp from the worker */
|
||||
timestamp: string;
|
||||
}
|
||||
|
||||
// ── Event Types (Worker → Host, unsolicited) ─────────────────────────────
|
||||
|
||||
/** Event type: Task created */
|
||||
export const TASK_CREATED = "TASK_CREATED" as const;
|
||||
/** Event type: Task moved */
|
||||
export const TASK_MOVED = "TASK_MOVED" as const;
|
||||
/** Event type: Task updated */
|
||||
export const TASK_UPDATED = "TASK_UPDATED" as const;
|
||||
/** Event type: Runtime error */
|
||||
export const ERROR_EVENT = "ERROR_EVENT" as const;
|
||||
/** Event type: Health status changed */
|
||||
export const HEALTH_CHANGED = "HEALTH_CHANGED" as const;
|
||||
|
||||
/**
|
||||
* Union of all event types.
|
||||
*/
|
||||
export type IpcEventType =
|
||||
| typeof TASK_CREATED
|
||||
| typeof TASK_MOVED
|
||||
| typeof TASK_UPDATED
|
||||
| typeof ERROR_EVENT
|
||||
| typeof HEALTH_CHANGED;
|
||||
|
||||
/**
|
||||
* Payload for TASK_CREATED event.
|
||||
*/
|
||||
export interface TaskCreatedPayload {
|
||||
task: Task;
|
||||
}
|
||||
|
||||
/**
|
||||
* Payload for TASK_MOVED event.
|
||||
*/
|
||||
export interface TaskMovedPayload {
|
||||
task: Task;
|
||||
from: string;
|
||||
to: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Payload for TASK_UPDATED event.
|
||||
*/
|
||||
export interface TaskUpdatedPayload {
|
||||
task: Task;
|
||||
}
|
||||
|
||||
/**
|
||||
* Payload for ERROR event.
|
||||
*/
|
||||
export interface ErrorEventPayload {
|
||||
message: string;
|
||||
code?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Payload for HEALTH_CHANGED event.
|
||||
*/
|
||||
export interface HealthChangedPayload {
|
||||
status: RuntimeStatus;
|
||||
previous: RuntimeStatus;
|
||||
}
|
||||
|
||||
// ── Type Guards ───────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Check if a message is a command.
|
||||
*/
|
||||
export function isIpcCommand(message: IpcMessage): boolean {
|
||||
const commandTypes: IpcCommandType[] = [
|
||||
START_RUNTIME,
|
||||
STOP_RUNTIME,
|
||||
GET_STATUS,
|
||||
GET_METRICS,
|
||||
GET_TASK_STORE,
|
||||
GET_SCHEDULER,
|
||||
PING,
|
||||
];
|
||||
return commandTypes.includes(message.type as IpcCommandType);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a message is a response.
|
||||
*/
|
||||
export function isIpcResponse(message: IpcMessage): boolean {
|
||||
const responseTypes: IpcResponseType[] = [OK, ERROR, PONG];
|
||||
return responseTypes.includes(message.type as IpcResponseType);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a message is an event.
|
||||
*/
|
||||
export function isIpcEvent(message: IpcMessage): boolean {
|
||||
const eventTypes: IpcEventType[] = [
|
||||
TASK_CREATED,
|
||||
TASK_MOVED,
|
||||
TASK_UPDATED,
|
||||
ERROR_EVENT,
|
||||
HEALTH_CHANGED,
|
||||
];
|
||||
return eventTypes.includes(message.type as IpcEventType);
|
||||
}
|
||||
|
||||
// ── Serialized State Types ────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Serialized TaskStore state (for IPC transfer).
|
||||
* Full TaskStore objects cannot be passed across process boundaries.
|
||||
*/
|
||||
export interface SerializedTaskStore {
|
||||
rootDir: string;
|
||||
taskCount: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Serialized Scheduler state (for IPC transfer).
|
||||
*/
|
||||
export interface SerializedScheduler {
|
||||
running: boolean;
|
||||
}
|
||||
|
||||
// ── Helper Functions ────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Create a command message.
|
||||
*/
|
||||
export function createCommand<T>(
|
||||
type: IpcCommandType,
|
||||
id: string,
|
||||
payload: T
|
||||
): IpcMessage {
|
||||
return { type, id, payload };
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a response message.
|
||||
*/
|
||||
export function createResponse<T>(
|
||||
type: IpcResponseType,
|
||||
id: string,
|
||||
payload: T
|
||||
): IpcMessage {
|
||||
return { type, id, payload };
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an event message.
|
||||
*/
|
||||
export function createEvent<T>(
|
||||
type: IpcEventType,
|
||||
id: string,
|
||||
payload: T
|
||||
): IpcMessage {
|
||||
return { type, id, payload };
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate a unique correlation ID.
|
||||
*/
|
||||
export function generateCorrelationId(): string {
|
||||
return `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`;
|
||||
}
|
||||
325
packages/engine/src/ipc/ipc-worker.ts
Normal file
325
packages/engine/src/ipc/ipc-worker.ts
Normal file
@@ -0,0 +1,325 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { IpcMessage, IpcCommandType, IpcEventType } from "./ipc-protocol.js";
|
||||
import {
|
||||
OK,
|
||||
ERROR,
|
||||
PONG,
|
||||
START_RUNTIME,
|
||||
STOP_RUNTIME,
|
||||
GET_STATUS,
|
||||
GET_METRICS,
|
||||
GET_TASK_STORE,
|
||||
GET_SCHEDULER,
|
||||
PING,
|
||||
ERROR_EVENT,
|
||||
isIpcCommand,
|
||||
generateCorrelationId,
|
||||
} from "./ipc-protocol.js";
|
||||
import { ipcLog } from "../logger.js";
|
||||
|
||||
type CommandHandler = (payload: unknown) => Promise<unknown> | unknown;
|
||||
|
||||
/**
|
||||
* IPC Worker handler for managing communication from a child process to the host.
|
||||
*
|
||||
* Handles:
|
||||
* - Receiving commands from the host
|
||||
* - Sending responses back to the host
|
||||
* - Sending events to the host
|
||||
* - Graceful shutdown signal handling
|
||||
*
|
||||
* This class is designed to run inside a forked child process.
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* // In child-process-worker.ts
|
||||
* if (process.send) {
|
||||
* const ipcWorker = new IpcWorker();
|
||||
*
|
||||
* // Register command handlers
|
||||
* ipcWorker.onCommand("START_RUNTIME", async (payload) => {
|
||||
* // ... start runtime
|
||||
* return { success: true };
|
||||
* });
|
||||
*
|
||||
* // Send events to host
|
||||
* ipcWorker.sendEvent("TASK_CREATED", { task });
|
||||
*
|
||||
* // Handle graceful shutdown
|
||||
* process.on("SIGTERM", () => {
|
||||
* ipcWorker.shutdown();
|
||||
* });
|
||||
* }
|
||||
* ```
|
||||
*/
|
||||
export class IpcWorker extends EventEmitter {
|
||||
private commandHandlers = new Map<IpcCommandType, CommandHandler>();
|
||||
private shuttingDown = false;
|
||||
|
||||
/**
|
||||
* Create an IpcWorker instance.
|
||||
* Throws if not running in a forked child process (process.send unavailable).
|
||||
*/
|
||||
constructor() {
|
||||
super();
|
||||
this.setMaxListeners(100);
|
||||
|
||||
// Verify we're running in an IPC context
|
||||
if (!process.send) {
|
||||
throw new Error(
|
||||
"IpcWorker can only be instantiated in a forked child process (process.send unavailable)"
|
||||
);
|
||||
}
|
||||
|
||||
this.setupListeners();
|
||||
this.setupSignalHandlers();
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up message listeners on the process.
|
||||
*/
|
||||
private setupListeners(): void {
|
||||
process.on("message", (message: IpcMessage) => {
|
||||
this.handleMessage(message);
|
||||
});
|
||||
|
||||
// Handle disconnect from parent
|
||||
process.on("disconnect", () => {
|
||||
ipcLog.warn("IPC channel disconnected from parent");
|
||||
this.emit("disconnect");
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up signal handlers for graceful shutdown.
|
||||
*/
|
||||
private setupSignalHandlers(): void {
|
||||
const handleSignal = (signal: string) => {
|
||||
ipcLog.log(`Received ${signal}, initiating graceful shutdown...`);
|
||||
this.emit("shutdown", signal);
|
||||
|
||||
// Give time for cleanup before exiting
|
||||
setTimeout(() => {
|
||||
process.exit(0);
|
||||
}, 5000);
|
||||
};
|
||||
|
||||
process.on("SIGTERM", () => handleSignal("SIGTERM"));
|
||||
process.on("SIGINT", () => handleSignal("SIGINT"));
|
||||
|
||||
// Handle uncaught errors
|
||||
process.on("uncaughtException", (error) => {
|
||||
ipcLog.error(`Uncaught exception: ${error.message}`);
|
||||
this.sendErrorEvent(error);
|
||||
this.emit("error", error);
|
||||
|
||||
// Give time for error to be sent before exiting
|
||||
setTimeout(() => {
|
||||
process.exit(1);
|
||||
}, 1000);
|
||||
});
|
||||
|
||||
process.on("unhandledRejection", (reason) => {
|
||||
const error = reason instanceof Error ? reason : new Error(String(reason));
|
||||
ipcLog.error(`Unhandled rejection: ${error.message}`);
|
||||
this.sendErrorEvent(error);
|
||||
this.emit("error", error);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle an incoming message from the parent process.
|
||||
*/
|
||||
private async handleMessage(message: unknown): Promise<void> {
|
||||
// Validate message structure first
|
||||
if (!this.isValidMessage(message)) {
|
||||
ipcLog.warn(`Received malformed IPC message: ${JSON.stringify(message)}`);
|
||||
// Use a generated ID since we can't trust the message structure
|
||||
this.sendResponse(ERROR, generateCorrelationId(), {
|
||||
message: "Malformed message received",
|
||||
code: "MALFORMED_MESSAGE",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Now TypeScript knows message is IpcMessage
|
||||
// Check if this is a command
|
||||
if (!isIpcCommand(message)) {
|
||||
// Unknown message type
|
||||
this.sendResponse(ERROR, message.id, {
|
||||
message: `Unknown command type: ${message.type}`,
|
||||
code: "UNKNOWN_COMMAND",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Handle PING specially (no handler registration needed)
|
||||
if (message.type === PING) {
|
||||
this.sendResponse(PONG, message.id, {
|
||||
timestamp: new Date().toISOString(),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Look up command handler
|
||||
const handler = this.commandHandlers.get(message.type as IpcCommandType);
|
||||
if (!handler) {
|
||||
this.sendResponse(ERROR, message.id, {
|
||||
message: `No handler registered for command: ${message.type}`,
|
||||
code: "NO_HANDLER",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Execute handler and send response
|
||||
try {
|
||||
const result = await handler(message.payload);
|
||||
this.sendResponse(OK, message.id, { data: result });
|
||||
} catch (error) {
|
||||
const err = error instanceof Error ? error : new Error(String(error));
|
||||
ipcLog.error(`Error handling command ${message.type}: ${err.message}`);
|
||||
this.sendResponse(ERROR, message.id, {
|
||||
message: err.message,
|
||||
code: (err as Error & { code?: string }).code ?? "HANDLER_ERROR",
|
||||
stack: process.env.NODE_ENV === "development" ? err.stack : undefined,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a command handler.
|
||||
*
|
||||
* @param type - Command type to handle
|
||||
* @param handler - Handler function that receives the payload and returns result
|
||||
*/
|
||||
onCommand(type: IpcCommandType, handler: CommandHandler): void {
|
||||
this.commandHandlers.set(type, handler);
|
||||
ipcLog.log(`Registered handler for command: ${type}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Unregister a command handler.
|
||||
*
|
||||
* @param type - Command type to unregister
|
||||
*/
|
||||
offCommand(type: IpcCommandType): void {
|
||||
this.commandHandlers.delete(type);
|
||||
ipcLog.log(`Unregistered handler for command: ${type}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a response to the parent process.
|
||||
*
|
||||
* @param type - Response type (OK, ERROR, or PONG)
|
||||
* @param id - Correlation ID matching the command
|
||||
* @param payload - Response payload
|
||||
*/
|
||||
sendResponse(type: IpcResponseType, id: string, payload: unknown): void {
|
||||
if (this.shuttingDown) return;
|
||||
|
||||
const message: IpcMessage = { type, id, payload };
|
||||
this.send(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* Send an event to the parent process.
|
||||
*
|
||||
* @param type - Event type
|
||||
* @param payload - Event payload
|
||||
*/
|
||||
sendEvent(type: IpcEventType, payload: unknown): void {
|
||||
if (this.shuttingDown) return;
|
||||
|
||||
const message: IpcMessage = {
|
||||
type,
|
||||
id: generateCorrelationId(),
|
||||
payload,
|
||||
};
|
||||
this.send(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* Send an error event to the parent process.
|
||||
*
|
||||
* @param error - Error to send
|
||||
*/
|
||||
sendErrorEvent(error: Error): void {
|
||||
this.sendEvent(ERROR_EVENT, {
|
||||
message: error.message,
|
||||
code: (error as Error & { code?: string }).code,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a message to the parent process.
|
||||
*/
|
||||
private send(message: IpcMessage): void {
|
||||
if (!process.send) {
|
||||
ipcLog.error("Cannot send message: process.send unavailable");
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
process.send(message, (err) => {
|
||||
if (err) {
|
||||
ipcLog.error(`Failed to send message: ${err.message}`);
|
||||
}
|
||||
});
|
||||
} catch (err) {
|
||||
ipcLog.error(`Error sending message: ${(err as Error).message}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initiate graceful shutdown.
|
||||
* Notifies the parent and prevents further message sending.
|
||||
*/
|
||||
shutdown(): void {
|
||||
if (this.shuttingDown) return;
|
||||
this.shuttingDown = true;
|
||||
|
||||
ipcLog.log("IPC worker shutting down...");
|
||||
this.emit("shutdown");
|
||||
|
||||
// Notify parent we're shutting down
|
||||
try {
|
||||
process.send?.({ type: "SHUTDOWN", id: generateCorrelationId(), payload: {} });
|
||||
} catch {
|
||||
// Ignore errors during shutdown
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if the worker is in shutdown mode.
|
||||
*/
|
||||
isShuttingDown(): boolean {
|
||||
return this.shuttingDown;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of registered command handlers.
|
||||
*/
|
||||
getHandlerCount(): number {
|
||||
return this.commandHandlers.size;
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate that a message has the expected structure.
|
||||
*/
|
||||
private isValidMessage(message: unknown): message is IpcMessage {
|
||||
if (typeof message !== "object" || message === null) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const msg = message as Record<string, unknown>;
|
||||
|
||||
return (
|
||||
typeof msg.type === "string" &&
|
||||
typeof msg.id === "string" &&
|
||||
"payload" in msg
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Import types for type checking
|
||||
import type { IpcResponseType } from "./ipc-protocol.js";
|
||||
@@ -64,3 +64,12 @@ export const reviewerLog = createLogger("reviewer");
|
||||
|
||||
/** Logger for the PR monitor subsystem. */
|
||||
export const prMonitorLog = createLogger("pr-monitor");
|
||||
|
||||
/** Logger for the project runtime subsystem. */
|
||||
export const runtimeLog = createLogger("runtime");
|
||||
|
||||
/** Logger for the IPC subsystem. */
|
||||
export const ipcLog = createLogger("ipc");
|
||||
|
||||
/** Logger for the project manager subsystem. */
|
||||
export const projectManagerLog = createLogger("project-manager");
|
||||
|
||||
290
packages/engine/src/project-manager.test.ts
Normal file
290
packages/engine/src/project-manager.test.ts
Normal file
@@ -0,0 +1,290 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import type { CentralCore, RegisteredProject, Task } from "@fusion/core";
|
||||
import { ProjectManager } from "./project-manager.js";
|
||||
import type { ProjectRuntimeConfig } from "./project-runtime.js";
|
||||
|
||||
// Mock the runtimes
|
||||
vi.mock("./runtimes/in-process-runtime.js", () => ({
|
||||
InProcessRuntime: vi.fn().mockImplementation(() => ({
|
||||
start: vi.fn().mockResolvedValue(undefined),
|
||||
stop: vi.fn().mockResolvedValue(undefined),
|
||||
getStatus: vi.fn().mockReturnValue("active"),
|
||||
getTaskStore: vi.fn(),
|
||||
getScheduler: vi.fn(),
|
||||
getMetrics: vi.fn().mockReturnValue({
|
||||
inFlightTasks: 0,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
}),
|
||||
on: vi.fn().mockReturnThis(),
|
||||
})),
|
||||
}));
|
||||
|
||||
vi.mock("./runtimes/child-process-runtime.js", () => ({
|
||||
ChildProcessRuntime: vi.fn().mockImplementation(() => ({
|
||||
start: vi.fn().mockResolvedValue(undefined),
|
||||
stop: vi.fn().mockResolvedValue(undefined),
|
||||
getStatus: vi.fn().mockReturnValue("active"),
|
||||
getTaskStore: vi.fn().mockImplementation(() => {
|
||||
throw new Error("Not accessible in child mode");
|
||||
}),
|
||||
getScheduler: vi.fn().mockImplementation(() => {
|
||||
throw new Error("Not accessible in child mode");
|
||||
}),
|
||||
getMetrics: vi.fn().mockReturnValue({
|
||||
inFlightTasks: 0,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
}),
|
||||
on: vi.fn().mockReturnThis(),
|
||||
})),
|
||||
}));
|
||||
|
||||
describe("ProjectManager", () => {
|
||||
let manager: ProjectManager;
|
||||
let mockCentralCore: CentralCore;
|
||||
const mockProject: RegisteredProject = {
|
||||
id: "proj_test123",
|
||||
name: "Test Project",
|
||||
path: "/tmp/test-project",
|
||||
status: "initializing",
|
||||
isolationMode: "in-process",
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
mockCentralCore = {
|
||||
getProject: vi.fn().mockResolvedValue(mockProject),
|
||||
getGlobalConcurrencyState: vi.fn().mockResolvedValue({
|
||||
globalMaxConcurrent: 4,
|
||||
currentlyActive: 0,
|
||||
queuedCount: 0,
|
||||
projectsActive: {},
|
||||
}),
|
||||
updateProjectHealth: vi.fn().mockResolvedValue(undefined),
|
||||
logActivity: vi.fn().mockResolvedValue(undefined),
|
||||
acquireGlobalSlot: vi.fn().mockResolvedValue(true),
|
||||
releaseGlobalSlot: vi.fn().mockResolvedValue(undefined),
|
||||
} as unknown as CentralCore;
|
||||
|
||||
manager = new ProjectManager(mockCentralCore);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
await manager.stopAll();
|
||||
} catch {
|
||||
// Ignore errors during cleanup
|
||||
}
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
describe("initialization", () => {
|
||||
it("should initialize with empty runtimes", () => {
|
||||
expect(manager.listRuntimes()).toHaveLength(0);
|
||||
expect(manager.getProjectIds()).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("should get global metrics with empty runtimes", async () => {
|
||||
const metrics = await manager.getGlobalMetrics();
|
||||
expect(metrics.totalRuntimes).toBe(0);
|
||||
expect(metrics.totalInFlightTasks).toBe(0);
|
||||
expect(metrics.totalActiveAgents).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("addProject", () => {
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_test123",
|
||||
workingDirectory: "/tmp/test-project",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
it("should throw if project not found in CentralCore", async () => {
|
||||
(mockCentralCore.getProject as ReturnType<typeof vi.fn>).mockResolvedValue(null);
|
||||
|
||||
await expect(manager.addProject(testConfig)).rejects.toThrow(
|
||||
"not found in CentralCore"
|
||||
);
|
||||
});
|
||||
|
||||
it("should throw if runtime already exists", async () => {
|
||||
await manager.addProject(testConfig);
|
||||
|
||||
await expect(manager.addProject(testConfig)).rejects.toThrow(
|
||||
"Runtime already exists"
|
||||
);
|
||||
});
|
||||
|
||||
it("should call logActivity after adding project", async () => {
|
||||
await manager.addProject(testConfig);
|
||||
|
||||
expect(mockCentralCore.logActivity).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should update project health after adding", async () => {
|
||||
await manager.addProject(testConfig);
|
||||
|
||||
expect(mockCentralCore.updateProjectHealth).toHaveBeenCalledWith(
|
||||
"proj_test123",
|
||||
expect.objectContaining({ status: "active" })
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("removeProject", () => {
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_test123",
|
||||
workingDirectory: "/tmp/test-project",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
it("should throw if runtime not found", async () => {
|
||||
await expect(manager.removeProject("non-existent")).rejects.toThrow(
|
||||
"Runtime not found"
|
||||
);
|
||||
});
|
||||
|
||||
it("should remove runtime after adding", async () => {
|
||||
await manager.addProject(testConfig);
|
||||
expect(manager.listRuntimes()).toHaveLength(1);
|
||||
|
||||
await manager.removeProject("proj_test123");
|
||||
expect(manager.listRuntimes()).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("should update project health after removing", async () => {
|
||||
await manager.addProject(testConfig);
|
||||
await manager.removeProject("proj_test123");
|
||||
|
||||
expect(mockCentralCore.updateProjectHealth).toHaveBeenCalledWith(
|
||||
"proj_test123",
|
||||
expect.objectContaining({ status: "paused" })
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getRuntime", () => {
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_test123",
|
||||
workingDirectory: "/tmp/test-project",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
it("should return undefined for non-existent runtime", () => {
|
||||
expect(manager.getRuntime("non-existent")).toBeUndefined();
|
||||
});
|
||||
|
||||
it("should return runtime after adding", async () => {
|
||||
await manager.addProject(testConfig);
|
||||
const runtime = manager.getRuntime("proj_test123");
|
||||
|
||||
expect(runtime).toBeDefined();
|
||||
expect(runtime?.getStatus()).toBe("active");
|
||||
});
|
||||
});
|
||||
|
||||
describe("global slots", () => {
|
||||
it("should acquire global slot", async () => {
|
||||
const acquired = await manager.acquireGlobalSlot("proj_test123");
|
||||
|
||||
expect(acquired).toBe(true);
|
||||
expect(mockCentralCore.acquireGlobalSlot).toHaveBeenCalledWith("proj_test123");
|
||||
});
|
||||
|
||||
it("should release global slot", async () => {
|
||||
await manager.releaseGlobalSlot("proj_test123");
|
||||
|
||||
expect(mockCentralCore.releaseGlobalSlot).toHaveBeenCalledWith("proj_test123");
|
||||
});
|
||||
|
||||
it("should handle acquire failure gracefully", async () => {
|
||||
(mockCentralCore.acquireGlobalSlot as ReturnType<typeof vi.fn>).mockRejectedValue(
|
||||
new Error("Slot unavailable")
|
||||
);
|
||||
|
||||
const acquired = await manager.acquireGlobalSlot("proj_test123");
|
||||
|
||||
expect(acquired).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("event forwarding", () => {
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_test123",
|
||||
workingDirectory: "/tmp/test-project",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
it("should support runtime:added event", async () => {
|
||||
const handler = vi.fn();
|
||||
manager.on("runtime:added", handler);
|
||||
|
||||
await manager.addProject(testConfig);
|
||||
|
||||
expect(handler).toHaveBeenCalledWith({
|
||||
projectId: "proj_test123",
|
||||
projectName: "Test Project",
|
||||
});
|
||||
});
|
||||
|
||||
it("should support runtime:removed event", async () => {
|
||||
const handler = vi.fn();
|
||||
manager.on("runtime:removed", handler);
|
||||
|
||||
await manager.addProject(testConfig);
|
||||
await manager.removeProject("proj_test123");
|
||||
|
||||
expect(handler).toHaveBeenCalledWith({
|
||||
projectId: "proj_test123",
|
||||
projectName: "Test Project",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("stopAll", () => {
|
||||
it("should stop all runtimes", async () => {
|
||||
const config1: ProjectRuntimeConfig = {
|
||||
projectId: "proj_1",
|
||||
workingDirectory: "/tmp/project1",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
const config2: ProjectRuntimeConfig = {
|
||||
projectId: "proj_2",
|
||||
workingDirectory: "/tmp/project2",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
(mockCentralCore.getProject as ReturnType<typeof vi.fn>).mockImplementation(
|
||||
(id: string) =>
|
||||
Promise.resolve({
|
||||
...mockProject,
|
||||
id,
|
||||
name: `Project ${id}`,
|
||||
})
|
||||
);
|
||||
|
||||
await manager.addProject(config1);
|
||||
await manager.addProject(config2);
|
||||
|
||||
expect(manager.listRuntimes()).toHaveLength(2);
|
||||
|
||||
await manager.stopAll();
|
||||
|
||||
expect(manager.listRuntimes()).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
});
|
||||
447
packages/engine/src/project-manager.ts
Normal file
447
packages/engine/src/project-manager.ts
Normal file
@@ -0,0 +1,447 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { Task, CentralCore, RegisteredProject } from "@fusion/core";
|
||||
import { InProcessRuntime } from "./runtimes/in-process-runtime.js";
|
||||
import { ChildProcessRuntime } from "./runtimes/child-process-runtime.js";
|
||||
import { AgentSemaphore } from "./concurrency.js";
|
||||
import type {
|
||||
ProjectRuntime,
|
||||
ProjectRuntimeConfig,
|
||||
RuntimeStatus,
|
||||
GlobalMetrics,
|
||||
} from "./project-runtime.js";
|
||||
import { projectManagerLog } from "./logger.js";
|
||||
|
||||
/**
|
||||
* Events emitted by ProjectManager with project attribution.
|
||||
*/
|
||||
export interface ProjectManagerEvents {
|
||||
/** Emitted when a task is created in any project */
|
||||
"task:created": [data: { projectId: string; projectName: string; task: Task }];
|
||||
/** Emitted when a task is moved in any project */
|
||||
"task:moved": [
|
||||
data: {
|
||||
projectId: string;
|
||||
projectName: string;
|
||||
task: Task;
|
||||
from: string;
|
||||
to: string;
|
||||
}
|
||||
];
|
||||
/** Emitted when a task is updated in any project */
|
||||
"task:updated": [data: { projectId: string; projectName: string; task: Task }];
|
||||
/** Emitted when an error occurs in any project */
|
||||
"error": [data: { projectId: string; projectName: string; error: Error }];
|
||||
/** Emitted when project health status changes */
|
||||
"health:changed": [
|
||||
data: {
|
||||
projectId: string;
|
||||
projectName: string;
|
||||
status: RuntimeStatus;
|
||||
previous: RuntimeStatus;
|
||||
}
|
||||
];
|
||||
/** Emitted when a runtime is added */
|
||||
"runtime:added": [data: { projectId: string; projectName: string }];
|
||||
/** Emitted when a runtime is removed */
|
||||
"runtime:removed": [data: { projectId: string; projectName: string }];
|
||||
}
|
||||
|
||||
/**
|
||||
* ProjectManager orchestrates all project runtimes and enforces global
|
||||
* concurrency limits from CentralCore.
|
||||
*
|
||||
* This is the main entry point for multi-project support in the engine.
|
||||
* It manages multiple ProjectRuntime instances (one per project) and:
|
||||
* - Creates appropriate runtime type based on isolation mode
|
||||
* - Forwards runtime events with project attribution
|
||||
* - Enforces global concurrency limits via CentralCore
|
||||
* - Updates project health in CentralCore
|
||||
* - Logs activity to CentralCore's unified feed
|
||||
*
|
||||
* **Project Registration Assumption:**
|
||||
* ProjectManager assumes projects are already registered in CentralCore.
|
||||
* Call `centralCore.registerProject()` before `addProject()`.
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* const central = new CentralCore();
|
||||
* await central.init();
|
||||
*
|
||||
* const manager = new ProjectManager(central);
|
||||
*
|
||||
* // Register project in CentralCore first
|
||||
* const project = await central.registerProject({
|
||||
* name: "My Project",
|
||||
* path: "/path/to/project"
|
||||
* });
|
||||
*
|
||||
* // Then add runtime
|
||||
* const runtime = await manager.addProject({
|
||||
* projectId: project.id,
|
||||
* workingDirectory: project.path,
|
||||
* isolationMode: "in-process",
|
||||
* maxConcurrent: 2,
|
||||
* maxWorktrees: 4,
|
||||
* });
|
||||
*
|
||||
* // Listen to all project events
|
||||
* manager.on("task:created", ({ projectId, projectName, task }) => {
|
||||
* console.log(`${projectName}: Task ${task.id} created`);
|
||||
* });
|
||||
* ```
|
||||
*/
|
||||
export class ProjectManager extends EventEmitter<ProjectManagerEvents> {
|
||||
private runtimes = new Map<string, ProjectRuntime>();
|
||||
private projectNames = new Map<string, string>();
|
||||
private globalSemaphore: AgentSemaphore;
|
||||
|
||||
/**
|
||||
* @param centralCore - CentralCore reference for global coordination
|
||||
*/
|
||||
constructor(private centralCore: CentralCore) {
|
||||
super();
|
||||
this.setMaxListeners(100);
|
||||
|
||||
// Initialize global semaphore with limit from CentralCore
|
||||
this.globalSemaphore = new AgentSemaphore(() => {
|
||||
// This will be updated dynamically from CentralCore
|
||||
return 4; // Default, will refresh
|
||||
});
|
||||
|
||||
// Refresh the global limit periodically
|
||||
this.refreshGlobalLimit();
|
||||
setInterval(() => this.refreshGlobalLimit(), 30000); // Refresh every 30s
|
||||
|
||||
projectManagerLog.log("ProjectManager initialized");
|
||||
}
|
||||
|
||||
/**
|
||||
* Refresh the global concurrency limit from CentralCore.
|
||||
*/
|
||||
private async refreshGlobalLimit(): Promise<void> {
|
||||
try {
|
||||
const state = await this.centralCore.getGlobalConcurrencyState();
|
||||
// Update semaphore limit dynamically
|
||||
// Note: AgentSemaphore reads limit via getter, so we update the source
|
||||
this.globalSemaphore = new AgentSemaphore(() => state.globalMaxConcurrent);
|
||||
} catch (error) {
|
||||
projectManagerLog.warn("Failed to refresh global concurrency limit:", error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a project runtime and start it.
|
||||
*
|
||||
* **Prerequisite:** Project must already be registered in CentralCore.
|
||||
*
|
||||
* @param config - Runtime configuration (must match a registered project)
|
||||
* @returns The created and started ProjectRuntime
|
||||
* @throws Error if project not found in CentralCore or runtime already exists
|
||||
*/
|
||||
async addProject(config: ProjectRuntimeConfig): Promise<ProjectRuntime> {
|
||||
// Validate project exists in CentralCore
|
||||
const project = await this.centralCore.getProject(config.projectId);
|
||||
if (!project) {
|
||||
throw new Error(
|
||||
`Project ${config.projectId} not found in CentralCore. ` +
|
||||
"Register the project first with centralCore.registerProject()."
|
||||
);
|
||||
}
|
||||
|
||||
// Check if runtime already exists
|
||||
if (this.runtimes.has(config.projectId)) {
|
||||
throw new Error(`Runtime already exists for project ${config.projectId}`);
|
||||
}
|
||||
|
||||
projectManagerLog.log(`Adding project runtime for ${config.projectId} (${project.name})`);
|
||||
|
||||
// Store project name for event attribution
|
||||
this.projectNames.set(config.projectId, project.name);
|
||||
|
||||
// Create appropriate runtime based on isolation mode
|
||||
let runtime: ProjectRuntime;
|
||||
if (config.isolationMode === "child-process") {
|
||||
runtime = new ChildProcessRuntime(config, this.centralCore);
|
||||
} else {
|
||||
// Default to in-process
|
||||
runtime = new InProcessRuntime(config, this.centralCore);
|
||||
}
|
||||
|
||||
// Set up event forwarding with project attribution
|
||||
this.setupEventForwarding(runtime, config.projectId, project.name);
|
||||
|
||||
// Start the runtime
|
||||
await runtime.start();
|
||||
|
||||
// Update project health to active
|
||||
await this.centralCore.updateProjectHealth(config.projectId, {
|
||||
status: "active",
|
||||
activeTaskCount: 0,
|
||||
inFlightAgentCount: 0,
|
||||
totalTasksCompleted: 0,
|
||||
totalTasksFailed: 0,
|
||||
updatedAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
// Store runtime
|
||||
this.runtimes.set(config.projectId, runtime);
|
||||
|
||||
// Log to activity feed
|
||||
await this.centralCore.logActivity({
|
||||
type: "task:created", // Using task:created as a generic activity type
|
||||
projectId: config.projectId,
|
||||
projectName: project.name,
|
||||
timestamp: new Date().toISOString(),
|
||||
details: `Project runtime started for ${project.name}`,
|
||||
});
|
||||
|
||||
this.emit("runtime:added", { projectId: config.projectId, projectName: project.name });
|
||||
|
||||
projectManagerLog.log(`Project runtime added for ${config.projectId}`);
|
||||
return runtime;
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove a project runtime and stop it.
|
||||
*
|
||||
* @param id - Project ID to remove
|
||||
* @throws Error if runtime not found
|
||||
*/
|
||||
async removeProject(id: string): Promise<void> {
|
||||
const runtime = this.runtimes.get(id);
|
||||
if (!runtime) {
|
||||
throw new Error(`Runtime not found for project ${id}`);
|
||||
}
|
||||
|
||||
const projectName = this.projectNames.get(id) ?? id;
|
||||
projectManagerLog.log(`Removing project runtime for ${id}`);
|
||||
|
||||
// Stop the runtime
|
||||
await runtime.stop();
|
||||
|
||||
// Remove from maps
|
||||
this.runtimes.delete(id);
|
||||
this.projectNames.delete(id);
|
||||
|
||||
// Update project health to paused
|
||||
await this.centralCore.updateProjectHealth(id, {
|
||||
status: "paused",
|
||||
activeTaskCount: 0,
|
||||
inFlightAgentCount: 0,
|
||||
updatedAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
// Log to activity feed
|
||||
await this.centralCore.logActivity({
|
||||
type: "task:created",
|
||||
projectId: id,
|
||||
projectName: projectName,
|
||||
timestamp: new Date().toISOString(),
|
||||
details: `Project runtime stopped for ${projectName}`,
|
||||
});
|
||||
|
||||
this.emit("runtime:removed", { projectId: id, projectName });
|
||||
|
||||
projectManagerLog.log(`Project runtime removed for ${id}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get a runtime by project ID.
|
||||
*/
|
||||
getRuntime(id: string): ProjectRuntime | undefined {
|
||||
return this.runtimes.get(id);
|
||||
}
|
||||
|
||||
/**
|
||||
* List all managed runtimes.
|
||||
*/
|
||||
listRuntimes(): ProjectRuntime[] {
|
||||
return Array.from(this.runtimes.values());
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all project IDs.
|
||||
*/
|
||||
getProjectIds(): string[] {
|
||||
return Array.from(this.runtimes.keys());
|
||||
}
|
||||
|
||||
/**
|
||||
* Get global metrics aggregated across all runtimes.
|
||||
*/
|
||||
async getGlobalMetrics(): Promise<GlobalMetrics> {
|
||||
const statusCounts: Record<RuntimeStatus, number> = {
|
||||
active: 0,
|
||||
paused: 0,
|
||||
errored: 0,
|
||||
stopped: 0,
|
||||
starting: 0,
|
||||
stopping: 0,
|
||||
};
|
||||
|
||||
let totalInFlight = 0;
|
||||
let totalActiveAgents = 0;
|
||||
|
||||
for (const [id, runtime] of this.runtimes) {
|
||||
const status = runtime.getStatus();
|
||||
statusCounts[status]++;
|
||||
|
||||
try {
|
||||
const metrics = runtime.getMetrics();
|
||||
totalInFlight += metrics.inFlightTasks;
|
||||
totalActiveAgents += metrics.activeAgents;
|
||||
|
||||
// Update health in CentralCore
|
||||
const project = await this.centralCore.getProject(id);
|
||||
if (project) {
|
||||
await this.centralCore.updateProjectHealth(id, {
|
||||
status: status as "active" | "paused" | "errored" | "initializing",
|
||||
activeTaskCount: metrics.inFlightTasks,
|
||||
inFlightAgentCount: metrics.activeAgents,
|
||||
lastActivityAt: metrics.lastActivityAt,
|
||||
updatedAt: new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
projectManagerLog.warn(`Failed to get metrics for ${id}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
totalInFlightTasks: totalInFlight,
|
||||
totalActiveAgents: totalActiveAgents,
|
||||
runtimeCountByStatus: statusCounts,
|
||||
totalRuntimes: this.runtimes.size,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up event forwarding from runtime to ProjectManager listeners.
|
||||
*/
|
||||
private setupEventForwarding(
|
||||
runtime: ProjectRuntime,
|
||||
projectId: string,
|
||||
projectName: string
|
||||
): void {
|
||||
// Forward task:created
|
||||
runtime.on("task:created", (task: Task) => {
|
||||
this.emit("task:created", { projectId, projectName, task });
|
||||
this.logActivity("task:created", projectId, projectName, `Task ${task.id} created`, task.id, task.title);
|
||||
});
|
||||
|
||||
// Forward task:moved
|
||||
runtime.on("task:moved", (data: { task: Task; from: string; to: string }) => {
|
||||
this.emit("task:moved", { projectId, projectName, task: data.task, from: data.from, to: data.to });
|
||||
this.logActivity(
|
||||
"task:moved",
|
||||
projectId,
|
||||
projectName,
|
||||
`Task ${data.task.id} moved: ${data.from} → ${data.to}`,
|
||||
data.task.id,
|
||||
data.task.title,
|
||||
{ from: data.from, to: data.to }
|
||||
);
|
||||
});
|
||||
|
||||
// Forward task:updated
|
||||
runtime.on("task:updated", (task: Task) => {
|
||||
this.emit("task:updated", { projectId, projectName, task });
|
||||
});
|
||||
|
||||
// Forward errors
|
||||
runtime.on("error", (error: Error) => {
|
||||
this.emit("error", { projectId, projectName, error });
|
||||
this.logActivity(
|
||||
"task:failed",
|
||||
projectId,
|
||||
projectName,
|
||||
`Error in ${projectName}: ${error.message}`
|
||||
);
|
||||
});
|
||||
|
||||
// Forward health changes
|
||||
runtime.on("health-changed", (data: { status: RuntimeStatus; previous: RuntimeStatus }) => {
|
||||
this.emit("health:changed", { projectId, projectName, status: data.status, previous: data.previous });
|
||||
|
||||
// Update health in CentralCore
|
||||
this.centralCore.updateProjectHealth(projectId, {
|
||||
status: data.status as "active" | "paused" | "errored" | "initializing",
|
||||
updatedAt: new Date().toISOString(),
|
||||
}).catch((err: unknown) => {
|
||||
projectManagerLog.warn(`Failed to update health for ${projectId}:`, err);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Log activity to CentralCore's unified feed.
|
||||
*/
|
||||
private async logActivity(
|
||||
type: "task:created" | "task:moved" | "task:failed",
|
||||
projectId: string,
|
||||
projectName: string,
|
||||
details: string,
|
||||
taskId?: string,
|
||||
taskTitle?: string,
|
||||
metadata?: Record<string, unknown>
|
||||
): Promise<void> {
|
||||
try {
|
||||
await this.centralCore.logActivity({
|
||||
type,
|
||||
projectId,
|
||||
projectName,
|
||||
timestamp: new Date().toISOString(),
|
||||
details,
|
||||
taskId,
|
||||
taskTitle,
|
||||
metadata,
|
||||
});
|
||||
} catch (error) {
|
||||
// Best-effort logging
|
||||
projectManagerLog.warn("Failed to log activity:", error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Acquire a global concurrency slot.
|
||||
* Call this before starting task execution.
|
||||
*/
|
||||
async acquireGlobalSlot(projectId: string): Promise<boolean> {
|
||||
try {
|
||||
return await this.centralCore.acquireGlobalSlot(projectId);
|
||||
} catch (error) {
|
||||
projectManagerLog.error(`Failed to acquire global slot for ${projectId}:`, error);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Release a global concurrency slot.
|
||||
* Call this after task execution completes.
|
||||
*/
|
||||
async releaseGlobalSlot(projectId: string): Promise<void> {
|
||||
try {
|
||||
await this.centralCore.releaseGlobalSlot(projectId);
|
||||
} catch (error) {
|
||||
projectManagerLog.error(`Failed to release global slot for ${projectId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop all runtimes and clean up.
|
||||
*/
|
||||
async stopAll(): Promise<void> {
|
||||
projectManagerLog.log("Stopping all project runtimes...");
|
||||
|
||||
const stopPromises = Array.from(this.runtimes.keys()).map((id) =>
|
||||
this.removeProject(id).catch((error) => {
|
||||
projectManagerLog.error(`Failed to stop runtime ${id}:`, error);
|
||||
})
|
||||
);
|
||||
|
||||
await Promise.all(stopPromises);
|
||||
|
||||
projectManagerLog.log("All project runtimes stopped");
|
||||
this.removeAllListeners();
|
||||
}
|
||||
}
|
||||
143
packages/engine/src/project-runtime.ts
Normal file
143
packages/engine/src/project-runtime.ts
Normal file
@@ -0,0 +1,143 @@
|
||||
import type { EventEmitter } from "node:events";
|
||||
import type { TaskStore, Task, IsolationMode, ProjectSettings } from "@fusion/core";
|
||||
import type { Scheduler } from "./scheduler.js";
|
||||
|
||||
/**
|
||||
* Runtime status for a ProjectRuntime instance.
|
||||
* Represents the lifecycle states of a project runtime.
|
||||
*/
|
||||
export type RuntimeStatus =
|
||||
| "active" // Runtime is running and processing tasks
|
||||
| "paused" // Runtime is temporarily suspended
|
||||
| "errored" // Runtime encountered a fatal error
|
||||
| "stopped" // Runtime is stopped (graceful shutdown complete)
|
||||
| "starting" // Runtime is in the process of starting
|
||||
| "stopping"; // Runtime is in the process of stopping
|
||||
|
||||
/**
|
||||
* Metrics for a ProjectRuntime instance.
|
||||
* Used for monitoring and health tracking.
|
||||
*/
|
||||
export interface RuntimeMetrics {
|
||||
/** Number of tasks currently in-progress */
|
||||
inFlightTasks: number;
|
||||
/** Number of active agents currently running */
|
||||
activeAgents: number;
|
||||
/** ISO-8601 timestamp of the last activity */
|
||||
lastActivityAt: string;
|
||||
/** Memory usage in bytes (optional, may not be available in all modes) */
|
||||
memoryBytes?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Configuration for creating a ProjectRuntime instance.
|
||||
*/
|
||||
export interface ProjectRuntimeConfig {
|
||||
/** Unique project ID (e.g., "proj_abc123") */
|
||||
projectId: string;
|
||||
/** Absolute path to the project working directory */
|
||||
workingDirectory: string;
|
||||
/** Execution isolation mode */
|
||||
isolationMode: IsolationMode;
|
||||
/** Maximum concurrent agents for this project */
|
||||
maxConcurrent: number;
|
||||
/** Maximum worktrees for this project */
|
||||
maxWorktrees: number;
|
||||
/** Optional project settings override */
|
||||
settings?: ProjectSettings;
|
||||
}
|
||||
|
||||
/**
|
||||
* Events emitted by a ProjectRuntime instance.
|
||||
*/
|
||||
export interface ProjectRuntimeEvents {
|
||||
/** Emitted when a task is created in the project */
|
||||
"task:created": [task: Task];
|
||||
/** Emitted when a task is moved between columns */
|
||||
"task:moved": [data: { task: Task; from: string; to: string }];
|
||||
/** Emitted when a task is updated */
|
||||
"task:updated": [task: Task];
|
||||
/** Emitted when an error occurs in the runtime */
|
||||
"error": [error: Error];
|
||||
/** Emitted when the runtime health status changes */
|
||||
"health-changed": [data: { status: RuntimeStatus; previous: RuntimeStatus }];
|
||||
}
|
||||
|
||||
/**
|
||||
* ProjectRuntime interface — core abstraction for multi-project support.
|
||||
*
|
||||
* Each project instance runs as a ProjectRuntime, either in-process (default)
|
||||
* or in an isolated child process (opt-in). The ProjectManager orchestrates
|
||||
* all runtimes and enforces global concurrency limits from CentralCore.
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* const runtime = new InProcessRuntime(config, centralCore);
|
||||
* await runtime.start();
|
||||
*
|
||||
* // Access project TaskStore
|
||||
* const taskStore = runtime.getTaskStore();
|
||||
*
|
||||
* // Listen for events
|
||||
* runtime.on("task:created", (task) => {
|
||||
* console.log(`Task ${task.id} created`);
|
||||
* });
|
||||
*
|
||||
* // Shutdown gracefully
|
||||
* await runtime.stop();
|
||||
* ```
|
||||
*/
|
||||
export interface ProjectRuntime extends EventEmitter<ProjectRuntimeEvents> {
|
||||
/**
|
||||
* Start the runtime and initialize all subsystems.
|
||||
* This includes initializing the TaskStore, Scheduler, Executor, and WorktreePool.
|
||||
*/
|
||||
start(): Promise<void>;
|
||||
|
||||
/**
|
||||
* Stop the runtime with graceful shutdown.
|
||||
* Waits for active tasks to complete (with timeout), stops the scheduler,
|
||||
* and cleans up resources.
|
||||
*/
|
||||
stop(): Promise<void>;
|
||||
|
||||
/**
|
||||
* Get the current runtime status.
|
||||
* @returns The current status of the runtime
|
||||
*/
|
||||
getStatus(): RuntimeStatus;
|
||||
|
||||
/**
|
||||
* Get the project's TaskStore instance.
|
||||
* @returns The TaskStore for this project
|
||||
* @throws Error if called on a ChildProcessRuntime (not accessible in child mode)
|
||||
*/
|
||||
getTaskStore(): TaskStore;
|
||||
|
||||
/**
|
||||
* Get the project's Scheduler instance.
|
||||
* @returns The Scheduler for this project
|
||||
* @throws Error if called on a ChildProcessRuntime (not accessible in child mode)
|
||||
*/
|
||||
getScheduler(): Scheduler;
|
||||
|
||||
/**
|
||||
* Get current runtime metrics.
|
||||
* @returns Metrics including in-flight tasks, active agents, and memory usage
|
||||
*/
|
||||
getMetrics(): RuntimeMetrics;
|
||||
}
|
||||
|
||||
/**
|
||||
* Global metrics aggregated across all project runtimes.
|
||||
*/
|
||||
export interface GlobalMetrics {
|
||||
/** Total number of in-flight tasks across all runtimes */
|
||||
totalInFlightTasks: number;
|
||||
/** Total number of active agents across all runtimes */
|
||||
totalActiveAgents: number;
|
||||
/** Number of runtimes by status */
|
||||
runtimeCountByStatus: Record<RuntimeStatus, number>;
|
||||
/** Total number of registered runtimes */
|
||||
totalRuntimes: number;
|
||||
}
|
||||
102
packages/engine/src/runtimes/child-process-runtime.test.ts
Normal file
102
packages/engine/src/runtimes/child-process-runtime.test.ts
Normal file
@@ -0,0 +1,102 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import type { CentralCore } from "@fusion/core";
|
||||
import { ChildProcessRuntime } from "./child-process-runtime.js";
|
||||
import type { ProjectRuntimeConfig } from "../project-runtime.js";
|
||||
|
||||
// Mock child_process
|
||||
vi.mock("node:child_process", () => ({
|
||||
fork: vi.fn().mockReturnValue({
|
||||
on: vi.fn(),
|
||||
kill: vi.fn(),
|
||||
killed: false,
|
||||
connected: false,
|
||||
send: vi.fn(),
|
||||
}),
|
||||
}));
|
||||
|
||||
describe("ChildProcessRuntime", () => {
|
||||
let runtime: ChildProcessRuntime;
|
||||
let mockCentralCore: CentralCore;
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_test123",
|
||||
workingDirectory: "/tmp/test-project",
|
||||
isolationMode: "child-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
mockCentralCore = {
|
||||
getGlobalConcurrencyState: vi.fn().mockResolvedValue({
|
||||
globalMaxConcurrent: 4,
|
||||
currentlyActive: 0,
|
||||
queuedCount: 0,
|
||||
projectsActive: {},
|
||||
}),
|
||||
} as unknown as CentralCore;
|
||||
|
||||
runtime = new ChildProcessRuntime(testConfig, mockCentralCore);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
await runtime.stop();
|
||||
} catch {
|
||||
// Ignore errors during cleanup
|
||||
}
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
describe("lifecycle", () => {
|
||||
it("should start with status 'stopped'", () => {
|
||||
expect(runtime.getStatus()).toBe("stopped");
|
||||
});
|
||||
|
||||
it("should throw when getting TaskStore", () => {
|
||||
expect(() => runtime.getTaskStore()).toThrow("not accessible in ChildProcessRuntime");
|
||||
});
|
||||
|
||||
it("should throw when getting Scheduler", () => {
|
||||
expect(() => runtime.getScheduler()).toThrow("not accessible in ChildProcessRuntime");
|
||||
});
|
||||
|
||||
it("should return metrics even when stopped", () => {
|
||||
const metrics = runtime.getMetrics();
|
||||
expect(metrics.inFlightTasks).toBe(0);
|
||||
expect(metrics.activeAgents).toBe(0);
|
||||
expect(metrics.lastActivityAt).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("configuration", () => {
|
||||
it("should store projectId in config", () => {
|
||||
expect(testConfig.projectId).toBe("proj_test123");
|
||||
});
|
||||
|
||||
it("should store workingDirectory in config", () => {
|
||||
expect(testConfig.workingDirectory).toBe("/tmp/test-project");
|
||||
});
|
||||
|
||||
it("should have child-process isolation mode", () => {
|
||||
expect(testConfig.isolationMode).toBe("child-process");
|
||||
});
|
||||
});
|
||||
|
||||
describe("event handling", () => {
|
||||
it("should support health-changed event", () => {
|
||||
const handler = vi.fn();
|
||||
runtime.on("health-changed", handler);
|
||||
|
||||
// The constructor may emit health-changed, so we just verify
|
||||
// the event listener can be registered
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should support error event", () => {
|
||||
const handler = vi.fn();
|
||||
runtime.on("error", handler);
|
||||
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
499
packages/engine/src/runtimes/child-process-runtime.ts
Normal file
499
packages/engine/src/runtimes/child-process-runtime.ts
Normal file
@@ -0,0 +1,499 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import { fork, type ChildProcess } from "node:child_process";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { dirname, join } from "node:path";
|
||||
import type {
|
||||
Task,
|
||||
TaskStore,
|
||||
CentralCore,
|
||||
} from "@fusion/core";
|
||||
import type { Scheduler } from "../scheduler.js";
|
||||
import type {
|
||||
ProjectRuntime,
|
||||
ProjectRuntimeConfig,
|
||||
RuntimeStatus,
|
||||
RuntimeMetrics,
|
||||
ProjectRuntimeEvents,
|
||||
} from "../project-runtime.js";
|
||||
import { IpcHost } from "../ipc/ipc-host.js";
|
||||
import {
|
||||
START_RUNTIME,
|
||||
STOP_RUNTIME,
|
||||
GET_STATUS,
|
||||
GET_METRICS,
|
||||
TASK_CREATED,
|
||||
TASK_MOVED,
|
||||
TASK_UPDATED,
|
||||
ERROR_EVENT,
|
||||
HEALTH_CHANGED,
|
||||
type TaskCreatedPayload,
|
||||
type TaskMovedPayload,
|
||||
type TaskUpdatedPayload,
|
||||
type ErrorEventPayload,
|
||||
type HealthChangedPayload,
|
||||
} from "../ipc/ipc-protocol.js";
|
||||
import { runtimeLog } from "../logger.js";
|
||||
|
||||
/**
|
||||
* Health monitor for tracking child process health.
|
||||
*/
|
||||
class HealthMonitor {
|
||||
private running = false;
|
||||
private missedHeartbeats = 0;
|
||||
private interval: ReturnType<typeof setInterval> | null = null;
|
||||
private restartAttempts = 0;
|
||||
private restartDelays = [1000, 5000, 15000]; // Exponential backoff: 1s, 5s, 15s
|
||||
|
||||
constructor(
|
||||
private onHealthCheck: () => Promise<boolean>,
|
||||
private onUnhealthy: () => void,
|
||||
private options: {
|
||||
intervalMs?: number;
|
||||
maxMissedHeartbeats?: number;
|
||||
maxRestartAttempts?: number;
|
||||
} = {}
|
||||
) {}
|
||||
|
||||
start(): void {
|
||||
if (this.running) return;
|
||||
this.running = true;
|
||||
|
||||
const intervalMs = this.options.intervalMs ?? 5000;
|
||||
const maxMissed = this.options.maxMissedHeartbeats ?? 3;
|
||||
|
||||
this.interval = setInterval(async () => {
|
||||
const healthy = await this.onHealthCheck();
|
||||
|
||||
if (healthy) {
|
||||
if (this.missedHeartbeats > 0) {
|
||||
runtimeLog.log(`Health recovered after ${this.missedHeartbeats} missed heartbeats`);
|
||||
}
|
||||
this.missedHeartbeats = 0;
|
||||
this.restartAttempts = 0; // Reset restart attempts on success
|
||||
} else {
|
||||
this.missedHeartbeats++;
|
||||
runtimeLog.warn(`Missed heartbeat ${this.missedHeartbeats}/${maxMissed}`);
|
||||
|
||||
if (this.missedHeartbeats >= maxMissed) {
|
||||
runtimeLog.error(`Health check failed after ${maxMissed} attempts`);
|
||||
this.onUnhealthy();
|
||||
}
|
||||
}
|
||||
}, intervalMs);
|
||||
|
||||
runtimeLog.log(`Health monitor started (interval: ${intervalMs}ms)`);
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
this.running = false;
|
||||
if (this.interval) {
|
||||
clearInterval(this.interval);
|
||||
this.interval = null;
|
||||
}
|
||||
this.missedHeartbeats = 0;
|
||||
runtimeLog.log("Health monitor stopped");
|
||||
}
|
||||
|
||||
getRestartDelay(): number {
|
||||
const delay = this.restartDelays[this.restartAttempts] ?? this.restartDelays[this.restartDelays.length - 1];
|
||||
return delay;
|
||||
}
|
||||
|
||||
incrementRestartAttempts(): void {
|
||||
this.restartAttempts++;
|
||||
}
|
||||
|
||||
getRestartAttempts(): number {
|
||||
return this.restartAttempts;
|
||||
}
|
||||
|
||||
getMissedHeartbeats(): number {
|
||||
return this.missedHeartbeats;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* ChildProcessRuntime runs a project in an isolated child process.
|
||||
*
|
||||
* This provides stronger isolation between projects at the cost of
|
||||
* IPC overhead. The child process runs an InProcessRuntime internally
|
||||
* and communicates with the host via IPC messages.
|
||||
*
|
||||
* Features:
|
||||
* - Process isolation (separate memory space)
|
||||
* - Automatic restart on crash with exponential backoff
|
||||
* - Health monitoring via heartbeat protocol
|
||||
* - Graceful shutdown with configurable timeout
|
||||
* - Event forwarding from child process to host listeners
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* const config: ProjectRuntimeConfig = {
|
||||
* projectId: "proj_abc123",
|
||||
* workingDirectory: "/path/to/project",
|
||||
* isolationMode: "child-process",
|
||||
* maxConcurrent: 2,
|
||||
* maxWorktrees: 4,
|
||||
* };
|
||||
*
|
||||
* const runtime = new ChildProcessRuntime(config, centralCore);
|
||||
* await runtime.start();
|
||||
*
|
||||
* // Access metrics via IPC
|
||||
* const metrics = runtime.getMetrics();
|
||||
*
|
||||
* await runtime.stop();
|
||||
* ```
|
||||
*/
|
||||
export class ChildProcessRuntime
|
||||
extends EventEmitter<ProjectRuntimeEvents>
|
||||
implements ProjectRuntime
|
||||
{
|
||||
private status: RuntimeStatus = "stopped";
|
||||
private child: ChildProcess | null = null;
|
||||
private ipcHost: IpcHost | null = null;
|
||||
private healthMonitor: HealthMonitor;
|
||||
private lastMetrics: RuntimeMetrics = {
|
||||
inFlightTasks: 0,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
};
|
||||
|
||||
/**
|
||||
* @param config - Runtime configuration
|
||||
* @param centralCore - CentralCore reference for global coordination
|
||||
*/
|
||||
constructor(
|
||||
private config: ProjectRuntimeConfig,
|
||||
private centralCore: CentralCore
|
||||
) {
|
||||
super();
|
||||
this.setMaxListeners(100);
|
||||
|
||||
// Initialize health monitor
|
||||
this.healthMonitor = new HealthMonitor(
|
||||
async () => this.checkHealth(),
|
||||
() => this.handleUnhealthy(),
|
||||
{ intervalMs: 5000, maxMissedHeartbeats: 3, maxRestartAttempts: 3 }
|
||||
);
|
||||
|
||||
runtimeLog.log(`Created ChildProcessRuntime for project ${config.projectId}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the runtime by spawning a child process.
|
||||
*
|
||||
* Startup sequence:
|
||||
* 1. Set status to "starting"
|
||||
* 2. Fork child process pointing to worker entry point
|
||||
* 3. Set up IPC host with the child process
|
||||
* 4. Send START_RUNTIME command with serialized config
|
||||
* 5. Wait for OK response or timeout (10s)
|
||||
* 6. Start health monitoring heartbeat
|
||||
* 7. Set status to "active"
|
||||
*/
|
||||
async start(): Promise<void> {
|
||||
if (this.status !== "stopped") {
|
||||
throw new Error(`Cannot start runtime: current status is ${this.status}`);
|
||||
}
|
||||
|
||||
this.setStatus("starting");
|
||||
runtimeLog.log(`Starting ChildProcessRuntime for project ${this.config.projectId}`);
|
||||
|
||||
try {
|
||||
await this.spawnChild();
|
||||
this.setStatus("active");
|
||||
runtimeLog.log(`ChildProcessRuntime started for project ${this.config.projectId}`);
|
||||
} catch (error) {
|
||||
const err = error instanceof Error ? error : new Error(String(error));
|
||||
this.setStatus("errored");
|
||||
runtimeLog.error(`Failed to start ChildProcessRuntime:`, err.message);
|
||||
this.emit("error", err);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Spawn the child process and set up IPC.
|
||||
*/
|
||||
private async spawnChild(): Promise<void> {
|
||||
// Determine worker entry point
|
||||
const workerPath = this.getWorkerPath();
|
||||
|
||||
runtimeLog.log(`Forking child process: ${workerPath}`);
|
||||
|
||||
// Fork child process
|
||||
this.child = fork(workerPath, [], {
|
||||
silent: true, // Pipe stdout/stderr
|
||||
execArgv: [], // Don't inherit exec arguments
|
||||
});
|
||||
|
||||
// Set up IPC host
|
||||
this.ipcHost = new IpcHost(this.child, { commandTimeoutMs: 10000 });
|
||||
|
||||
// Set up event forwarding
|
||||
this.setupEventForwarding();
|
||||
|
||||
// Send START_RUNTIME command
|
||||
runtimeLog.log("Sending START_RUNTIME command to child");
|
||||
await this.ipcHost.sendCommand(START_RUNTIME, { config: this.config });
|
||||
|
||||
// Start health monitoring
|
||||
this.healthMonitor.start();
|
||||
|
||||
// Handle child process exit
|
||||
this.child.on("exit", (code, signal) => {
|
||||
runtimeLog.warn(`Child process exited (code: ${code}, signal: ${signal})`);
|
||||
this.handleChildExit(code, signal);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the path to the worker entry point.
|
||||
*/
|
||||
private getWorkerPath(): string {
|
||||
// In production, use the compiled .js file
|
||||
// In development/tests, use .ts with tsx
|
||||
const isCompiled = !import.meta.url.endsWith(".ts");
|
||||
|
||||
const currentDir = dirname(fileURLToPath(import.meta.url));
|
||||
const workerFile = isCompiled ? "child-process-worker.js" : "child-process-worker.ts";
|
||||
|
||||
return join(currentDir, workerFile);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up event forwarding from IPC host to runtime listeners.
|
||||
*/
|
||||
private setupEventForwarding(): void {
|
||||
if (!this.ipcHost) return;
|
||||
|
||||
// Forward task events
|
||||
this.ipcHost.on(TASK_CREATED, (payload: TaskCreatedPayload) => {
|
||||
this.emit("task:created", payload.task);
|
||||
});
|
||||
|
||||
this.ipcHost.on(TASK_MOVED, (payload: TaskMovedPayload) => {
|
||||
this.emit("task:moved", { task: payload.task, from: payload.from, to: payload.to });
|
||||
});
|
||||
|
||||
this.ipcHost.on(TASK_UPDATED, (payload: TaskUpdatedPayload) => {
|
||||
this.emit("task:updated", payload.task);
|
||||
});
|
||||
|
||||
// Forward error events
|
||||
this.ipcHost.on(ERROR_EVENT, (payload: ErrorEventPayload) => {
|
||||
const error = new Error(payload.message);
|
||||
if (payload.code) {
|
||||
(error as Error & { code: string }).code = payload.code;
|
||||
}
|
||||
this.emit("error", error);
|
||||
});
|
||||
|
||||
// Forward health change events
|
||||
this.ipcHost.on(HEALTH_CHANGED, (payload: HealthChangedPayload) => {
|
||||
this.status = payload.status as RuntimeStatus;
|
||||
this.emit("health-changed", { status: payload.status, previous: payload.previous });
|
||||
});
|
||||
|
||||
// Handle disconnect
|
||||
this.ipcHost.on("disconnect", () => {
|
||||
runtimeLog.warn("IPC host disconnected");
|
||||
this.handleDisconnection();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the runtime with graceful shutdown.
|
||||
*
|
||||
* Shutdown sequence:
|
||||
* 1. Set status to "stopping"
|
||||
* 2. Stop health monitoring
|
||||
* 3. Send STOP_RUNTIME command with 30s timeout
|
||||
* 4. Kill child process if graceful shutdown fails
|
||||
* 5. Set status to "stopped"
|
||||
*/
|
||||
async stop(): Promise<void> {
|
||||
if (this.status === "stopped" || this.status === "stopping") {
|
||||
return;
|
||||
}
|
||||
|
||||
this.setStatus("stopping");
|
||||
runtimeLog.log(`Stopping ChildProcessRuntime for project ${this.config.projectId}`);
|
||||
|
||||
// Stop health monitoring
|
||||
this.healthMonitor.stop();
|
||||
|
||||
try {
|
||||
// Send graceful shutdown command
|
||||
if (this.ipcHost?.isConnected()) {
|
||||
runtimeLog.log("Sending STOP_RUNTIME command to child");
|
||||
await this.ipcHost.sendCommand(STOP_RUNTIME, { timeoutMs: 30000 }, 35000);
|
||||
}
|
||||
} catch (error) {
|
||||
runtimeLog.warn(`Graceful shutdown failed: ${error}`);
|
||||
}
|
||||
|
||||
// Kill child process if still running
|
||||
this.killChild();
|
||||
|
||||
this.setStatus("stopped");
|
||||
runtimeLog.log(`ChildProcessRuntime stopped for project ${this.config.projectId}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Kill the child process forcefully.
|
||||
*/
|
||||
private killChild(): void {
|
||||
if (this.child && !this.child.killed) {
|
||||
runtimeLog.log("Killing child process");
|
||||
this.child.kill("SIGTERM");
|
||||
|
||||
// Force kill after 5 seconds if still running
|
||||
setTimeout(() => {
|
||||
if (this.child && !this.child.killed) {
|
||||
runtimeLog.warn("Force killing child process");
|
||||
this.child.kill("SIGKILL");
|
||||
}
|
||||
}, 5000);
|
||||
}
|
||||
|
||||
this.child = null;
|
||||
this.ipcHost = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the current runtime status.
|
||||
*/
|
||||
getStatus(): RuntimeStatus {
|
||||
return this.status;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the project's TaskStore instance.
|
||||
* @throws Error - Not accessible in child mode (use IPC instead)
|
||||
*/
|
||||
getTaskStore(): TaskStore {
|
||||
throw new Error(
|
||||
"TaskStore is not accessible in ChildProcessRuntime. " +
|
||||
"Use IPC methods to access task data."
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the project's Scheduler instance.
|
||||
* @throws Error - Not accessible in child mode
|
||||
*/
|
||||
getScheduler(): Scheduler {
|
||||
throw new Error(
|
||||
"Scheduler is not accessible in ChildProcessRuntime. " +
|
||||
"Use IPC methods to interact with the scheduler."
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current runtime metrics (via IPC query).
|
||||
*/
|
||||
getMetrics(): RuntimeMetrics {
|
||||
// Query metrics via IPC if connected
|
||||
if (this.ipcHost?.isConnected()) {
|
||||
// Fire-and-forget metrics request - returns cached value immediately
|
||||
this.ipcHost
|
||||
.sendCommand(GET_METRICS, {})
|
||||
.then((metrics: unknown) => {
|
||||
this.lastMetrics = metrics as RuntimeMetrics;
|
||||
})
|
||||
.catch(() => {
|
||||
// Ignore errors, use cached value
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
...this.lastMetrics,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Check health by pinging the child process.
|
||||
*/
|
||||
private async checkHealth(): Promise<boolean> {
|
||||
if (!this.ipcHost?.isConnected()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
await this.ipcHost.ping(5000);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle unhealthy child process (restart or error).
|
||||
*/
|
||||
private handleUnhealthy(): void {
|
||||
const maxRestarts = 3;
|
||||
|
||||
if (this.healthMonitor.getRestartAttempts() >= maxRestarts) {
|
||||
runtimeLog.error(`Max restart attempts (${maxRestarts}) reached, transitioning to errored`);
|
||||
this.setStatus("errored");
|
||||
this.emit("error", new Error("Child process failed after max restart attempts"));
|
||||
return;
|
||||
}
|
||||
|
||||
const delay = this.healthMonitor.getRestartDelay();
|
||||
this.healthMonitor.incrementRestartAttempts();
|
||||
|
||||
runtimeLog.log(`Attempting restart ${this.healthMonitor.getRestartAttempts()}/${maxRestarts} after ${delay}ms`);
|
||||
|
||||
setTimeout(async () => {
|
||||
try {
|
||||
this.killChild();
|
||||
await this.spawnChild();
|
||||
runtimeLog.log("Child process restarted successfully");
|
||||
} catch (error) {
|
||||
runtimeLog.error("Failed to restart child process:", error);
|
||||
this.setStatus("errored");
|
||||
this.emit("error", error instanceof Error ? error : new Error(String(error)));
|
||||
}
|
||||
}, delay);
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle child process exit.
|
||||
*/
|
||||
private handleChildExit(code: number | null, signal: string | null): void {
|
||||
// Don't restart if we're intentionally stopping
|
||||
if (this.status === "stopping" || this.status === "stopped") {
|
||||
return;
|
||||
}
|
||||
|
||||
// Unexpected exit - trigger restart
|
||||
runtimeLog.warn(`Unexpected child exit (code: ${code}, signal: ${signal})`);
|
||||
this.handleUnhealthy();
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle IPC disconnection.
|
||||
*/
|
||||
private handleDisconnection(): void {
|
||||
if (this.status !== "stopping" && this.status !== "stopped") {
|
||||
runtimeLog.error("IPC channel disconnected unexpectedly");
|
||||
this.handleUnhealthy();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Update status and emit health-changed event.
|
||||
*/
|
||||
private setStatus(newStatus: RuntimeStatus): void {
|
||||
const previous = this.status;
|
||||
this.status = newStatus;
|
||||
|
||||
if (previous !== newStatus) {
|
||||
this.emit("health-changed", { status: newStatus, previous });
|
||||
}
|
||||
}
|
||||
}
|
||||
174
packages/engine/src/runtimes/child-process-worker.ts
Normal file
174
packages/engine/src/runtimes/child-process-worker.ts
Normal file
@@ -0,0 +1,174 @@
|
||||
/**
|
||||
* Child Process Worker Entry Point
|
||||
*
|
||||
* This module runs inside a forked child process and creates an InProcessRuntime
|
||||
* internally. It communicates with the host via IPC using the IpcWorker class.
|
||||
*
|
||||
* The worker:
|
||||
* 1. Detects if it's running as a forked child (process.send available)
|
||||
* 2. Creates an IpcWorker instance
|
||||
* 3. Registers command handlers (START_RUNTIME, STOP_RUNTIME, etc.)
|
||||
* 4. Forwards all runtime events to the host via IPC
|
||||
* 5. Handles graceful shutdown on SIGTERM
|
||||
*/
|
||||
|
||||
import { IpcWorker } from "../ipc/ipc-worker.js";
|
||||
import {
|
||||
START_RUNTIME,
|
||||
STOP_RUNTIME,
|
||||
GET_STATUS,
|
||||
GET_METRICS,
|
||||
ERROR_EVENT,
|
||||
type StartRuntimePayload,
|
||||
type StopRuntimePayload,
|
||||
} from "../ipc/ipc-protocol.js";
|
||||
import { InProcessRuntime } from "./in-process-runtime.js";
|
||||
import type { ProjectRuntimeConfig } from "../project-runtime.js";
|
||||
import { runtimeLog } from "../logger.js";
|
||||
import { CentralCore } from "@fusion/core";
|
||||
|
||||
// Only run if we're in a forked child process
|
||||
if (!process.send) {
|
||||
console.error("This module must be run as a forked child process");
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
runtimeLog.log("Child process worker starting...");
|
||||
|
||||
// Create IPC worker
|
||||
const ipcWorker = new IpcWorker();
|
||||
|
||||
// InProcessRuntime instance (created when START_RUNTIME is received)
|
||||
let runtime: InProcessRuntime | null = null;
|
||||
|
||||
// Create a minimal CentralCore stub for the child process
|
||||
// The child doesn't need full CentralCore functionality
|
||||
const createStubCentralCore = (): CentralCore => {
|
||||
return {
|
||||
getGlobalConcurrencyState: async () => ({
|
||||
globalMaxConcurrent: 4,
|
||||
currentlyActive: 0,
|
||||
queuedCount: 0,
|
||||
projectsActive: {},
|
||||
}),
|
||||
recordTaskCompletion: async () => {},
|
||||
} as unknown as CentralCore;
|
||||
};
|
||||
|
||||
// Register command handlers
|
||||
|
||||
// START_RUNTIME: Create and start the InProcessRuntime
|
||||
ipcWorker.onCommand(START_RUNTIME, async (payload: unknown) => {
|
||||
const { config } = payload as StartRuntimePayload;
|
||||
runtimeLog.log(`Received START_RUNTIME command for project ${config.projectId}`);
|
||||
|
||||
if (runtime) {
|
||||
throw new Error("Runtime already started");
|
||||
}
|
||||
|
||||
// Create stub CentralCore (real coordination happens in host)
|
||||
const centralCore = createStubCentralCore();
|
||||
|
||||
// Create InProcessRuntime
|
||||
runtime = new InProcessRuntime(config, centralCore);
|
||||
|
||||
// Forward runtime events to host
|
||||
runtime.on("task:created", (task) => {
|
||||
ipcWorker.sendEvent("TASK_CREATED", { task });
|
||||
});
|
||||
|
||||
runtime.on("task:moved", (data) => {
|
||||
ipcWorker.sendEvent("TASK_MOVED", data);
|
||||
});
|
||||
|
||||
runtime.on("task:updated", (task) => {
|
||||
ipcWorker.sendEvent("TASK_UPDATED", { task });
|
||||
});
|
||||
|
||||
runtime.on("error", (error) => {
|
||||
ipcWorker.sendEvent(ERROR_EVENT, {
|
||||
message: error.message,
|
||||
code: (error as Error & { code?: string }).code,
|
||||
});
|
||||
});
|
||||
|
||||
runtime.on("health-changed", (data) => {
|
||||
ipcWorker.sendEvent("HEALTH_CHANGED", data);
|
||||
});
|
||||
|
||||
// Start the runtime
|
||||
await runtime.start();
|
||||
|
||||
runtimeLog.log("Runtime started successfully");
|
||||
return { status: runtime.getStatus() };
|
||||
});
|
||||
|
||||
// STOP_RUNTIME: Stop the runtime gracefully
|
||||
ipcWorker.onCommand(STOP_RUNTIME, async (payload: unknown) => {
|
||||
runtimeLog.log("Received STOP_RUNTIME command");
|
||||
|
||||
if (!runtime) {
|
||||
throw new Error("Runtime not started");
|
||||
}
|
||||
|
||||
const { timeoutMs } = (payload as StopRuntimePayload) || {};
|
||||
|
||||
await runtime.stop();
|
||||
runtime = null;
|
||||
|
||||
runtimeLog.log("Runtime stopped successfully");
|
||||
return { stopped: true };
|
||||
});
|
||||
|
||||
// GET_STATUS: Return current runtime status
|
||||
ipcWorker.onCommand(GET_STATUS, async () => {
|
||||
if (!runtime) {
|
||||
return { status: "stopped" };
|
||||
}
|
||||
return { status: runtime.getStatus() };
|
||||
});
|
||||
|
||||
// GET_METRICS: Return runtime metrics
|
||||
ipcWorker.onCommand(GET_METRICS, async () => {
|
||||
if (!runtime) {
|
||||
return {
|
||||
inFlightTasks: 0,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
return runtime.getMetrics();
|
||||
});
|
||||
|
||||
// Handle graceful shutdown
|
||||
process.on("SIGTERM", async () => {
|
||||
runtimeLog.log("Received SIGTERM, initiating graceful shutdown...");
|
||||
|
||||
if (runtime) {
|
||||
try {
|
||||
await runtime.stop();
|
||||
runtimeLog.log("Runtime stopped gracefully");
|
||||
} catch (error) {
|
||||
runtimeLog.error("Error during graceful shutdown:", error);
|
||||
}
|
||||
}
|
||||
|
||||
ipcWorker.shutdown();
|
||||
});
|
||||
|
||||
process.on("SIGINT", async () => {
|
||||
runtimeLog.log("Received SIGINT, initiating graceful shutdown...");
|
||||
|
||||
if (runtime) {
|
||||
try {
|
||||
await runtime.stop();
|
||||
runtimeLog.log("Runtime stopped gracefully");
|
||||
} catch (error) {
|
||||
runtimeLog.error("Error during graceful shutdown:", error);
|
||||
}
|
||||
}
|
||||
|
||||
ipcWorker.shutdown();
|
||||
});
|
||||
|
||||
runtimeLog.log("Child process worker initialized and ready");
|
||||
268
packages/engine/src/runtimes/in-process-runtime.test.ts
Normal file
268
packages/engine/src/runtimes/in-process-runtime.test.ts
Normal file
@@ -0,0 +1,268 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { Task, TaskStore, CentralCore } from "@fusion/core";
|
||||
import { InProcessRuntime } from "./in-process-runtime.js";
|
||||
import type { ProjectRuntimeConfig } from "../project-runtime.js";
|
||||
|
||||
// Mock the TaskStore class
|
||||
vi.mock("@fusion/core", async () => {
|
||||
const actual = await vi.importActual<typeof import("@fusion/core")>("@fusion/core");
|
||||
|
||||
return {
|
||||
...actual,
|
||||
TaskStore: vi.fn().mockImplementation(function(this: TaskStore, rootDir: string) {
|
||||
const self = this as unknown as Record<string, unknown>;
|
||||
self.getRootDir = () => rootDir;
|
||||
self.init = vi.fn().mockResolvedValue(undefined);
|
||||
self.listTasks = vi.fn().mockResolvedValue([]);
|
||||
self.getSettings = vi.fn().mockResolvedValue({});
|
||||
self.on = vi.fn().mockReturnValue(self);
|
||||
self.emit = vi.fn().mockReturnValue(true);
|
||||
return self;
|
||||
}),
|
||||
};
|
||||
});
|
||||
|
||||
// Mock the worktree pool
|
||||
vi.mock("../worktree-pool.js", async () => {
|
||||
const actual = await vi.importActual<typeof import("../worktree-pool.js")>("../worktree-pool.js");
|
||||
|
||||
return {
|
||||
...actual,
|
||||
scanIdleWorktrees: vi.fn().mockResolvedValue([]),
|
||||
};
|
||||
});
|
||||
|
||||
// Mock the scheduler
|
||||
vi.mock("../scheduler.js", async () => {
|
||||
return {
|
||||
Scheduler: vi.fn().mockImplementation(() => {
|
||||
const self = {} as Record<string, unknown>;
|
||||
self.start = vi.fn();
|
||||
self.stop = vi.fn();
|
||||
return self;
|
||||
}),
|
||||
};
|
||||
});
|
||||
|
||||
// Mock the executor
|
||||
vi.mock("../executor.js", async () => {
|
||||
return {
|
||||
TaskExecutor: vi.fn().mockImplementation(() => {
|
||||
const self = {} as Record<string, unknown>;
|
||||
self.resumeOrphaned = vi.fn().mockResolvedValue(undefined);
|
||||
self.activeWorktrees = new Map();
|
||||
return self;
|
||||
}),
|
||||
};
|
||||
});
|
||||
|
||||
describe("InProcessRuntime", () => {
|
||||
let runtime: InProcessRuntime;
|
||||
let mockCentralCore: CentralCore;
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_test123",
|
||||
workingDirectory: "/tmp/test-project",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
// Create mock CentralCore
|
||||
mockCentralCore = {
|
||||
getGlobalConcurrencyState: vi.fn().mockResolvedValue({
|
||||
globalMaxConcurrent: 4,
|
||||
currentlyActive: 0,
|
||||
queuedCount: 0,
|
||||
projectsActive: {},
|
||||
}),
|
||||
recordTaskCompletion: vi.fn().mockResolvedValue(undefined),
|
||||
} as unknown as CentralCore;
|
||||
|
||||
runtime = new InProcessRuntime(testConfig, mockCentralCore);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
await runtime.stop();
|
||||
} catch {
|
||||
// Ignore errors during cleanup
|
||||
}
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
describe("lifecycle", () => {
|
||||
it("should start with status 'stopped'", () => {
|
||||
expect(runtime.getStatus()).toBe("stopped");
|
||||
});
|
||||
|
||||
it("should transition to 'active' after start", async () => {
|
||||
await runtime.start();
|
||||
expect(runtime.getStatus()).toBe("active");
|
||||
});
|
||||
|
||||
it("should transition to 'stopped' after stop", async () => {
|
||||
await runtime.start();
|
||||
await runtime.stop();
|
||||
expect(runtime.getStatus()).toBe("stopped");
|
||||
});
|
||||
|
||||
it("should throw if starting when not stopped", async () => {
|
||||
await runtime.start();
|
||||
await expect(runtime.start()).rejects.toThrow("Cannot start runtime");
|
||||
});
|
||||
|
||||
it("should handle stop when already stopped", async () => {
|
||||
// Should not throw
|
||||
await runtime.stop();
|
||||
expect(runtime.getStatus()).toBe("stopped");
|
||||
});
|
||||
|
||||
it("should transition through 'starting' during start", async () => {
|
||||
const statusChanges: string[] = [];
|
||||
runtime.on("health-changed", (data) => {
|
||||
statusChanges.push(data.status);
|
||||
});
|
||||
|
||||
await runtime.start();
|
||||
|
||||
expect(statusChanges).toContain("starting");
|
||||
expect(statusChanges).toContain("active");
|
||||
});
|
||||
|
||||
it("should transition through 'stopping' during stop", async () => {
|
||||
await runtime.start();
|
||||
|
||||
const statusChanges: string[] = [];
|
||||
runtime.on("health-changed", (data) => {
|
||||
statusChanges.push(data.status);
|
||||
});
|
||||
|
||||
await runtime.stop();
|
||||
|
||||
expect(statusChanges).toContain("stopping");
|
||||
expect(statusChanges).toContain("stopped");
|
||||
});
|
||||
});
|
||||
|
||||
describe("event forwarding", () => {
|
||||
it("should emit health-changed on status transitions", async () => {
|
||||
const healthChangedSpy = vi.fn();
|
||||
runtime.on("health-changed", healthChangedSpy);
|
||||
|
||||
await runtime.start();
|
||||
|
||||
expect(healthChangedSpy).toHaveBeenCalled();
|
||||
const calls = healthChangedSpy.mock.calls;
|
||||
const lastCall = calls[calls.length - 1][0];
|
||||
expect(lastCall.status).toBe("active");
|
||||
expect(lastCall.previous).toBe("starting");
|
||||
});
|
||||
|
||||
it("should emit task:created when task store emits task:created", async () => {
|
||||
await runtime.start();
|
||||
|
||||
const taskCreatedSpy = vi.fn();
|
||||
runtime.on("task:created", taskCreatedSpy);
|
||||
|
||||
// Get the mock TaskStore and simulate an event
|
||||
const taskStore = runtime.getTaskStore();
|
||||
const mockTask = { id: "KB-001", title: "Test Task" } as Task;
|
||||
|
||||
// Get the registered handler and call it
|
||||
const onCalls = (taskStore.on as ReturnType<typeof vi.fn>).mock.calls;
|
||||
const taskCreatedHandler = onCalls.find((call: unknown[]) => call[0] === "task:created");
|
||||
|
||||
if (taskCreatedHandler) {
|
||||
(taskCreatedHandler[1] as (task: Task) => void)(mockTask);
|
||||
}
|
||||
|
||||
expect(taskCreatedSpy).toHaveBeenCalledWith(mockTask);
|
||||
});
|
||||
|
||||
it("should emit task:moved when task store emits task:moved", async () => {
|
||||
await runtime.start();
|
||||
|
||||
const taskMovedSpy = vi.fn();
|
||||
runtime.on("task:moved", taskMovedSpy);
|
||||
|
||||
const taskStore = runtime.getTaskStore();
|
||||
const mockTask = { id: "KB-001", title: "Test Task" } as Task;
|
||||
const moveData = { task: mockTask, from: "todo", to: "in-progress" };
|
||||
|
||||
const onCalls = (taskStore.on as ReturnType<typeof vi.fn>).mock.calls;
|
||||
const taskMovedHandler = onCalls.find((call: unknown[]) => call[0] === "task:moved");
|
||||
|
||||
if (taskMovedHandler) {
|
||||
(taskMovedHandler[1] as (data: { task: Task; from: string; to: string }) => void)(moveData);
|
||||
}
|
||||
|
||||
expect(taskMovedSpy).toHaveBeenCalledWith(moveData);
|
||||
});
|
||||
});
|
||||
|
||||
describe("metrics", () => {
|
||||
it("should return metrics with default values before start", () => {
|
||||
const metrics = runtime.getMetrics();
|
||||
|
||||
expect(metrics.inFlightTasks).toBe(0);
|
||||
expect(metrics.activeAgents).toBe(0);
|
||||
expect(metrics.lastActivityAt).toBeDefined();
|
||||
});
|
||||
|
||||
it("should include memory usage in metrics", () => {
|
||||
const metrics = runtime.getMetrics();
|
||||
|
||||
// Memory usage may or may not be available depending on environment
|
||||
if (metrics.memoryBytes !== undefined) {
|
||||
expect(typeof metrics.memoryBytes).toBe("number");
|
||||
expect(metrics.memoryBytes).toBeGreaterThanOrEqual(0);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("accessors", () => {
|
||||
it("should throw when accessing TaskStore before start", () => {
|
||||
expect(() => runtime.getTaskStore()).toThrow("TaskStore not initialized");
|
||||
});
|
||||
|
||||
it("should throw when accessing Scheduler before start", () => {
|
||||
expect(() => runtime.getScheduler()).toThrow("Scheduler not initialized");
|
||||
});
|
||||
|
||||
it("should return TaskStore after start", async () => {
|
||||
await runtime.start();
|
||||
const taskStore = runtime.getTaskStore();
|
||||
|
||||
expect(taskStore).toBeDefined();
|
||||
expect(taskStore.getRootDir()).toBe(testConfig.workingDirectory);
|
||||
});
|
||||
|
||||
it("should return Scheduler after start", async () => {
|
||||
await runtime.start();
|
||||
const scheduler = runtime.getScheduler();
|
||||
|
||||
expect(scheduler).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("configuration", () => {
|
||||
it("should store projectId in config", () => {
|
||||
// Access via the constructor params - runtime is created with testConfig
|
||||
expect(testConfig.projectId).toBe("proj_test123");
|
||||
});
|
||||
|
||||
it("should store workingDirectory in config", () => {
|
||||
expect(testConfig.workingDirectory).toBe("/tmp/test-project");
|
||||
});
|
||||
|
||||
it("should store maxConcurrent in config", () => {
|
||||
expect(testConfig.maxConcurrent).toBe(2);
|
||||
});
|
||||
|
||||
it("should store maxWorktrees in config", () => {
|
||||
expect(testConfig.maxWorktrees).toBe(4);
|
||||
});
|
||||
});
|
||||
});
|
||||
395
packages/engine/src/runtimes/in-process-runtime.ts
Normal file
395
packages/engine/src/runtimes/in-process-runtime.ts
Normal file
@@ -0,0 +1,395 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import type {
|
||||
TaskStore,
|
||||
Task,
|
||||
CentralCore,
|
||||
} from "@fusion/core";
|
||||
import { Scheduler } from "../scheduler.js";
|
||||
import { TaskExecutor, type TaskExecutorOptions } from "../executor.js";
|
||||
import { WorktreePool } from "../worktree-pool.js";
|
||||
import { AgentSemaphore } from "../concurrency.js";
|
||||
import type {
|
||||
ProjectRuntime,
|
||||
ProjectRuntimeConfig,
|
||||
RuntimeStatus,
|
||||
RuntimeMetrics,
|
||||
ProjectRuntimeEvents,
|
||||
} from "../project-runtime.js";
|
||||
import { runtimeLog } from "../logger.js";
|
||||
import type { StuckTaskDetector } from "../stuck-task-detector.js";
|
||||
import type { UsageLimitPauser } from "../usage-limit-detector.js";
|
||||
|
||||
/**
|
||||
* InProcessRuntime runs a project within the main process.
|
||||
*
|
||||
* This is the default execution mode — all components (TaskStore, Scheduler,
|
||||
* Executor, WorktreePool) share the same memory space and event loop.
|
||||
*
|
||||
* Features:
|
||||
* - Direct access to TaskStore and Scheduler via getter methods
|
||||
* - Synchronous event forwarding from TaskStore to runtime listeners
|
||||
* - Graceful shutdown with configurable timeout
|
||||
* - Automatic orphaned task recovery on startup
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* const config: ProjectRuntimeConfig = {
|
||||
* projectId: "proj_abc123",
|
||||
* workingDirectory: "/path/to/project",
|
||||
* isolationMode: "in-process",
|
||||
* maxConcurrent: 2,
|
||||
* maxWorktrees: 4,
|
||||
* };
|
||||
*
|
||||
* const runtime = new InProcessRuntime(config, centralCore);
|
||||
* await runtime.start();
|
||||
*
|
||||
* // Access components directly
|
||||
* const taskStore = runtime.getTaskStore();
|
||||
* const scheduler = runtime.getScheduler();
|
||||
*
|
||||
* await runtime.stop();
|
||||
* ```
|
||||
*/
|
||||
export class InProcessRuntime
|
||||
extends EventEmitter<ProjectRuntimeEvents>
|
||||
implements ProjectRuntime
|
||||
{
|
||||
private status: RuntimeStatus = "stopped";
|
||||
private taskStore!: TaskStore;
|
||||
private scheduler!: Scheduler;
|
||||
private executor!: TaskExecutor;
|
||||
private worktreePool!: WorktreePool;
|
||||
private globalSemaphore?: AgentSemaphore;
|
||||
private stuckTaskDetector?: StuckTaskDetector;
|
||||
private usageLimitPauser?: UsageLimitPauser;
|
||||
private lastActivityAt: string = new Date().toISOString();
|
||||
|
||||
/**
|
||||
* @param config - Runtime configuration
|
||||
* @param centralCore - CentralCore reference for global coordination
|
||||
*/
|
||||
constructor(
|
||||
private config: ProjectRuntimeConfig,
|
||||
private centralCore: CentralCore
|
||||
) {
|
||||
super();
|
||||
this.setMaxListeners(100);
|
||||
runtimeLog.log(`Created InProcessRuntime for project ${config.projectId}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the runtime and initialize all subsystems.
|
||||
*
|
||||
* Initialization order:
|
||||
* 1. Initialize TaskStore
|
||||
* 2. Initialize WorktreePool
|
||||
* 3. Initialize Scheduler (with TaskStore)
|
||||
* 4. Initialize TaskExecutor (with TaskStore, worktree pool, global semaphore)
|
||||
* 5. Resume orphaned in-progress tasks
|
||||
* 6. Start scheduler
|
||||
*/
|
||||
async start(): Promise<void> {
|
||||
if (this.status !== "stopped") {
|
||||
throw new Error(`Cannot start runtime: current status is ${this.status}`);
|
||||
}
|
||||
|
||||
this.setStatus("starting");
|
||||
runtimeLog.log(`Starting InProcessRuntime for project ${this.config.projectId}`);
|
||||
|
||||
try {
|
||||
// 1. Initialize TaskStore
|
||||
const { TaskStore } = await import("@fusion/core");
|
||||
this.taskStore = new TaskStore(this.config.workingDirectory);
|
||||
await this.taskStore.init();
|
||||
runtimeLog.log(`TaskStore initialized for project ${this.config.projectId}`);
|
||||
|
||||
// 2. Initialize WorktreePool
|
||||
this.worktreePool = new WorktreePool();
|
||||
|
||||
// Rehydrate pool from disk state (idle worktrees)
|
||||
const { scanIdleWorktrees } = await import("../worktree-pool.js");
|
||||
const idleWorktrees = await scanIdleWorktrees(
|
||||
this.config.workingDirectory,
|
||||
this.taskStore
|
||||
);
|
||||
if (idleWorktrees.length > 0) {
|
||||
this.worktreePool.rehydrate(idleWorktrees);
|
||||
runtimeLog.log(
|
||||
`Rehydrated worktree pool with ${idleWorktrees.length} idle worktrees`
|
||||
);
|
||||
}
|
||||
|
||||
// 3. Initialize global semaphore from CentralCore
|
||||
const globalLimit = await this.getGlobalConcurrencyLimit();
|
||||
this.globalSemaphore = new AgentSemaphore(() => globalLimit);
|
||||
|
||||
// 4. Initialize Scheduler
|
||||
this.scheduler = new Scheduler(this.taskStore, {
|
||||
maxConcurrent: this.config.maxConcurrent,
|
||||
maxWorktrees: this.config.maxWorktrees,
|
||||
semaphore: this.globalSemaphore,
|
||||
onSchedule: (task) => {
|
||||
this.recordActivity();
|
||||
runtimeLog.log(`Scheduled task ${task.id}`);
|
||||
},
|
||||
onBlocked: (task, blockedBy) => {
|
||||
runtimeLog.log(`Task ${task.id} blocked by: ${blockedBy.join(", ")}`);
|
||||
},
|
||||
});
|
||||
|
||||
// 5. Initialize TaskExecutor
|
||||
const executorOptions: TaskExecutorOptions = {
|
||||
semaphore: this.globalSemaphore,
|
||||
pool: this.worktreePool,
|
||||
usageLimitPauser: this.usageLimitPauser,
|
||||
stuckTaskDetector: this.stuckTaskDetector,
|
||||
onStart: (task, worktreePath) => {
|
||||
this.recordActivity();
|
||||
runtimeLog.log(`Started executing task ${task.id} in ${worktreePath}`);
|
||||
},
|
||||
onComplete: (task) => {
|
||||
this.recordActivity();
|
||||
runtimeLog.log(`Completed task ${task.id}`);
|
||||
// Record task completion in CentralCore
|
||||
this.recordTaskCompletion(task.id, true);
|
||||
},
|
||||
onError: (task, error) => {
|
||||
this.recordActivity();
|
||||
runtimeLog.error(`Task ${task.id} failed:`, error.message);
|
||||
this.recordTaskCompletion(task.id, false);
|
||||
},
|
||||
};
|
||||
|
||||
this.executor = new TaskExecutor(
|
||||
this.taskStore,
|
||||
this.config.workingDirectory,
|
||||
executorOptions
|
||||
);
|
||||
|
||||
// 6. Set up event forwarding from TaskStore
|
||||
this.setupEventForwarding();
|
||||
|
||||
// 7. Resume orphaned in-progress tasks
|
||||
await this.executor.resumeOrphaned();
|
||||
|
||||
// 8. Start scheduler
|
||||
this.scheduler.start();
|
||||
|
||||
this.setStatus("active");
|
||||
runtimeLog.log(`InProcessRuntime started for project ${this.config.projectId}`);
|
||||
} catch (error) {
|
||||
const err = error instanceof Error ? error : new Error(String(error));
|
||||
this.setStatus("errored");
|
||||
runtimeLog.error(`Failed to start InProcessRuntime:`, err.message);
|
||||
this.emit("error", err);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the runtime with graceful shutdown.
|
||||
*
|
||||
* Shutdown sequence:
|
||||
* 1. Set status to "stopping"
|
||||
* 2. Stop scheduler (no new tasks)
|
||||
* 3. Wait for executor to finish active tasks (with timeout)
|
||||
* 4. Drain and cleanup worktree pool
|
||||
* 5. Set status to "stopped"
|
||||
*
|
||||
* @throws Error if shutdown timeout is exceeded
|
||||
*/
|
||||
async stop(): Promise<void> {
|
||||
if (this.status === "stopped" || this.status === "stopping") {
|
||||
return;
|
||||
}
|
||||
|
||||
this.setStatus("stopping");
|
||||
runtimeLog.log(`Stopping InProcessRuntime for project ${this.config.projectId}`);
|
||||
|
||||
try {
|
||||
// 1. Stop scheduler (prevents new task scheduling)
|
||||
if (this.scheduler) {
|
||||
this.scheduler.stop();
|
||||
runtimeLog.log("Scheduler stopped");
|
||||
}
|
||||
|
||||
// 2. Wait for active tasks to complete (30 second timeout)
|
||||
const shutdownTimeout = 30000;
|
||||
const startTime = Date.now();
|
||||
|
||||
while (Date.now() - startTime < shutdownTimeout) {
|
||||
const metrics = this.getMetrics();
|
||||
if (metrics.inFlightTasks === 0) {
|
||||
break;
|
||||
}
|
||||
runtimeLog.log(
|
||||
`Waiting for ${metrics.inFlightTasks} in-flight tasks to complete...`
|
||||
);
|
||||
await new Promise((resolve) => setTimeout(resolve, 1000));
|
||||
}
|
||||
|
||||
// Check if we timed out
|
||||
const finalMetrics = this.getMetrics();
|
||||
if (finalMetrics.inFlightTasks > 0) {
|
||||
runtimeLog.warn(
|
||||
`Shutdown timeout reached with ${finalMetrics.inFlightTasks} tasks still in-flight`
|
||||
);
|
||||
}
|
||||
|
||||
// 3. Drain and cleanup worktree pool
|
||||
if (this.worktreePool) {
|
||||
const worktrees = this.worktreePool.drain();
|
||||
if (worktrees.length > 0) {
|
||||
runtimeLog.log(`Drained ${worktrees.length} worktrees from pool`);
|
||||
}
|
||||
}
|
||||
|
||||
this.setStatus("stopped");
|
||||
runtimeLog.log(`InProcessRuntime stopped for project ${this.config.projectId}`);
|
||||
} catch (error) {
|
||||
const err = error instanceof Error ? error : new Error(String(error));
|
||||
this.setStatus("errored");
|
||||
runtimeLog.error(`Error during shutdown:`, err.message);
|
||||
this.emit("error", err);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the current runtime status.
|
||||
*/
|
||||
getStatus(): RuntimeStatus {
|
||||
return this.status;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the project's TaskStore instance.
|
||||
* @throws Error if runtime has not been started
|
||||
*/
|
||||
getTaskStore(): TaskStore {
|
||||
if (!this.taskStore) {
|
||||
throw new Error("TaskStore not initialized. Call start() first.");
|
||||
}
|
||||
return this.taskStore;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the project's Scheduler instance.
|
||||
* @throws Error if runtime has not been started
|
||||
*/
|
||||
getScheduler(): Scheduler {
|
||||
if (!this.scheduler) {
|
||||
throw new Error("Scheduler not initialized. Call start() first.");
|
||||
}
|
||||
return this.scheduler;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current runtime metrics.
|
||||
*/
|
||||
getMetrics(): RuntimeMetrics {
|
||||
// Estimate in-flight tasks by checking active sessions
|
||||
const inFlightTasks = this.executor
|
||||
? (this.executor as unknown as { activeWorktrees?: Map<string, string> }).activeWorktrees?.size ?? 0
|
||||
: 0;
|
||||
|
||||
// Get active agent count from the semaphore
|
||||
const activeAgents = this.globalSemaphore?.activeCount ?? 0;
|
||||
|
||||
// Get memory usage if available
|
||||
const memoryBytes = process.memoryUsage?.().heapUsed;
|
||||
|
||||
return {
|
||||
inFlightTasks,
|
||||
activeAgents,
|
||||
lastActivityAt: this.lastActivityAt,
|
||||
memoryBytes,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the StuckTaskDetector for this runtime.
|
||||
*/
|
||||
setStuckTaskDetector(detector: StuckTaskDetector): void {
|
||||
this.stuckTaskDetector = detector;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the UsageLimitPauser for this runtime.
|
||||
*/
|
||||
setUsageLimitPauser(pauser: UsageLimitPauser): void {
|
||||
this.usageLimitPauser = pauser;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up event forwarding from TaskStore to runtime listeners.
|
||||
*/
|
||||
private setupEventForwarding(): void {
|
||||
// Forward task:created events
|
||||
this.taskStore.on("task:created", (task: Task) => {
|
||||
this.recordActivity();
|
||||
this.emit("task:created", task);
|
||||
});
|
||||
|
||||
// Forward task:moved events
|
||||
this.taskStore.on("task:moved", (data: { task: Task; from: string; to: string }) => {
|
||||
this.recordActivity();
|
||||
this.emit("task:moved", data);
|
||||
});
|
||||
|
||||
// Forward task:updated events
|
||||
this.taskStore.on("task:updated", (task: Task) => {
|
||||
this.recordActivity();
|
||||
this.emit("task:updated", task);
|
||||
});
|
||||
|
||||
runtimeLog.log("Event forwarding setup complete");
|
||||
}
|
||||
|
||||
/**
|
||||
* Update status and emit health-changed event.
|
||||
*/
|
||||
private setStatus(newStatus: RuntimeStatus): void {
|
||||
const previous = this.status;
|
||||
this.status = newStatus;
|
||||
|
||||
if (previous !== newStatus) {
|
||||
this.emit("health-changed", { status: newStatus, previous });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Record activity timestamp.
|
||||
*/
|
||||
private recordActivity(): void {
|
||||
this.lastActivityAt = new Date().toISOString();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get global concurrency limit from CentralCore.
|
||||
*/
|
||||
private async getGlobalConcurrencyLimit(): Promise<number> {
|
||||
try {
|
||||
const state = await this.centralCore.getGlobalConcurrencyState();
|
||||
return state.globalMaxConcurrent;
|
||||
} catch {
|
||||
// Fallback to default if CentralCore is unavailable
|
||||
return 4;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Record task completion in CentralCore.
|
||||
*/
|
||||
private async recordTaskCompletion(taskId: string, success: boolean): Promise<void> {
|
||||
try {
|
||||
// Estimate duration (simplified - in reality, we'd track start time)
|
||||
const durationMs = 0; // Placeholder
|
||||
await this.centralCore.recordTaskCompletion(this.config.projectId, durationMs, success);
|
||||
} catch (error) {
|
||||
// Non-fatal: logging is best-effort
|
||||
runtimeLog.warn(`Failed to record task completion: ${error}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user