chore: consolidate remaining tests into __tests__/ dirs and refactor mocks
Mostly mechanical cleanup left over from the earlier test-consolidation pass: - Update import paths to ../../ for mocks now that test files moved deeper - Simplify mock setup (drop usePluginUiSlots inline mock, etc.) - Move engine ipc + runtimes tests into __tests__/ subdirs - Move dashboard utils tests into __tests__/ subdir - Refresh fusion-plugin-hermes-runtime/dist artifacts build-exe.test.ts: spawn-import fix from a parallel branch (resolved during worktree merge of the CSS extraction work). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
File diff suppressed because it is too large
Load Diff
471
packages/engine/src/ipc/__tests__/ipc-host.test.ts
Normal file
471
packages/engine/src/ipc/__tests__/ipc-host.test.ts
Normal file
@@ -0,0 +1,471 @@
|
||||
/**
|
||||
* Unit tests for IpcHost — the parent-side IPC handler that sends commands
|
||||
* to a child process worker and correlates responses.
|
||||
*
|
||||
* Coverage:
|
||||
* - Constructor: listener setup, options, initial state
|
||||
* - sendCommand: serialization, response correlation (OK/ERROR/PONG), timeout, disconnection
|
||||
* - ping: convenience wrapper for sendCommand("PING")
|
||||
* - Event forwarding: worker events emitted on IpcHost
|
||||
* - Malformed/unknown messages: silently ignored
|
||||
* - Disconnection cascade: child error/exit/disconnect → pending commands rejected
|
||||
* - disconnect(): explicit cleanup and listener removal
|
||||
*/
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { ChildProcess } from "node:child_process";
|
||||
import { IpcHost } from "../ipc-host.js";
|
||||
import { OK, ERROR, PONG, TASK_CREATED } from "../ipc-protocol.js";
|
||||
|
||||
// ── Mock logger to suppress console output ──────────────────────────────
|
||||
vi.mock("../../logger.js", () => ({
|
||||
ipcLog: {
|
||||
log: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
},
|
||||
}));
|
||||
|
||||
// ── Mock ChildProcess factory ───────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Creates a mock ChildProcess that is an EventEmitter with the required
|
||||
* properties for IpcHost: `send`, `connected`, `disconnect`.
|
||||
*/
|
||||
function createMockChildProcess(
|
||||
overrides: {
|
||||
connected?: boolean;
|
||||
send?: ((...args: any[]) => any) | undefined;
|
||||
} = {}
|
||||
): ChildProcess {
|
||||
const emitter = new EventEmitter();
|
||||
const mock = emitter as unknown as ChildProcess & EventEmitter;
|
||||
|
||||
// Default: connected with a working send
|
||||
Object.defineProperty(mock, "connected", {
|
||||
get: () => overrides.connected ?? true,
|
||||
configurable: true,
|
||||
});
|
||||
|
||||
if (overrides.send === undefined && !("send" in overrides)) {
|
||||
// Default: working send that invokes callback with no error
|
||||
(mock as any).send = vi.fn((...args: any[]) => {
|
||||
const callback = args.find((a: unknown) => typeof a === "function");
|
||||
if (callback) callback(null);
|
||||
return true;
|
||||
});
|
||||
} else {
|
||||
(mock as any).send = overrides.send;
|
||||
}
|
||||
|
||||
(mock as any).disconnect = vi.fn();
|
||||
(mock as any).kill = vi.fn();
|
||||
(mock as any).killed = false;
|
||||
(mock as any).pid = 12345;
|
||||
|
||||
return mock;
|
||||
}
|
||||
|
||||
// ── Tests ────────────────────────────────────────────────────────────────
|
||||
|
||||
describe("IpcHost", () => {
|
||||
let child: ChildProcess & EventEmitter;
|
||||
let host: IpcHost;
|
||||
|
||||
beforeEach(() => {
|
||||
child = createMockChildProcess() as ChildProcess & EventEmitter;
|
||||
host = new IpcHost(child);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
host.removeAllListeners();
|
||||
});
|
||||
|
||||
// ── Constructor & initial state ──────────────────────────────────────
|
||||
|
||||
describe("constructor and initial state", () => {
|
||||
it("registers listeners on child process for message, error, exit, disconnect events", () => {
|
||||
// EventEmitter.listenerCount shows listeners were added
|
||||
expect(child.listenerCount("message")).toBeGreaterThanOrEqual(1);
|
||||
expect(child.listenerCount("error")).toBeGreaterThanOrEqual(1);
|
||||
expect(child.listenerCount("exit")).toBeGreaterThanOrEqual(1);
|
||||
expect(child.listenerCount("disconnect")).toBeGreaterThanOrEqual(1);
|
||||
});
|
||||
|
||||
it("isConnected() returns true when child is connected and not disconnected", () => {
|
||||
expect(host.isConnected()).toBe(true);
|
||||
});
|
||||
|
||||
it("isConnected() returns false after disconnection", () => {
|
||||
child.emit("disconnect");
|
||||
expect(host.isConnected()).toBe(false);
|
||||
});
|
||||
|
||||
it("getChildProcess() returns the child process instance", () => {
|
||||
expect(host.getChildProcess()).toBe(child);
|
||||
});
|
||||
|
||||
it("getPendingCommandCount() returns 0 initially", () => {
|
||||
expect(host.getPendingCommandCount()).toBe(0);
|
||||
});
|
||||
|
||||
it("accepts custom commandTimeoutMs option", () => {
|
||||
// We verify this indirectly in the timeout test in Step 2
|
||||
const customHost = new IpcHost(child, { commandTimeoutMs: 500 });
|
||||
expect(customHost).toBeInstanceOf(IpcHost);
|
||||
customHost.removeAllListeners();
|
||||
});
|
||||
});
|
||||
|
||||
// ── sendCommand and response correlation ────────────────────────────
|
||||
|
||||
describe("sendCommand", () => {
|
||||
it("sends a valid IpcMessage via childProcess.send() with correct type, unique id, and payload", async () => {
|
||||
const sendFn = child.send as ReturnType<typeof vi.fn>;
|
||||
const commandPromise = host.sendCommand("GET_STATUS", { foo: "bar" });
|
||||
|
||||
// Extract the message from the mock send call
|
||||
expect(sendFn).toHaveBeenCalledTimes(1);
|
||||
const sentMessage = sendFn.mock.calls[0][0];
|
||||
expect(sentMessage.type).toBe("GET_STATUS");
|
||||
expect(typeof sentMessage.id).toBe("string");
|
||||
expect(sentMessage.id.length).toBeGreaterThan(0);
|
||||
expect(sentMessage.payload).toEqual({ foo: "bar" });
|
||||
|
||||
// Respond to resolve the promise
|
||||
child.emit("message", { type: OK, id: sentMessage.id, payload: { data: "result" } });
|
||||
await expect(commandPromise).resolves.toBe("result");
|
||||
});
|
||||
|
||||
it("resolves with data when child responds with OK matching the correlation ID", async () => {
|
||||
const sendFn = child.send as ReturnType<typeof vi.fn>;
|
||||
const promise = host.sendCommand("GET_METRICS", {});
|
||||
|
||||
const sentId = sendFn.mock.calls[0][0].id;
|
||||
child.emit("message", { type: OK, id: sentId, payload: { data: { tasks: 5 } } });
|
||||
|
||||
await expect(promise).resolves.toEqual({ tasks: 5 });
|
||||
});
|
||||
|
||||
it("rejects with an Error (including message and code) when child responds with ERROR", async () => {
|
||||
const sendFn = child.send as ReturnType<typeof vi.fn>;
|
||||
const promise = host.sendCommand("GET_STATUS", {});
|
||||
|
||||
const sentId = sendFn.mock.calls[0][0].id;
|
||||
child.emit("message", {
|
||||
type: ERROR,
|
||||
id: sentId,
|
||||
payload: { message: "Something went wrong", code: "HANDLER_ERROR" },
|
||||
});
|
||||
|
||||
await expect(promise).rejects.toThrow("Something went wrong");
|
||||
try {
|
||||
await promise;
|
||||
} catch (err: any) {
|
||||
expect(err.code).toBe("HANDLER_ERROR");
|
||||
}
|
||||
});
|
||||
|
||||
it("resolves with pong payload when child responds with PONG", async () => {
|
||||
const sendFn = child.send as ReturnType<typeof vi.fn>;
|
||||
const promise = host.sendCommand("PING", {});
|
||||
|
||||
const sentId = sendFn.mock.calls[0][0].id;
|
||||
child.emit("message", {
|
||||
type: PONG,
|
||||
id: sentId,
|
||||
payload: { timestamp: "2026-04-01T00:00:00.000Z" },
|
||||
});
|
||||
|
||||
await expect(promise).resolves.toEqual({ timestamp: "2026-04-01T00:00:00.000Z" });
|
||||
});
|
||||
|
||||
it("rejects after timeout using fake timers", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const promise = host.sendCommand("GET_STATUS", {}, 1000);
|
||||
|
||||
// Advance past the timeout
|
||||
vi.advanceTimersByTime(1001);
|
||||
|
||||
await expect(promise).rejects.toThrow("timed out after 1000ms");
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("uses custom commandTimeoutMs when no per-call override provided", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const shortHost = new IpcHost(child, { commandTimeoutMs: 200 });
|
||||
const promise = shortHost.sendCommand("GET_STATUS", {});
|
||||
|
||||
vi.advanceTimersByTime(201);
|
||||
|
||||
await expect(promise).rejects.toThrow("timed out after 200ms");
|
||||
shortHost.removeAllListeners();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("clears pending command on successful response (getPendingCommandCount returns 0)", async () => {
|
||||
const sendFn = child.send as ReturnType<typeof vi.fn>;
|
||||
const promise = host.sendCommand("GET_STATUS", {});
|
||||
|
||||
expect(host.getPendingCommandCount()).toBe(1);
|
||||
|
||||
const sentId = sendFn.mock.calls[0][0].id;
|
||||
child.emit("message", { type: OK, id: sentId, payload: { data: null } });
|
||||
await promise;
|
||||
|
||||
expect(host.getPendingCommandCount()).toBe(0);
|
||||
});
|
||||
|
||||
it("rejects immediately when IPC is already disconnected", async () => {
|
||||
child.emit("disconnect");
|
||||
|
||||
await expect(host.sendCommand("GET_STATUS", {})).rejects.toThrow(
|
||||
"Cannot send command: IPC channel disconnected"
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects when childProcess.send is undefined (no IPC channel)", async () => {
|
||||
const noSendChild = createMockChildProcess({ send: undefined }) as ChildProcess & EventEmitter;
|
||||
const noSendHost = new IpcHost(noSendChild);
|
||||
|
||||
await expect(noSendHost.sendCommand("GET_STATUS", {})).rejects.toThrow(
|
||||
"Child process does not have IPC channel"
|
||||
);
|
||||
noSendHost.removeAllListeners();
|
||||
});
|
||||
|
||||
it("rejects when childProcess.send callback returns an error", async () => {
|
||||
const errChild = createMockChildProcess({
|
||||
send: vi.fn((...args: any[]) => {
|
||||
// Find the callback argument (last function arg)
|
||||
const callback = args.find((a: unknown) => typeof a === "function");
|
||||
if (callback) callback(new Error("Send failed"));
|
||||
return false;
|
||||
}) as any,
|
||||
}) as ChildProcess & EventEmitter;
|
||||
const errHost = new IpcHost(errChild);
|
||||
|
||||
await expect(errHost.sendCommand("GET_STATUS", {})).rejects.toThrow("Failed to send command: Send failed");
|
||||
errHost.removeAllListeners();
|
||||
});
|
||||
});
|
||||
|
||||
// ── ping ─────────────────────────────────────────────────────────────
|
||||
|
||||
describe("ping", () => {
|
||||
it("calls sendCommand('PING', {}, 5000) and resolves with timestamp", async () => {
|
||||
const sendFn = child.send as ReturnType<typeof vi.fn>;
|
||||
const promise = host.ping();
|
||||
|
||||
const sentMessage = sendFn.mock.calls[0][0];
|
||||
expect(sentMessage.type).toBe("PING");
|
||||
|
||||
child.emit("message", {
|
||||
type: PONG,
|
||||
id: sentMessage.id,
|
||||
payload: { timestamp: "2026-04-01T12:00:00.000Z" },
|
||||
});
|
||||
|
||||
const result = await promise;
|
||||
expect(result).toEqual({ timestamp: "2026-04-01T12:00:00.000Z" });
|
||||
});
|
||||
});
|
||||
|
||||
// ── Event forwarding ────────────────────────────────────────────────
|
||||
|
||||
describe("event forwarding", () => {
|
||||
it("incoming event messages are emitted on IpcHost with the event type and payload", () => {
|
||||
const handler = vi.fn();
|
||||
host.on(TASK_CREATED, handler);
|
||||
|
||||
const payload = { task: { id: "KB-001", title: "Test" } };
|
||||
child.emit("message", {
|
||||
type: TASK_CREATED,
|
||||
id: "evt-1",
|
||||
payload,
|
||||
});
|
||||
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
expect(handler).toHaveBeenCalledWith(payload);
|
||||
});
|
||||
|
||||
it('generic "message" event is also emitted for every incoming event message', () => {
|
||||
const handler = vi.fn();
|
||||
host.on("message", handler);
|
||||
|
||||
const message = { type: TASK_CREATED, id: "evt-2", payload: { task: {} } };
|
||||
child.emit("message", message);
|
||||
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
expect(handler).toHaveBeenCalledWith(message);
|
||||
});
|
||||
});
|
||||
|
||||
// ── Malformed messages ──────────────────────────────────────────────
|
||||
|
||||
describe("malformed messages", () => {
|
||||
it("silently ignores message missing type", () => {
|
||||
const handler = vi.fn();
|
||||
host.on("message", handler);
|
||||
|
||||
// Missing type
|
||||
child.emit("message", { id: "x", payload: {} });
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("silently ignores message missing id", () => {
|
||||
const handler = vi.fn();
|
||||
host.on("message", handler);
|
||||
|
||||
child.emit("message", { type: "SOME_TYPE", payload: {} });
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("silently ignores message missing payload", () => {
|
||||
const handler = vi.fn();
|
||||
host.on("message", handler);
|
||||
|
||||
child.emit("message", { type: "SOME_TYPE", id: "x" });
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("silently ignores non-object messages", () => {
|
||||
const handler = vi.fn();
|
||||
host.on("message", handler);
|
||||
|
||||
child.emit("message", "not an object");
|
||||
child.emit("message", null);
|
||||
child.emit("message", 42);
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("ignores response for unknown correlation ID without crashing", () => {
|
||||
// Should not throw
|
||||
child.emit("message", {
|
||||
type: OK,
|
||||
id: "unknown-correlation-id",
|
||||
payload: { data: "phantom" },
|
||||
});
|
||||
|
||||
expect(host.getPendingCommandCount()).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
// ── Disconnection cascade ───────────────────────────────────────────
|
||||
|
||||
describe("disconnection", () => {
|
||||
it("child error event rejects all pending commands with 'IPC disconnected' error and emits 'disconnect'", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const disconnectHandler = vi.fn();
|
||||
host.on("disconnect", disconnectHandler);
|
||||
|
||||
const promise = host.sendCommand("GET_STATUS", {});
|
||||
expect(host.getPendingCommandCount()).toBe(1);
|
||||
|
||||
child.emit("error", new Error("child crash"));
|
||||
|
||||
await expect(promise).rejects.toThrow("IPC disconnected");
|
||||
expect(host.getPendingCommandCount()).toBe(0);
|
||||
expect(disconnectHandler).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("child exit event (with code) triggers disconnection", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const disconnectHandler = vi.fn();
|
||||
host.on("disconnect", disconnectHandler);
|
||||
|
||||
const promise = host.sendCommand("GET_STATUS", {});
|
||||
|
||||
child.emit("exit", 1, null);
|
||||
|
||||
await expect(promise).rejects.toThrow("IPC disconnected");
|
||||
expect(disconnectHandler).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("child exit event (with signal) triggers disconnection", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const disconnectHandler = vi.fn();
|
||||
host.on("disconnect", disconnectHandler);
|
||||
|
||||
const promise = host.sendCommand("GET_STATUS", {});
|
||||
|
||||
child.emit("exit", null, "SIGTERM");
|
||||
|
||||
await expect(promise).rejects.toThrow("IPC disconnected");
|
||||
expect(disconnectHandler).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("child disconnect event triggers disconnection", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const disconnectHandler = vi.fn();
|
||||
host.on("disconnect", disconnectHandler);
|
||||
|
||||
const promise = host.sendCommand("GET_STATUS", {});
|
||||
|
||||
child.emit("disconnect");
|
||||
|
||||
await expect(promise).rejects.toThrow("IPC disconnected");
|
||||
expect(disconnectHandler).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("double disconnection is idempotent (no re-reject or double-emit)", () => {
|
||||
const disconnectHandler = vi.fn();
|
||||
host.on("disconnect", disconnectHandler);
|
||||
|
||||
child.emit("disconnect");
|
||||
child.emit("disconnect");
|
||||
|
||||
expect(disconnectHandler).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("disconnect() method rejects pending commands, calls childProcess.disconnect(), removes all listeners", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const promise = host.sendCommand("GET_STATUS", {});
|
||||
expect(host.getPendingCommandCount()).toBe(1);
|
||||
|
||||
host.disconnect();
|
||||
|
||||
await expect(promise).rejects.toThrow("IPC disconnected");
|
||||
expect(host.getPendingCommandCount()).toBe(0);
|
||||
expect((child as any).disconnect).toHaveBeenCalled();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("disconnect() skips childProcess.disconnect() when already disconnected", () => {
|
||||
// Simulate child already disconnected
|
||||
Object.defineProperty(child, "connected", {
|
||||
get: () => false,
|
||||
configurable: true,
|
||||
});
|
||||
|
||||
host.disconnect();
|
||||
// disconnect() should not call child.disconnect() since connected is false
|
||||
expect((child as any).disconnect).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
175
packages/engine/src/ipc/__tests__/ipc-protocol.test.ts
Normal file
175
packages/engine/src/ipc/__tests__/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();
|
||||
});
|
||||
});
|
||||
});
|
||||
510
packages/engine/src/ipc/__tests__/ipc-worker.test.ts
Normal file
510
packages/engine/src/ipc/__tests__/ipc-worker.test.ts
Normal file
@@ -0,0 +1,510 @@
|
||||
/**
|
||||
* Unit tests for IpcWorker — the child-process-side IPC handler that receives
|
||||
* commands from a host, dispatches to registered handlers, and sends responses/events.
|
||||
*
|
||||
* Coverage:
|
||||
* - Constructor: process.send validation, listener registration, initial state
|
||||
* - PING auto-response (no handler needed)
|
||||
* - onCommand / offCommand: handler registration and dispatch
|
||||
* - Command execution: OK response, ERROR response (Error and non-Error), NO_HANDLER, UNKNOWN_COMMAND, MALFORMED_MESSAGE
|
||||
* - sendEvent / sendErrorEvent: event message construction
|
||||
* - sendResponse: response message construction
|
||||
* - shutdown: idempotent, suppresses further sends, emits event
|
||||
* - disconnect event forwarding
|
||||
* - Edge cases: process.send undefined after construction, graceful fallback
|
||||
*/
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import { PING, PONG, OK, ERROR, TASK_CREATED, ERROR_EVENT } from "../ipc-protocol.js";
|
||||
import { ipcLog } from "../../logger.js";
|
||||
|
||||
// ── Mock logger to suppress console output ──────────────────────────────
|
||||
vi.mock("../../logger.js", () => ({
|
||||
ipcLog: {
|
||||
log: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
},
|
||||
}));
|
||||
|
||||
// ── Process mock utilities ──────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* We need to mock process.send and intercept process.on("message") handlers
|
||||
* without breaking the real process. Strategy:
|
||||
* - Set process.send to a vi.fn() before creating IpcWorker
|
||||
* - Track message handlers registered via process.on("message")
|
||||
* - Simulate incoming messages by calling those handlers directly
|
||||
*/
|
||||
|
||||
// Store the original process.send to restore after tests
|
||||
const originalProcessSend = process.send;
|
||||
|
||||
// Track registered message/disconnect handlers so we can invoke them
|
||||
let messageHandlers: Array<(msg: unknown) => void> = [];
|
||||
let disconnectHandlers: Array<() => void> = [];
|
||||
|
||||
// Spies for process.on and process.removeAllListeners
|
||||
let processOnSpy: ReturnType<typeof vi.fn>;
|
||||
|
||||
function setupProcessMocks() {
|
||||
// Set up process.send as a mock function
|
||||
process.send = vi.fn((_msg: unknown, _handle?: unknown, _options?: unknown, callback?: (err: Error | null) => void) => {
|
||||
if (typeof callback === "function") callback(null);
|
||||
return true;
|
||||
});
|
||||
|
||||
messageHandlers = [];
|
||||
disconnectHandlers = [];
|
||||
|
||||
// Intercept process.on to capture message/disconnect handlers
|
||||
const originalProcessOn = process.on.bind(process);
|
||||
processOnSpy = vi.fn((event: string, handler: (...args: any[]) => void) => {
|
||||
if (event === "message") {
|
||||
messageHandlers.push(handler);
|
||||
} else if (event === "disconnect") {
|
||||
disconnectHandlers.push(handler);
|
||||
}
|
||||
// Don't register signal handlers on real process during tests
|
||||
if (event === "SIGTERM" || event === "SIGINT" || event === "uncaughtException" || event === "unhandledRejection") {
|
||||
return process;
|
||||
}
|
||||
return originalProcessOn(event, handler);
|
||||
});
|
||||
process.on = processOnSpy as any;
|
||||
}
|
||||
|
||||
function teardownProcessMocks() {
|
||||
// Restore process.send
|
||||
if (originalProcessSend === undefined) {
|
||||
delete (process as any).send;
|
||||
} else {
|
||||
process.send = originalProcessSend;
|
||||
}
|
||||
|
||||
// Remove any listeners we added during the test
|
||||
for (const handler of messageHandlers) {
|
||||
process.removeListener("message", handler);
|
||||
}
|
||||
for (const handler of disconnectHandlers) {
|
||||
process.removeListener("disconnect", handler);
|
||||
}
|
||||
messageHandlers = [];
|
||||
disconnectHandlers = [];
|
||||
}
|
||||
|
||||
/** Simulate an incoming message from the host */
|
||||
function simulateMessage(msg: unknown) {
|
||||
for (const handler of messageHandlers) {
|
||||
handler(msg);
|
||||
}
|
||||
}
|
||||
|
||||
/** Simulate a disconnect event */
|
||||
function simulateDisconnect() {
|
||||
for (const handler of disconnectHandlers) {
|
||||
handler();
|
||||
}
|
||||
}
|
||||
|
||||
// ── Tests ────────────────────────────────────────────────────────────────
|
||||
|
||||
describe("IpcWorker", () => {
|
||||
// We need to dynamically import IpcWorker after mocks are set up
|
||||
let IpcWorker: typeof import("../ipc-worker.js").IpcWorker;
|
||||
|
||||
beforeEach(async () => {
|
||||
setupProcessMocks();
|
||||
// Dynamic import to get fresh module (the mock setup needs to be in place)
|
||||
const mod = await import("../ipc-worker.js");
|
||||
IpcWorker = mod.IpcWorker;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
teardownProcessMocks();
|
||||
});
|
||||
|
||||
// ── Constructor & initial state ──────────────────────────────────────
|
||||
|
||||
describe("constructor and initial state", () => {
|
||||
it("throws when process.send is undefined", async () => {
|
||||
teardownProcessMocks(); // Remove mock
|
||||
// Ensure process.send is undefined
|
||||
delete (process as any).send;
|
||||
|
||||
expect(() => new IpcWorker()).toThrow(
|
||||
"IpcWorker can only be instantiated in a forked child process"
|
||||
);
|
||||
|
||||
// Re-set up for afterEach
|
||||
setupProcessMocks();
|
||||
const mod = await import("../ipc-worker.js");
|
||||
IpcWorker = mod.IpcWorker;
|
||||
});
|
||||
|
||||
it("registers listeners on process for message and disconnect events", () => {
|
||||
const worker = new IpcWorker();
|
||||
expect(messageHandlers.length).toBeGreaterThanOrEqual(1);
|
||||
expect(disconnectHandlers.length).toBeGreaterThanOrEqual(1);
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("getHandlerCount() returns 0 initially", () => {
|
||||
const worker = new IpcWorker();
|
||||
expect(worker.getHandlerCount()).toBe(0);
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("isShuttingDown() returns false initially", () => {
|
||||
const worker = new IpcWorker();
|
||||
expect(worker.isShuttingDown()).toBe(false);
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* Helper: creates a worker and returns it along with its dedicated message handler.
|
||||
* Also clears the process.send mock so each test starts fresh.
|
||||
*/
|
||||
function createWorker() {
|
||||
const msgCountBefore = messageHandlers.length;
|
||||
const discCountBefore = disconnectHandlers.length;
|
||||
const worker = new IpcWorker();
|
||||
const sendFn = process.send as ReturnType<typeof vi.fn>;
|
||||
sendFn.mockClear();
|
||||
|
||||
// The worker's handlers are the ones added after the counts
|
||||
const workerMsgHandler = messageHandlers[messageHandlers.length - 1];
|
||||
const workerDiscHandler = disconnectHandlers[disconnectHandlers.length - 1];
|
||||
|
||||
/** Send a message to this worker's handler */
|
||||
const sendMessage = (msg: unknown) => workerMsgHandler(msg);
|
||||
|
||||
/** Simulate disconnect for this specific worker */
|
||||
const triggerDisconnect = () => workerDiscHandler?.();
|
||||
|
||||
/** Get all messages sent to parent via process.send since last clear */
|
||||
const getSentMessages = () => sendFn.mock.calls.map((call: any[]) => call[0]);
|
||||
|
||||
/** Find the first sent message matching a type */
|
||||
const findSent = (type: string) =>
|
||||
sendFn.mock.calls.find((call: any[]) => call[0]?.type === type)?.[0];
|
||||
|
||||
return { worker, sendMessage, triggerDisconnect, sendFn, getSentMessages, findSent };
|
||||
}
|
||||
|
||||
// ── PING auto-response ──────────────────────────────────────────────
|
||||
|
||||
describe("PING handling", () => {
|
||||
it("incoming PING message automatically responds with PONG containing a timestamp", async () => {
|
||||
const { worker, sendMessage, findSent } = createWorker();
|
||||
|
||||
sendMessage({ type: PING, id: "ping-1", payload: {} });
|
||||
|
||||
// handleMessage is async, give it a tick
|
||||
await vi.waitFor(() => {
|
||||
expect(findSent(PONG)).toBeDefined();
|
||||
});
|
||||
|
||||
const response = findSent(PONG);
|
||||
expect(response.type).toBe(PONG);
|
||||
expect(response.id).toBe("ping-1");
|
||||
expect(typeof response.payload.timestamp).toBe("string");
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
});
|
||||
|
||||
// ── Command handling ────────────────────────────────────────────────
|
||||
|
||||
describe("command handling", () => {
|
||||
it("onCommand() registers a handler: getHandlerCount() increments", () => {
|
||||
const { worker } = createWorker();
|
||||
expect(worker.getHandlerCount()).toBe(0);
|
||||
|
||||
worker.onCommand("START_RUNTIME", async () => ({ success: true }));
|
||||
expect(worker.getHandlerCount()).toBe(1);
|
||||
|
||||
worker.onCommand("STOP_RUNTIME", async () => {});
|
||||
expect(worker.getHandlerCount()).toBe(2);
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("offCommand() removes a handler: getHandlerCount() decrements", () => {
|
||||
const { worker } = createWorker();
|
||||
worker.onCommand("START_RUNTIME", async () => {});
|
||||
expect(worker.getHandlerCount()).toBe(1);
|
||||
|
||||
worker.offCommand("START_RUNTIME");
|
||||
expect(worker.getHandlerCount()).toBe(0);
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("receiving a registered command invokes the handler with the message payload", async () => {
|
||||
const handler = vi.fn().mockResolvedValue("ok");
|
||||
const { worker, sendMessage } = createWorker();
|
||||
worker.onCommand("GET_STATUS", handler);
|
||||
|
||||
const payload = { detail: "test" };
|
||||
sendMessage({ type: "GET_STATUS", id: "cmd-1", payload });
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(handler).toHaveBeenCalledWith(payload);
|
||||
});
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("handler returning a value sends OK response with { data: returnValue }", async () => {
|
||||
const { worker, sendMessage, findSent } = createWorker();
|
||||
worker.onCommand("GET_METRICS", async () => ({ tasks: 10 }));
|
||||
|
||||
sendMessage({ type: "GET_METRICS", id: "cmd-2", payload: {} });
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(findSent(OK)).toBeDefined();
|
||||
});
|
||||
|
||||
const response = findSent(OK);
|
||||
expect(response.type).toBe(OK);
|
||||
expect(response.id).toBe("cmd-2");
|
||||
expect(response.payload).toEqual({ data: { tasks: 10 } });
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("handler throwing an error sends ERROR response with { message, code: 'HANDLER_ERROR' }", async () => {
|
||||
const { worker, sendMessage, findSent } = createWorker();
|
||||
worker.onCommand("GET_STATUS", async () => {
|
||||
throw new Error("Something broke");
|
||||
});
|
||||
|
||||
sendMessage({ type: "GET_STATUS", id: "cmd-3", payload: {} });
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(findSent(ERROR)).toBeDefined();
|
||||
});
|
||||
|
||||
const response = findSent(ERROR);
|
||||
expect(response.type).toBe(ERROR);
|
||||
expect(response.id).toBe("cmd-3");
|
||||
expect(response.payload.message).toBe("Something broke");
|
||||
expect(response.payload.code).toBe("HANDLER_ERROR");
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("handler throwing a non-Error value still sends ERROR response with stringified message", async () => {
|
||||
const { worker, sendMessage, findSent } = createWorker();
|
||||
worker.onCommand("GET_STATUS", async () => {
|
||||
throw "string error";
|
||||
});
|
||||
|
||||
sendMessage({ type: "GET_STATUS", id: "cmd-4", payload: {} });
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(findSent(ERROR)).toBeDefined();
|
||||
});
|
||||
|
||||
const response = findSent(ERROR);
|
||||
expect(response.type).toBe(ERROR);
|
||||
expect(response.id).toBe("cmd-4");
|
||||
expect(response.payload.message).toBe("string error");
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("receiving a command with no registered handler sends ERROR with code: 'NO_HANDLER'", async () => {
|
||||
const { worker, sendMessage, findSent } = createWorker();
|
||||
// Don't register any handler for START_RUNTIME
|
||||
sendMessage({ type: "START_RUNTIME", id: "cmd-5", payload: {} });
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(findSent(ERROR)).toBeDefined();
|
||||
});
|
||||
|
||||
const response = findSent(ERROR);
|
||||
expect(response.payload.code).toBe("NO_HANDLER");
|
||||
expect(response.id).toBe("cmd-5");
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("receiving a non-command (unknown type) sends ERROR with code: 'UNKNOWN_COMMAND'", async () => {
|
||||
const { worker, sendMessage, findSent } = createWorker();
|
||||
sendMessage({ type: "TOTALLY_UNKNOWN", id: "cmd-6", payload: {} });
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(findSent(ERROR)).toBeDefined();
|
||||
});
|
||||
|
||||
const response = findSent(ERROR);
|
||||
expect(response.payload.code).toBe("UNKNOWN_COMMAND");
|
||||
expect(response.id).toBe("cmd-6");
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("receiving a malformed message (not a valid IpcMessage) sends ERROR with code: 'MALFORMED_MESSAGE'", async () => {
|
||||
const { worker, sendMessage, findSent } = createWorker();
|
||||
sendMessage({ noType: true }); // Missing type, id, payload
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(findSent(ERROR)).toBeDefined();
|
||||
});
|
||||
|
||||
const response = findSent(ERROR);
|
||||
expect(response.payload.code).toBe("MALFORMED_MESSAGE");
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
});
|
||||
|
||||
// ── sendEvent / sendErrorEvent ──────────────────────────────────────
|
||||
|
||||
describe("sendEvent and sendErrorEvent", () => {
|
||||
it("sendEvent() sends an IpcMessage with the given event type, a generated correlation ID, and payload", () => {
|
||||
const { worker, sendFn } = createWorker();
|
||||
|
||||
worker.sendEvent(TASK_CREATED, { task: { id: "KB-001" } });
|
||||
|
||||
expect(sendFn).toHaveBeenCalledTimes(1);
|
||||
const msg = sendFn.mock.calls[0][0];
|
||||
expect(msg.type).toBe(TASK_CREATED);
|
||||
expect(typeof msg.id).toBe("string");
|
||||
expect(msg.id.length).toBeGreaterThan(0);
|
||||
expect(msg.payload).toEqual({ task: { id: "KB-001" } });
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("sendErrorEvent() sends an ERROR_EVENT typed message with error message and code", () => {
|
||||
const { worker, sendFn } = createWorker();
|
||||
|
||||
const err = new Error("Runtime crashed");
|
||||
(err as any).code = "RUNTIME_ERROR";
|
||||
worker.sendErrorEvent(err);
|
||||
|
||||
expect(sendFn).toHaveBeenCalledTimes(1);
|
||||
const msg = sendFn.mock.calls[0][0];
|
||||
expect(msg.type).toBe(ERROR_EVENT);
|
||||
expect(msg.payload).toEqual({
|
||||
message: "Runtime crashed",
|
||||
code: "RUNTIME_ERROR",
|
||||
});
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
});
|
||||
|
||||
// ── Shutdown ────────────────────────────────────────────────────────
|
||||
|
||||
describe("shutdown", () => {
|
||||
it("sets isShuttingDown() to true", () => {
|
||||
const { worker } = createWorker();
|
||||
expect(worker.isShuttingDown()).toBe(false);
|
||||
worker.shutdown();
|
||||
expect(worker.isShuttingDown()).toBe(true);
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("sends a SHUTDOWN message to parent via process.send", () => {
|
||||
const { worker, sendFn } = createWorker();
|
||||
worker.shutdown();
|
||||
|
||||
expect(sendFn).toHaveBeenCalledTimes(1);
|
||||
const msg = sendFn.mock.calls[0][0];
|
||||
expect(msg.type).toBe("SHUTDOWN");
|
||||
expect(typeof msg.id).toBe("string");
|
||||
expect(msg.payload).toEqual({});
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("logs warning when process.send throws during shutdown", () => {
|
||||
const { worker, sendFn } = createWorker();
|
||||
vi.mocked(ipcLog.warn).mockClear();
|
||||
|
||||
sendFn.mockImplementation(() => {
|
||||
throw new Error("channel closed");
|
||||
});
|
||||
|
||||
worker.shutdown();
|
||||
|
||||
expect(vi.mocked(ipcLog.warn)).toHaveBeenCalledWith(
|
||||
expect.stringContaining("Failed to send SHUTDOWN message to parent: channel closed"),
|
||||
);
|
||||
expect(worker.isShuttingDown()).toBe(true);
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it('emits "shutdown" event on the IpcWorker instance', () => {
|
||||
const { worker } = createWorker();
|
||||
const handler = vi.fn();
|
||||
worker.on("shutdown", handler);
|
||||
worker.shutdown();
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("is idempotent (calling twice only sends one SHUTDOWN message)", () => {
|
||||
const { worker, sendFn } = createWorker();
|
||||
worker.shutdown();
|
||||
worker.shutdown();
|
||||
|
||||
// Only one SHUTDOWN message should be sent
|
||||
expect(sendFn).toHaveBeenCalledTimes(1);
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("after shutdown(), sendEvent() and sendResponse() are no-ops", () => {
|
||||
const { worker, sendFn } = createWorker();
|
||||
worker.shutdown();
|
||||
sendFn.mockClear();
|
||||
|
||||
worker.sendEvent(TASK_CREATED, { task: {} });
|
||||
worker.sendResponse(OK, "some-id", { data: null });
|
||||
|
||||
expect(sendFn).not.toHaveBeenCalled();
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
});
|
||||
|
||||
// ── Disconnect ──────────────────────────────────────────────────────
|
||||
|
||||
describe("disconnect", () => {
|
||||
it('process disconnect event emits "disconnect" on IpcWorker', () => {
|
||||
const { worker, triggerDisconnect } = createWorker();
|
||||
const handler = vi.fn();
|
||||
worker.on("disconnect", handler);
|
||||
|
||||
triggerDisconnect();
|
||||
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
});
|
||||
|
||||
// ── Edge cases ──────────────────────────────────────────────────────
|
||||
|
||||
describe("edge cases", () => {
|
||||
it("sendEvent() when process.send is undefined does not throw (graceful fallback)", () => {
|
||||
const { worker } = createWorker();
|
||||
|
||||
// Remove process.send after construction
|
||||
const savedSend = process.send;
|
||||
delete (process as any).send;
|
||||
|
||||
expect(() => {
|
||||
worker.sendEvent(TASK_CREATED, { task: {} });
|
||||
}).not.toThrow();
|
||||
|
||||
// Restore
|
||||
process.send = savedSend;
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
|
||||
it("sendResponse() sends correctly structured IpcMessage with type, id, and payload", () => {
|
||||
const { worker, sendFn } = createWorker();
|
||||
|
||||
worker.sendResponse(OK, "resp-id-1", { data: { status: "active" } });
|
||||
|
||||
expect(sendFn).toHaveBeenCalledTimes(1);
|
||||
const msg = sendFn.mock.calls[0][0];
|
||||
expect(msg).toEqual({
|
||||
type: OK,
|
||||
id: "resp-id-1",
|
||||
payload: { data: { status: "active" } },
|
||||
});
|
||||
worker.removeAllListeners();
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,715 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import type { CentralCore, Task } from "@fusion/core";
|
||||
import { ChildProcessRuntime } from "../child-process-runtime.js";
|
||||
import type {
|
||||
ProjectRuntimeConfig,
|
||||
RuntimeMetrics,
|
||||
RuntimeStatus,
|
||||
} from "../../project-runtime.js";
|
||||
import { runtimeLog } from "../../logger.js";
|
||||
import {
|
||||
START_RUNTIME,
|
||||
STOP_RUNTIME,
|
||||
GET_METRICS,
|
||||
TASK_CREATED,
|
||||
TASK_MOVED,
|
||||
TASK_UPDATED,
|
||||
ERROR_EVENT,
|
||||
HEALTH_CHANGED,
|
||||
OK,
|
||||
ERROR,
|
||||
PONG,
|
||||
} from "../../ipc/ipc-protocol.js";
|
||||
|
||||
type Listener = (...args: any[]) => void;
|
||||
|
||||
type CommandMessage = {
|
||||
type: string;
|
||||
id: string;
|
||||
payload: unknown;
|
||||
};
|
||||
|
||||
type MockChildOptions = {
|
||||
pingResults?: boolean[];
|
||||
metricsResponse?: RuntimeMetrics;
|
||||
sendCallbackErrors?: Partial<Record<string, Error>>;
|
||||
markKilledOnSigterm?: boolean;
|
||||
emitExitOnKill?: boolean;
|
||||
};
|
||||
|
||||
type MockChildProcess = {
|
||||
on: ReturnType<typeof vi.fn>;
|
||||
send: ReturnType<typeof vi.fn>;
|
||||
kill: ReturnType<typeof vi.fn>;
|
||||
disconnect: ReturnType<typeof vi.fn>;
|
||||
emit: (event: string, ...args: unknown[]) => void;
|
||||
connected: boolean;
|
||||
killed: boolean;
|
||||
sentMessages: CommandMessage[];
|
||||
};
|
||||
|
||||
const forkedChildren: MockChildProcess[] = [];
|
||||
const queuedForkOptions: MockChildOptions[] = [];
|
||||
|
||||
function createMockChildProcess(options: MockChildOptions = {}): MockChildProcess {
|
||||
const listeners = new Map<string, Listener[]>();
|
||||
const pingResults = [...(options.pingResults ?? [])];
|
||||
|
||||
const child: MockChildProcess = {
|
||||
on: vi.fn((event: string, handler: Listener) => {
|
||||
const existing = listeners.get(event) ?? [];
|
||||
existing.push(handler);
|
||||
listeners.set(event, existing);
|
||||
return child;
|
||||
}),
|
||||
send: vi.fn((message: CommandMessage, callback?: (error: Error | null) => void) => {
|
||||
child.sentMessages.push(message);
|
||||
|
||||
const sendError = options.sendCallbackErrors?.[message.type];
|
||||
if (sendError) {
|
||||
callback?.(sendError);
|
||||
return false;
|
||||
}
|
||||
|
||||
callback?.(null);
|
||||
|
||||
const respond = (type: string, payload: unknown) => {
|
||||
Promise.resolve().then(() => {
|
||||
child.emit("message", {
|
||||
type,
|
||||
id: message.id,
|
||||
payload,
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
if (message.type === START_RUNTIME) {
|
||||
respond(OK, { data: { status: "active" } });
|
||||
} else if (message.type === STOP_RUNTIME) {
|
||||
respond(OK, { data: { stopped: true } });
|
||||
} else if (message.type === GET_METRICS) {
|
||||
respond(OK, {
|
||||
data:
|
||||
options.metricsResponse ??
|
||||
{
|
||||
inFlightTasks: 4,
|
||||
activeAgents: 2,
|
||||
lastActivityAt: "2026-04-08T00:00:00.000Z",
|
||||
},
|
||||
});
|
||||
} else if (message.type === "PING") {
|
||||
const pingOk = pingResults.shift() ?? true;
|
||||
if (pingOk) {
|
||||
respond(PONG, { timestamp: "2026-04-08T00:00:00.000Z" });
|
||||
} else {
|
||||
respond(ERROR, { message: "Ping failed", code: "PING_FAILED" });
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}),
|
||||
kill: vi.fn((signal?: string | number) => {
|
||||
if (signal === "SIGKILL" || (signal === "SIGTERM" && options.markKilledOnSigterm !== false)) {
|
||||
child.killed = true;
|
||||
}
|
||||
|
||||
if (options.emitExitOnKill) {
|
||||
child.emit("exit", signal === "SIGKILL" ? 137 : 0, typeof signal === "string" ? signal : null);
|
||||
}
|
||||
|
||||
return true;
|
||||
}),
|
||||
disconnect: vi.fn(() => {
|
||||
child.connected = false;
|
||||
child.emit("disconnect");
|
||||
}),
|
||||
emit: (event: string, ...args: unknown[]) => {
|
||||
for (const handler of listeners.get(event) ?? []) {
|
||||
handler(...(args as any[]));
|
||||
}
|
||||
},
|
||||
connected: true,
|
||||
killed: false,
|
||||
sentMessages: [],
|
||||
};
|
||||
|
||||
return child;
|
||||
}
|
||||
|
||||
const mockFork = vi.fn(() => {
|
||||
const options = queuedForkOptions.shift() ?? {};
|
||||
const child = createMockChildProcess(options);
|
||||
forkedChildren.push(child);
|
||||
return child;
|
||||
});
|
||||
|
||||
vi.mock("node:child_process", () => ({
|
||||
fork: (...args: unknown[]) => (mockFork as (...mockArgs: unknown[]) => unknown)(...args),
|
||||
}));
|
||||
|
||||
function queueChild(options: MockChildOptions = {}): void {
|
||||
queuedForkOptions.push(options);
|
||||
}
|
||||
|
||||
function getLatestChild(): MockChildProcess {
|
||||
const child = forkedChildren.at(-1);
|
||||
if (!child) {
|
||||
throw new Error("Expected a forked child process");
|
||||
}
|
||||
return child;
|
||||
}
|
||||
|
||||
function getMessages(child: MockChildProcess, type: string): CommandMessage[] {
|
||||
return child.sentMessages.filter((message) => message.type === type);
|
||||
}
|
||||
|
||||
function createMockTask(id: string): Task {
|
||||
return {
|
||||
id,
|
||||
title: `${id} title`,
|
||||
description: `${id} description`,
|
||||
column: "todo",
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
createdAt: "2026-04-08T00:00:00.000Z",
|
||||
updatedAt: "2026-04-08T00:00:00.000Z",
|
||||
size: "M",
|
||||
reviewLevel: 1,
|
||||
log: [],
|
||||
attachments: [],
|
||||
} as Task;
|
||||
}
|
||||
|
||||
describe("ChildProcessRuntime", () => {
|
||||
let runtime: ChildProcessRuntime;
|
||||
let runtimeAny: any;
|
||||
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_test123",
|
||||
workingDirectory: "/tmp/test-project",
|
||||
isolationMode: "child-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
mockFork.mockClear();
|
||||
forkedChildren.length = 0;
|
||||
queuedForkOptions.length = 0;
|
||||
|
||||
const mockCentralCore = {
|
||||
getGlobalConcurrencyState: vi.fn().mockResolvedValue({
|
||||
globalMaxConcurrent: 4,
|
||||
currentlyActive: 0,
|
||||
queuedCount: 0,
|
||||
projectsActive: {},
|
||||
}),
|
||||
} as unknown as CentralCore;
|
||||
|
||||
runtime = new ChildProcessRuntime(testConfig, mockCentralCore);
|
||||
runtimeAny = runtime as any;
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
await runtime.stop();
|
||||
} catch {
|
||||
// Ignore cleanup failures
|
||||
}
|
||||
|
||||
vi.useRealTimers();
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
describe("startup sequence", () => {
|
||||
it("transitions stopped → starting → active, forks worker path, and sends START_RUNTIME config", async () => {
|
||||
queueChild();
|
||||
|
||||
const transitions: RuntimeStatus[] = [];
|
||||
runtime.on("health-changed", (data) => transitions.push(data.status));
|
||||
|
||||
await runtime.start();
|
||||
|
||||
const child = getLatestChild();
|
||||
|
||||
expect(transitions).toEqual(["starting", "active"]);
|
||||
expect(runtime.getStatus()).toBe("active");
|
||||
expect(mockFork).toHaveBeenCalledWith(
|
||||
expect.stringMatching(/child-process-worker\.(ts|js)$/),
|
||||
[],
|
||||
expect.objectContaining({
|
||||
silent: true,
|
||||
execArgv: [],
|
||||
})
|
||||
);
|
||||
|
||||
const startMessages = getMessages(child, START_RUNTIME);
|
||||
expect(startMessages).toHaveLength(1);
|
||||
expect(startMessages[0]?.payload).toEqual({ config: testConfig });
|
||||
});
|
||||
|
||||
it("sets status to errored and emits error when startup fails", async () => {
|
||||
queueChild({
|
||||
sendCallbackErrors: {
|
||||
[START_RUNTIME]: new Error("start send failed"),
|
||||
},
|
||||
});
|
||||
|
||||
const errorSpy = vi.fn();
|
||||
runtime.on("error", errorSpy);
|
||||
|
||||
await expect(runtime.start()).rejects.toThrow("Failed to send command: start send failed");
|
||||
expect(runtime.getStatus()).toBe("errored");
|
||||
expect(errorSpy).toHaveBeenCalledTimes(1);
|
||||
expect(errorSpy.mock.calls[0]?.[0]).toBeInstanceOf(Error);
|
||||
});
|
||||
|
||||
it("throws when start() is called in non-stopped states", async () => {
|
||||
const blockedStates: RuntimeStatus[] = ["starting", "active", "stopping"];
|
||||
|
||||
for (const status of blockedStates) {
|
||||
runtimeAny.status = status;
|
||||
await expect(runtime.start()).rejects.toThrow(`Cannot start runtime: current status is ${status}`);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("shutdown sequence", () => {
|
||||
it("transitions active → stopping → stopped and sends STOP_RUNTIME with timeout", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
const transitions: RuntimeStatus[] = [];
|
||||
runtime.on("health-changed", (data) => transitions.push(data.status));
|
||||
|
||||
await runtime.stop();
|
||||
|
||||
expect(transitions).toEqual(["stopping", "stopped"]);
|
||||
expect(runtime.getStatus()).toBe("stopped");
|
||||
expect(getMessages(child, STOP_RUNTIME)).toHaveLength(1);
|
||||
expect(getMessages(child, STOP_RUNTIME)[0]?.payload).toEqual({ timeoutMs: 30000 });
|
||||
expect(child.kill).toHaveBeenCalledWith("SIGTERM");
|
||||
});
|
||||
|
||||
it("is idempotent and does not send duplicate STOP_RUNTIME commands", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
await runtime.stop();
|
||||
await runtime.stop();
|
||||
|
||||
expect(getMessages(child, STOP_RUNTIME)).toHaveLength(1);
|
||||
expect(child.kill).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("returns without error when stop() is called while already stopped", async () => {
|
||||
await expect(runtime.stop()).resolves.toBeUndefined();
|
||||
expect(runtime.getStatus()).toBe("stopped");
|
||||
});
|
||||
|
||||
it("handles stop() gracefully when IPC is already disconnected", async () => {
|
||||
queueChild();
|
||||
runtime.on("error", () => {
|
||||
// swallow asynchronous error events from disconnection path
|
||||
});
|
||||
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
child.connected = false;
|
||||
child.emit("disconnect");
|
||||
|
||||
await expect(runtime.stop()).resolves.toBeUndefined();
|
||||
expect(runtime.getStatus()).toBe("stopped");
|
||||
});
|
||||
|
||||
it("force-kills with SIGKILL after 5s timeout when child remains alive", async () => {
|
||||
vi.useFakeTimers();
|
||||
queueChild({ markKilledOnSigterm: false });
|
||||
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
await runtime.stop();
|
||||
|
||||
// Keep a live child reference so the delayed SIGKILL callback can execute the force-kill path.
|
||||
runtimeAny.child = child;
|
||||
child.killed = false;
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
|
||||
expect(child.kill).toHaveBeenCalledWith("SIGTERM");
|
||||
expect(child.kill).toHaveBeenCalledWith("SIGKILL");
|
||||
});
|
||||
});
|
||||
|
||||
describe("health monitoring and restart", () => {
|
||||
it("starts health monitoring after start() and performs periodic pings", async () => {
|
||||
vi.useFakeTimers();
|
||||
queueChild({ pingResults: [true, true] });
|
||||
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
expect(getMessages(child, "PING")).toHaveLength(0);
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
expect(getMessages(child, "PING")).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("resets missed heartbeat count to 0 after a successful ping", async () => {
|
||||
vi.useFakeTimers();
|
||||
queueChild({ pingResults: [false, true] });
|
||||
runtime.on("error", () => {
|
||||
// swallow
|
||||
});
|
||||
|
||||
await runtime.start();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
expect(runtimeAny.healthMonitor.getMissedHeartbeats()).toBe(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
expect(runtimeAny.healthMonitor.getMissedHeartbeats()).toBe(0);
|
||||
});
|
||||
|
||||
it("triggers handleUnhealthy after three missed heartbeats", async () => {
|
||||
vi.useFakeTimers();
|
||||
queueChild({ pingResults: [false, false, false] });
|
||||
|
||||
const unhealthySpy = vi.spyOn(runtimeAny, "handleUnhealthy").mockImplementation(() => {});
|
||||
|
||||
await runtime.start();
|
||||
await vi.advanceTimersByTimeAsync(15000);
|
||||
|
||||
expect(unhealthySpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("uses exponential restart delays: 1000ms, 5000ms, 15000ms", () => {
|
||||
vi.useFakeTimers();
|
||||
runtimeAny.status = "active";
|
||||
|
||||
const timeoutSpy = vi.spyOn(globalThis, "setTimeout");
|
||||
|
||||
runtimeAny.handleUnhealthy();
|
||||
runtimeAny.handleUnhealthy();
|
||||
runtimeAny.handleUnhealthy();
|
||||
|
||||
const delays = timeoutSpy.mock.calls.map((call) => Number(call[1]));
|
||||
expect(delays.slice(0, 3)).toEqual([1000, 5000, 15000]);
|
||||
});
|
||||
|
||||
it("transitions to errored and emits error after max restart attempts", () => {
|
||||
runtimeAny.status = "active";
|
||||
|
||||
const errorSpy = vi.fn();
|
||||
runtime.on("error", errorSpy);
|
||||
|
||||
runtimeAny.handleUnhealthy();
|
||||
runtimeAny.handleUnhealthy();
|
||||
runtimeAny.handleUnhealthy();
|
||||
runtimeAny.handleUnhealthy();
|
||||
|
||||
expect(runtime.getStatus()).toBe("errored");
|
||||
expect(errorSpy).toHaveBeenCalledTimes(1);
|
||||
expect(errorSpy.mock.calls[0]?.[0]).toBeInstanceOf(Error);
|
||||
expect((errorSpy.mock.calls[0]?.[0] as Error).message).toContain("max restart attempts");
|
||||
});
|
||||
|
||||
it("resets restart attempt counter after a successful health check", async () => {
|
||||
vi.useFakeTimers();
|
||||
queueChild({ pingResults: [true] });
|
||||
|
||||
await runtime.start();
|
||||
|
||||
runtimeAny.healthMonitor.incrementRestartAttempts();
|
||||
runtimeAny.healthMonitor.incrementRestartAttempts();
|
||||
expect(runtimeAny.healthMonitor.getRestartAttempts()).toBe(2);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
|
||||
expect(runtimeAny.healthMonitor.getRestartAttempts()).toBe(0);
|
||||
});
|
||||
|
||||
it("stops health checks after stop()", async () => {
|
||||
vi.useFakeTimers();
|
||||
queueChild({ pingResults: [true, true, true] });
|
||||
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
const pingCountBeforeStop = getMessages(child, "PING").length;
|
||||
|
||||
await runtime.stop();
|
||||
await vi.advanceTimersByTimeAsync(20000);
|
||||
|
||||
expect(getMessages(child, "PING").length).toBe(pingCountBeforeStop);
|
||||
});
|
||||
});
|
||||
|
||||
describe("child process exit and disconnect", () => {
|
||||
it("unexpected child exit while active triggers restart handling", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
|
||||
const child = getLatestChild();
|
||||
const unhealthySpy = vi.spyOn(runtimeAny, "handleUnhealthy").mockImplementation(() => {});
|
||||
|
||||
child.emit("exit", 1, null);
|
||||
|
||||
expect(unhealthySpy).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("child exit while stopping does not trigger restart", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
|
||||
const child = getLatestChild();
|
||||
const unhealthySpy = vi.spyOn(runtimeAny, "handleUnhealthy").mockImplementation(() => {});
|
||||
runtimeAny.status = "stopping";
|
||||
|
||||
child.emit("exit", 1, null);
|
||||
|
||||
expect(unhealthySpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("child exit while stopped does not trigger restart", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
|
||||
const child = getLatestChild();
|
||||
const unhealthySpy = vi.spyOn(runtimeAny, "handleUnhealthy").mockImplementation(() => {});
|
||||
runtimeAny.status = "stopped";
|
||||
|
||||
child.emit("exit", 1, null);
|
||||
|
||||
expect(unhealthySpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("IPC disconnect while active triggers restart handling", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
|
||||
const child = getLatestChild();
|
||||
const unhealthySpy = vi.spyOn(runtimeAny, "handleUnhealthy").mockImplementation(() => {});
|
||||
|
||||
child.emit("disconnect");
|
||||
|
||||
expect(unhealthySpy).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("IPC disconnect while stopping does not trigger restart", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
|
||||
const child = getLatestChild();
|
||||
const unhealthySpy = vi.spyOn(runtimeAny, "handleUnhealthy").mockImplementation(() => {});
|
||||
runtimeAny.status = "stopping";
|
||||
|
||||
child.emit("disconnect");
|
||||
|
||||
expect(unhealthySpy).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("event forwarding", () => {
|
||||
it("forwards TASK_CREATED as task:created", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
const task = createMockTask("FN-1279-A");
|
||||
const createdSpy = vi.fn();
|
||||
runtime.on("task:created", createdSpy);
|
||||
|
||||
child.emit("message", {
|
||||
type: TASK_CREATED,
|
||||
id: "evt-created",
|
||||
payload: { task },
|
||||
});
|
||||
|
||||
expect(createdSpy).toHaveBeenCalledWith(task);
|
||||
});
|
||||
|
||||
it("forwards TASK_MOVED as task:moved with { task, from, to } shape", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
const task = createMockTask("FN-1279-B");
|
||||
const movedSpy = vi.fn();
|
||||
runtime.on("task:moved", movedSpy);
|
||||
|
||||
child.emit("message", {
|
||||
type: TASK_MOVED,
|
||||
id: "evt-moved",
|
||||
payload: { task, from: "todo", to: "in-progress" },
|
||||
});
|
||||
|
||||
expect(movedSpy).toHaveBeenCalledWith({ task, from: "todo", to: "in-progress" });
|
||||
});
|
||||
|
||||
it("forwards TASK_UPDATED as task:updated", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
const task = createMockTask("FN-1279-C");
|
||||
const updatedSpy = vi.fn();
|
||||
runtime.on("task:updated", updatedSpy);
|
||||
|
||||
child.emit("message", {
|
||||
type: TASK_UPDATED,
|
||||
id: "evt-updated",
|
||||
payload: { task },
|
||||
});
|
||||
|
||||
expect(updatedSpy).toHaveBeenCalledWith(task);
|
||||
});
|
||||
|
||||
it("forwards ERROR_EVENT as Error instance and preserves error code", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
const errorSpy = vi.fn();
|
||||
runtime.on("error", errorSpy);
|
||||
|
||||
child.emit("message", {
|
||||
type: ERROR_EVENT,
|
||||
id: "evt-error",
|
||||
payload: { message: "worker failed", code: "WORKER_FAILURE" },
|
||||
});
|
||||
|
||||
expect(errorSpy).toHaveBeenCalledTimes(1);
|
||||
const forwardedError = errorSpy.mock.calls[0]?.[0] as Error & { code?: string };
|
||||
expect(forwardedError).toBeInstanceOf(Error);
|
||||
expect(forwardedError.message).toBe("worker failed");
|
||||
expect(forwardedError.code).toBe("WORKER_FAILURE");
|
||||
});
|
||||
|
||||
it("applies HEALTH_CHANGED payload to status and emits health-changed", async () => {
|
||||
queueChild();
|
||||
await runtime.start();
|
||||
const child = getLatestChild();
|
||||
|
||||
const healthSpy = vi.fn();
|
||||
runtime.on("health-changed", healthSpy);
|
||||
healthSpy.mockClear();
|
||||
|
||||
child.emit("message", {
|
||||
type: HEALTH_CHANGED,
|
||||
id: "evt-health",
|
||||
payload: { status: "paused", previous: "active" },
|
||||
});
|
||||
|
||||
expect(runtime.getStatus()).toBe("paused");
|
||||
expect(healthSpy).toHaveBeenCalledWith({ status: "paused", previous: "active" });
|
||||
});
|
||||
});
|
||||
|
||||
describe("metrics and inaccessible accessors", () => {
|
||||
it("returns cached metrics when IPC is disconnected", () => {
|
||||
runtimeAny.lastMetrics = {
|
||||
inFlightTasks: 9,
|
||||
activeAgents: 3,
|
||||
lastActivityAt: "2026-04-08T01:00:00.000Z",
|
||||
};
|
||||
|
||||
const metrics = runtime.getMetrics();
|
||||
|
||||
expect(metrics.inFlightTasks).toBe(9);
|
||||
expect(metrics.activeAgents).toBe(3);
|
||||
expect(typeof metrics.lastActivityAt).toBe("string");
|
||||
});
|
||||
|
||||
it("updates cached metrics when GET_METRICS response is received", async () => {
|
||||
queueChild({
|
||||
metricsResponse: {
|
||||
inFlightTasks: 12,
|
||||
activeAgents: 5,
|
||||
lastActivityAt: "2026-04-08T02:00:00.000Z",
|
||||
},
|
||||
});
|
||||
await runtime.start();
|
||||
|
||||
runtime.getMetrics();
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(runtimeAny.lastMetrics).toEqual({
|
||||
inFlightTasks: 12,
|
||||
activeAgents: 5,
|
||||
lastActivityAt: "2026-04-08T02:00:00.000Z",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
it("ignores GET_METRICS IPC errors and returns the last known metrics", async () => {
|
||||
queueChild({
|
||||
sendCallbackErrors: {
|
||||
[GET_METRICS]: new Error("metrics unavailable"),
|
||||
},
|
||||
});
|
||||
await runtime.start();
|
||||
|
||||
runtimeAny.lastMetrics = {
|
||||
inFlightTasks: 21,
|
||||
activeAgents: 8,
|
||||
lastActivityAt: "2026-04-08T03:00:00.000Z",
|
||||
};
|
||||
|
||||
const metrics = runtime.getMetrics();
|
||||
|
||||
expect(metrics.inFlightTasks).toBe(21);
|
||||
expect(metrics.activeAgents).toBe(8);
|
||||
|
||||
await Promise.resolve();
|
||||
expect(runtimeAny.lastMetrics).toEqual({
|
||||
inFlightTasks: 21,
|
||||
activeAgents: 8,
|
||||
lastActivityAt: "2026-04-08T03:00:00.000Z",
|
||||
});
|
||||
});
|
||||
|
||||
it("logs warning when GET_METRICS IPC query fails", async () => {
|
||||
const warnSpy = vi.spyOn(runtimeLog, "warn").mockImplementation(() => {});
|
||||
|
||||
queueChild({
|
||||
sendCallbackErrors: {
|
||||
[GET_METRICS]: new Error("metrics unavailable"),
|
||||
},
|
||||
});
|
||||
await runtime.start();
|
||||
|
||||
runtimeAny.lastMetrics = {
|
||||
inFlightTasks: 1,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: "2026-04-08T04:00:00.000Z",
|
||||
};
|
||||
|
||||
const metrics = runtime.getMetrics();
|
||||
expect(metrics.inFlightTasks).toBe(1);
|
||||
expect(metrics.activeAgents).toBe(0);
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(warnSpy).toHaveBeenCalledWith(
|
||||
expect.stringContaining("GET_METRICS IPC query failed, using cached value"),
|
||||
);
|
||||
});
|
||||
expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("metrics unavailable"));
|
||||
|
||||
warnSpy.mockRestore();
|
||||
});
|
||||
|
||||
it("getTaskStore() always throws not accessible error", () => {
|
||||
expect(() => runtime.getTaskStore()).toThrow("not accessible in ChildProcessRuntime");
|
||||
});
|
||||
|
||||
it("getScheduler() always throws not accessible error", () => {
|
||||
expect(() => runtime.getScheduler()).toThrow("not accessible in ChildProcessRuntime");
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,390 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import type { RuntimeMetrics, RuntimeStatus, ProjectRuntimeConfig } from "../../project-runtime.js";
|
||||
import {
|
||||
START_RUNTIME,
|
||||
STOP_RUNTIME,
|
||||
GET_STATUS,
|
||||
GET_METRICS,
|
||||
ERROR_EVENT,
|
||||
} from "../../ipc/ipc-protocol.js";
|
||||
|
||||
const mockState = vi.hoisted(() => ({
|
||||
ipcWorkers: [] as any[],
|
||||
runtimes: [] as any[],
|
||||
}));
|
||||
|
||||
vi.mock("../../logger.js", () => {
|
||||
const mockLogger = { log: vi.fn(), warn: vi.fn(), error: vi.fn() };
|
||||
return {
|
||||
runtimeLog: mockLogger,
|
||||
createLogger: () => mockLogger,
|
||||
schedulerLog: mockLogger,
|
||||
triageLog: mockLogger,
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("@fusion/core", () => ({
|
||||
CentralCore: class MockCentralCore {},
|
||||
}));
|
||||
|
||||
vi.mock("../../ipc/ipc-worker.js", () => {
|
||||
class MockIpcWorker {
|
||||
handlers = new Map<string, (payload: unknown) => Promise<unknown> | unknown>();
|
||||
onCommand = vi.fn((type: string, handler: (payload: unknown) => Promise<unknown> | unknown) => {
|
||||
this.handlers.set(type, handler);
|
||||
});
|
||||
sendEvent = vi.fn();
|
||||
shutdown = vi.fn();
|
||||
|
||||
constructor() {
|
||||
mockState.ipcWorkers.push(this);
|
||||
}
|
||||
}
|
||||
|
||||
return { IpcWorker: MockIpcWorker };
|
||||
});
|
||||
|
||||
vi.mock("../in-process-runtime.js", () => {
|
||||
class MockInProcessRuntime {
|
||||
status: RuntimeStatus = "stopped";
|
||||
metrics: RuntimeMetrics = {
|
||||
inFlightTasks: 1,
|
||||
activeAgents: 1,
|
||||
lastActivityAt: "2026-04-08T00:00:00.000Z",
|
||||
};
|
||||
listeners = new Map<string, Array<(...args: any[]) => void>>();
|
||||
|
||||
start = vi.fn(async () => {
|
||||
this.status = "active";
|
||||
});
|
||||
|
||||
stop = vi.fn(async () => {
|
||||
this.status = "stopped";
|
||||
});
|
||||
|
||||
getStatus = vi.fn(() => this.status);
|
||||
|
||||
getMetrics = vi.fn(() => this.metrics);
|
||||
|
||||
on = vi.fn((event: string, handler: (...args: any[]) => void) => {
|
||||
const existing = this.listeners.get(event) ?? [];
|
||||
existing.push(handler);
|
||||
this.listeners.set(event, existing);
|
||||
return this;
|
||||
});
|
||||
|
||||
emit(event: string, ...args: any[]) {
|
||||
for (const handler of this.listeners.get(event) ?? []) {
|
||||
handler(...args);
|
||||
}
|
||||
}
|
||||
|
||||
constructor(
|
||||
public config: ProjectRuntimeConfig,
|
||||
public centralCore: unknown
|
||||
) {
|
||||
mockState.runtimes.push(this);
|
||||
}
|
||||
}
|
||||
|
||||
return { InProcessRuntime: MockInProcessRuntime };
|
||||
});
|
||||
|
||||
vi.mock("../../project-engine.js", async () => {
|
||||
const { InProcessRuntime } = await import("../in-process-runtime.js");
|
||||
class MockProjectEngine {
|
||||
private runtime: any;
|
||||
constructor(config: any, centralCore: any, _options?: any) {
|
||||
this.runtime = new InProcessRuntime(config, centralCore);
|
||||
}
|
||||
start = vi.fn(async () => { await this.runtime.start(); });
|
||||
stop = vi.fn(async () => { await this.runtime.stop(); });
|
||||
getRuntime = vi.fn(() => this.runtime);
|
||||
getTaskStore = vi.fn(() => null);
|
||||
}
|
||||
return { ProjectEngine: MockProjectEngine };
|
||||
});
|
||||
|
||||
type MockWorker = {
|
||||
handlers: Map<string, (payload: unknown) => Promise<unknown> | unknown>;
|
||||
onCommand: ReturnType<typeof vi.fn>;
|
||||
sendEvent: ReturnType<typeof vi.fn>;
|
||||
shutdown: ReturnType<typeof vi.fn>;
|
||||
};
|
||||
|
||||
type MockRuntime = {
|
||||
config: ProjectRuntimeConfig;
|
||||
centralCore: {
|
||||
getGlobalConcurrencyState?: () => Promise<unknown>;
|
||||
recordTaskCompletion?: () => Promise<void>;
|
||||
};
|
||||
status: RuntimeStatus;
|
||||
metrics: RuntimeMetrics;
|
||||
start: ReturnType<typeof vi.fn>;
|
||||
stop: ReturnType<typeof vi.fn>;
|
||||
getStatus: ReturnType<typeof vi.fn>;
|
||||
getMetrics: ReturnType<typeof vi.fn>;
|
||||
on: ReturnType<typeof vi.fn>;
|
||||
emit: (event: string, ...args: unknown[]) => void;
|
||||
};
|
||||
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_worker_test",
|
||||
workingDirectory: "/tmp/test-worker",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
async function loadWorkerModule(): Promise<MockWorker> {
|
||||
await import("../child-process-worker.js");
|
||||
|
||||
const ipcWorker = mockState.ipcWorkers.at(-1) as MockWorker | undefined;
|
||||
if (!ipcWorker) {
|
||||
throw new Error("Expected child-process-worker to instantiate IpcWorker");
|
||||
}
|
||||
|
||||
return ipcWorker;
|
||||
}
|
||||
|
||||
function getHandler<T = unknown>(
|
||||
worker: MockWorker,
|
||||
type: string
|
||||
): (payload: unknown) => Promise<T> {
|
||||
const handler = worker.handlers.get(type);
|
||||
if (!handler) {
|
||||
throw new Error(`Missing handler for ${type}`);
|
||||
}
|
||||
return handler as (payload: unknown) => Promise<T>;
|
||||
}
|
||||
|
||||
describe("child-process-worker", () => {
|
||||
type SignalListener = (...args: unknown[]) => void;
|
||||
const originalProcessSend = process.send;
|
||||
let sigtermBaseline: SignalListener[] = [];
|
||||
let sigintBaseline: SignalListener[] = [];
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetModules();
|
||||
vi.clearAllMocks();
|
||||
|
||||
mockState.ipcWorkers.length = 0;
|
||||
mockState.runtimes.length = 0;
|
||||
|
||||
sigtermBaseline = process.listeners("SIGTERM") as unknown as SignalListener[];
|
||||
sigintBaseline = process.listeners("SIGINT") as unknown as SignalListener[];
|
||||
|
||||
(process as NodeJS.Process & { send?: (...args: unknown[]) => unknown }).send = vi.fn(() => true);
|
||||
vi.spyOn(process, "exit").mockImplementation((() => undefined) as never);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
for (const listener of process.listeners("SIGTERM")) {
|
||||
if (!sigtermBaseline.some((l) => l === listener)) {
|
||||
process.removeListener("SIGTERM", listener as unknown as SignalListener);
|
||||
}
|
||||
}
|
||||
|
||||
for (const listener of process.listeners("SIGINT")) {
|
||||
if (!sigintBaseline.some((l) => l === listener)) {
|
||||
process.removeListener("SIGINT", listener as unknown as SignalListener);
|
||||
}
|
||||
}
|
||||
|
||||
if (originalProcessSend) {
|
||||
process.send = originalProcessSend;
|
||||
} else {
|
||||
delete (process as NodeJS.Process & { send?: unknown }).send;
|
||||
}
|
||||
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("instantiates IpcWorker and registers START/STOP/GET_STATUS/GET_METRICS handlers", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
|
||||
expect(mockState.ipcWorkers).toHaveLength(1);
|
||||
expect(worker.onCommand).toHaveBeenCalledTimes(4);
|
||||
expect(worker.onCommand).toHaveBeenCalledWith(START_RUNTIME, expect.any(Function));
|
||||
expect(worker.onCommand).toHaveBeenCalledWith(STOP_RUNTIME, expect.any(Function));
|
||||
expect(worker.onCommand).toHaveBeenCalledWith(GET_STATUS, expect.any(Function));
|
||||
expect(worker.onCommand).toHaveBeenCalledWith(GET_METRICS, expect.any(Function));
|
||||
expect(worker.handlers.size).toBe(4);
|
||||
});
|
||||
|
||||
it("START_RUNTIME creates and starts InProcessRuntime, then returns status", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const startHandler = getHandler<{ status: RuntimeStatus }>(worker, START_RUNTIME);
|
||||
|
||||
const result = await startHandler({ config: testConfig });
|
||||
|
||||
expect(result).toEqual({ status: "active" });
|
||||
expect(mockState.runtimes).toHaveLength(1);
|
||||
|
||||
const runtime = mockState.runtimes[0] as MockRuntime;
|
||||
expect(runtime.config).toEqual(testConfig);
|
||||
expect(runtime.start).toHaveBeenCalledTimes(1);
|
||||
expect(runtime.getStatus).toHaveBeenCalled();
|
||||
expect(typeof runtime.centralCore.getGlobalConcurrencyState).toBe("function");
|
||||
expect(typeof runtime.centralCore.recordTaskCompletion).toBe("function");
|
||||
});
|
||||
|
||||
it("START_RUNTIME throws if runtime is already started", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const startHandler = getHandler(worker, START_RUNTIME);
|
||||
|
||||
await startHandler({ config: testConfig });
|
||||
await expect(startHandler({ config: testConfig })).rejects.toThrow("Runtime already started");
|
||||
});
|
||||
|
||||
it("START_RUNTIME forwards runtime events via ipcWorker.sendEvent", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const startHandler = getHandler(worker, START_RUNTIME);
|
||||
|
||||
await startHandler({ config: testConfig });
|
||||
const runtime = mockState.runtimes[0] as MockRuntime;
|
||||
|
||||
const task = {
|
||||
id: "FN-1279",
|
||||
title: "task",
|
||||
description: "desc",
|
||||
column: "todo",
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
createdAt: "2026-04-08T00:00:00.000Z",
|
||||
updatedAt: "2026-04-08T00:00:00.000Z",
|
||||
size: "M",
|
||||
reviewLevel: 1,
|
||||
log: [],
|
||||
attachments: [],
|
||||
};
|
||||
|
||||
runtime.emit("task:created", task);
|
||||
runtime.emit("task:moved", { task, from: "todo", to: "in-progress" });
|
||||
runtime.emit("task:updated", task);
|
||||
const runtimeError = new Error("runtime boom") as Error & { code?: string };
|
||||
runtimeError.code = "RUNTIME_ERR";
|
||||
runtime.emit("error", runtimeError);
|
||||
runtime.emit("health-changed", { status: "active", previous: "starting" });
|
||||
|
||||
expect(worker.sendEvent).toHaveBeenCalledWith("TASK_CREATED", { task });
|
||||
expect(worker.sendEvent).toHaveBeenCalledWith("TASK_MOVED", {
|
||||
task,
|
||||
from: "todo",
|
||||
to: "in-progress",
|
||||
});
|
||||
expect(worker.sendEvent).toHaveBeenCalledWith("TASK_UPDATED", { task });
|
||||
expect(worker.sendEvent).toHaveBeenCalledWith(ERROR_EVENT, {
|
||||
message: "runtime boom",
|
||||
code: "RUNTIME_ERR",
|
||||
});
|
||||
expect(worker.sendEvent).toHaveBeenCalledWith("HEALTH_CHANGED", {
|
||||
status: "active",
|
||||
previous: "starting",
|
||||
});
|
||||
});
|
||||
|
||||
it("STOP_RUNTIME stops existing runtime and returns stopped true", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const startHandler = getHandler(worker, START_RUNTIME);
|
||||
const stopHandler = getHandler<{ stopped: boolean }>(worker, STOP_RUNTIME);
|
||||
|
||||
await startHandler({ config: testConfig });
|
||||
const runtime = mockState.runtimes[0] as MockRuntime;
|
||||
|
||||
const result = await stopHandler({ timeoutMs: 12345 });
|
||||
|
||||
expect(result).toEqual({ stopped: true });
|
||||
expect(runtime.stop).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("STOP_RUNTIME throws when runtime has not been started", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const stopHandler = getHandler(worker, STOP_RUNTIME);
|
||||
|
||||
await expect(stopHandler({ timeoutMs: 30000 })).rejects.toThrow("Runtime not started");
|
||||
});
|
||||
|
||||
it("GET_STATUS returns stopped when runtime is null", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const getStatusHandler = getHandler<{ status: RuntimeStatus }>(worker, GET_STATUS);
|
||||
|
||||
await expect(getStatusHandler({})).resolves.toEqual({ status: "stopped" });
|
||||
});
|
||||
|
||||
it("GET_STATUS returns runtime status when runtime exists", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const startHandler = getHandler(worker, START_RUNTIME);
|
||||
const getStatusHandler = getHandler<{ status: RuntimeStatus }>(worker, GET_STATUS);
|
||||
|
||||
await startHandler({ config: testConfig });
|
||||
|
||||
const runtime = mockState.runtimes[0] as MockRuntime;
|
||||
runtime.status = "paused";
|
||||
|
||||
await expect(getStatusHandler({})).resolves.toEqual({ status: "paused" });
|
||||
});
|
||||
|
||||
it("GET_METRICS returns default metrics when runtime is null", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const getMetricsHandler = getHandler<RuntimeMetrics>(worker, GET_METRICS);
|
||||
|
||||
const result = await getMetricsHandler({});
|
||||
|
||||
expect(result.inFlightTasks).toBe(0);
|
||||
expect(result.activeAgents).toBe(0);
|
||||
expect(typeof result.lastActivityAt).toBe("string");
|
||||
});
|
||||
|
||||
it("GET_METRICS returns runtime metrics when runtime exists", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const startHandler = getHandler(worker, START_RUNTIME);
|
||||
const getMetricsHandler = getHandler<RuntimeMetrics>(worker, GET_METRICS);
|
||||
|
||||
await startHandler({ config: testConfig });
|
||||
const runtime = mockState.runtimes[0] as MockRuntime;
|
||||
runtime.metrics = {
|
||||
inFlightTasks: 7,
|
||||
activeAgents: 4,
|
||||
lastActivityAt: "2026-04-08T05:00:00.000Z",
|
||||
};
|
||||
|
||||
await expect(getMetricsHandler({})).resolves.toEqual(runtime.metrics);
|
||||
expect(runtime.getMetrics).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("SIGTERM stops runtime and shuts down IPC worker", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const startHandler = getHandler(worker, START_RUNTIME);
|
||||
|
||||
await startHandler({ config: testConfig });
|
||||
const runtime = mockState.runtimes[0] as MockRuntime;
|
||||
|
||||
process.emit("SIGTERM");
|
||||
await vi.waitFor(() => {
|
||||
expect(runtime.stop).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(worker.shutdown).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
it("SIGINT stops runtime and shuts down IPC worker", async () => {
|
||||
const worker = await loadWorkerModule();
|
||||
const startHandler = getHandler(worker, START_RUNTIME);
|
||||
|
||||
await startHandler({ config: testConfig });
|
||||
const runtime = mockState.runtimes[0] as MockRuntime;
|
||||
|
||||
process.emit("SIGINT");
|
||||
await vi.waitFor(() => {
|
||||
expect(runtime.stop).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(worker.shutdown).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
});
|
||||
1163
packages/engine/src/runtimes/__tests__/in-process-runtime.test.ts
Normal file
1163
packages/engine/src/runtimes/__tests__/in-process-runtime.test.ts
Normal file
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,334 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { RuntimeMetrics } from "../../project-runtime.js";
|
||||
import { RemoteNodeClient } from "../remote-node-client.js";
|
||||
|
||||
const BASE_URL = "https://node.example.com";
|
||||
const API_KEY = "secret-token";
|
||||
|
||||
describe("RemoteNodeClient", () => {
|
||||
const originalFetch = globalThis.fetch;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
globalThis.fetch = originalFetch;
|
||||
vi.restoreAllMocks();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("health() parses successful response and sends auth header", async () => {
|
||||
const fetchMock = vi.fn().mockResolvedValue(
|
||||
new Response(JSON.stringify({ status: "ok", version: "1.0.0", uptime: 123 }), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
);
|
||||
globalThis.fetch = fetchMock as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
|
||||
const health = await client.health();
|
||||
|
||||
expect(health).toEqual({ status: "ok", version: "1.0.0", uptime: 123 });
|
||||
expect(fetchMock).toHaveBeenCalledWith(`${BASE_URL}/api/health`, expect.objectContaining({
|
||||
method: "GET",
|
||||
headers: expect.objectContaining({
|
||||
Authorization: `Bearer ${API_KEY}`,
|
||||
}),
|
||||
}));
|
||||
});
|
||||
|
||||
it("getMetrics() parses runtime metrics", async () => {
|
||||
const metrics: RuntimeMetrics = {
|
||||
inFlightTasks: 4,
|
||||
activeAgents: 2,
|
||||
lastActivityAt: "2026-04-08T00:00:00.000Z",
|
||||
};
|
||||
|
||||
globalThis.fetch = vi.fn().mockResolvedValue(
|
||||
new Response(JSON.stringify(metrics), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
) as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
|
||||
await expect(client.getMetrics()).resolves.toEqual(metrics);
|
||||
});
|
||||
|
||||
it("createTask() sends POST with JSON body", async () => {
|
||||
const createdTask = {
|
||||
id: "KB-001",
|
||||
description: "Create me",
|
||||
column: "todo",
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
status: "pending",
|
||||
log: [],
|
||||
attachments: [],
|
||||
createdAt: "2026-04-08T00:00:00.000Z",
|
||||
updatedAt: "2026-04-08T00:00:00.000Z",
|
||||
size: "M",
|
||||
reviewLevel: 1,
|
||||
};
|
||||
|
||||
const fetchMock = vi.fn().mockResolvedValue(
|
||||
new Response(JSON.stringify(createdTask), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
);
|
||||
globalThis.fetch = fetchMock as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
|
||||
await client.createTask({ description: "Create me" });
|
||||
|
||||
const options = fetchMock.mock.calls[0]?.[1] as RequestInit;
|
||||
expect(fetchMock).toHaveBeenCalledWith(`${BASE_URL}/api/tasks`, expect.any(Object));
|
||||
expect(options.method).toBe("POST");
|
||||
expect(options.headers).toEqual(expect.objectContaining({
|
||||
"Content-Type": "application/json",
|
||||
Authorization: `Bearer ${API_KEY}`,
|
||||
}));
|
||||
expect(options.body).toBe(JSON.stringify({ description: "Create me" }));
|
||||
});
|
||||
|
||||
it("listTasks() sends optional query params", async () => {
|
||||
const fetchMock = vi.fn().mockResolvedValue(
|
||||
new Response(JSON.stringify([]), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
);
|
||||
globalThis.fetch = fetchMock as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
|
||||
await client.listTasks({ column: "in-progress", limit: 10 });
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledWith(
|
||||
`${BASE_URL}/api/tasks?column=in-progress&limit=10`,
|
||||
expect.objectContaining({ method: "GET" })
|
||||
);
|
||||
});
|
||||
|
||||
it("executeTask() posts to execute endpoint", async () => {
|
||||
const fetchMock = vi.fn().mockResolvedValue(
|
||||
new Response(JSON.stringify({ acknowledged: true }), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
);
|
||||
globalThis.fetch = fetchMock as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
|
||||
const result = await client.executeTask("KB-123");
|
||||
|
||||
expect(result).toEqual({ acknowledged: true });
|
||||
expect(fetchMock).toHaveBeenCalledWith(
|
||||
`${BASE_URL}/api/tasks/KB-123/execute`,
|
||||
expect.objectContaining({ method: "POST" })
|
||||
);
|
||||
});
|
||||
|
||||
it("streamEvents() yields parsed events from SSE stream", async () => {
|
||||
const sseBody = [
|
||||
"event: task:created",
|
||||
'data: {"type":"task:created","payload":{"id":"KB-1"},"timestamp":"2026-04-08T00:00:00.000Z"}',
|
||||
"",
|
||||
"event: task:updated",
|
||||
'data: {"type":"task:updated","payload":{"id":"KB-1","column":"in-progress"},"timestamp":"2026-04-08T00:01:00.000Z"}',
|
||||
"",
|
||||
].join("\n");
|
||||
|
||||
globalThis.fetch = vi.fn().mockResolvedValue(
|
||||
new Response(sseBody, {
|
||||
status: 200,
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
})
|
||||
) as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
|
||||
|
||||
const events: unknown[] = [];
|
||||
for await (const event of client.streamEvents()) {
|
||||
events.push(event);
|
||||
}
|
||||
|
||||
expect(events).toEqual([
|
||||
{
|
||||
type: "task:created",
|
||||
payload: { id: "KB-1" },
|
||||
timestamp: "2026-04-08T00:00:00.000Z",
|
||||
},
|
||||
{
|
||||
type: "task:updated",
|
||||
payload: { id: "KB-1", column: "in-progress" },
|
||||
timestamp: "2026-04-08T00:01:00.000Z",
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it("retries on network errors", async () => {
|
||||
vi.useFakeTimers();
|
||||
|
||||
const fetchMock = vi
|
||||
.fn()
|
||||
.mockRejectedValueOnce(new TypeError("network down"))
|
||||
.mockResolvedValueOnce(
|
||||
new Response(JSON.stringify({ status: "ok", version: "1.0.0", uptime: 123 }), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
);
|
||||
globalThis.fetch = fetchMock as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
|
||||
|
||||
const request = client.health();
|
||||
const expectation = expect(request).resolves.toEqual({
|
||||
status: "ok",
|
||||
version: "1.0.0",
|
||||
uptime: 123,
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await expectation;
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("does not retry on 4xx responses", async () => {
|
||||
const fetchMock = vi.fn().mockResolvedValue(
|
||||
new Response(JSON.stringify({ error: "unauthorized" }), {
|
||||
status: 401,
|
||||
statusText: "Unauthorized",
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
);
|
||||
globalThis.fetch = fetchMock as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
|
||||
|
||||
await expect(client.health()).rejects.toThrow("401 Unauthorized");
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("retries on 5xx responses", async () => {
|
||||
vi.useFakeTimers();
|
||||
|
||||
const fetchMock = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce(
|
||||
new Response("server error", { status: 500, statusText: "Internal Server Error" })
|
||||
)
|
||||
.mockResolvedValueOnce(
|
||||
new Response("server error", { status: 502, statusText: "Bad Gateway" })
|
||||
)
|
||||
.mockResolvedValueOnce(
|
||||
new Response(JSON.stringify({ status: "ok", version: "1.0.0", uptime: 999 }), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
);
|
||||
globalThis.fetch = fetchMock as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
|
||||
|
||||
const request = client.health();
|
||||
const expectation = expect(request).resolves.toEqual({
|
||||
status: "ok",
|
||||
version: "1.0.0",
|
||||
uptime: 999,
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await vi.advanceTimersByTimeAsync(2000);
|
||||
await expectation;
|
||||
expect(fetchMock).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it("aborts requests after timeoutMs", async () => {
|
||||
vi.useFakeTimers();
|
||||
|
||||
const fetchMock = vi.fn().mockImplementation((_: unknown, init?: RequestInit) => {
|
||||
return new Promise((_resolve, reject) => {
|
||||
const signal = init?.signal;
|
||||
signal?.addEventListener("abort", () => {
|
||||
const abortError = new Error("aborted");
|
||||
abortError.name = "AbortError";
|
||||
reject(abortError);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
globalThis.fetch = fetchMock as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({
|
||||
baseUrl: BASE_URL,
|
||||
apiKey: API_KEY,
|
||||
timeoutMs: 5,
|
||||
});
|
||||
|
||||
const request = client.health();
|
||||
const expectation = expect(request).rejects.toThrow("timed out");
|
||||
|
||||
await vi.runAllTimersAsync();
|
||||
await expectation;
|
||||
expect(fetchMock).toHaveBeenCalledTimes(4); // initial + 3 retries
|
||||
});
|
||||
|
||||
it("sends auth header on all request methods", async () => {
|
||||
const fetchMock = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce(
|
||||
new Response(JSON.stringify({ status: "ok", version: "1.0.0", uptime: 1 }), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
)
|
||||
.mockResolvedValueOnce(
|
||||
new Response(JSON.stringify({ inFlightTasks: 0, activeAgents: 0, lastActivityAt: "now" }), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
)
|
||||
.mockResolvedValueOnce(
|
||||
new Response(JSON.stringify([]), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
)
|
||||
.mockResolvedValueOnce(
|
||||
new Response(JSON.stringify({ acknowledged: true }), {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
})
|
||||
)
|
||||
.mockResolvedValueOnce(
|
||||
new Response("event: ping\ndata: {}\n\n", {
|
||||
status: 200,
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
})
|
||||
);
|
||||
|
||||
globalThis.fetch = fetchMock as unknown as typeof fetch;
|
||||
|
||||
const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
|
||||
|
||||
await client.health();
|
||||
await client.getMetrics();
|
||||
await client.listTasks();
|
||||
await client.executeTask("KB-777");
|
||||
for await (const _event of client.streamEvents()) {
|
||||
// Drain one-response event stream
|
||||
}
|
||||
|
||||
for (const call of fetchMock.mock.calls) {
|
||||
const options = call[1] as RequestInit;
|
||||
expect(options.headers).toEqual(
|
||||
expect.objectContaining({
|
||||
Authorization: `Bearer ${API_KEY}`,
|
||||
})
|
||||
);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,268 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { NodeConfig } from "@fusion/core";
|
||||
import type { RuntimeMetrics } from "../../project-runtime.js";
|
||||
import { RemoteNodeRuntime } from "../remote-node-runtime.js";
|
||||
|
||||
const mockClientConstructor = vi.hoisted(() => vi.fn());
|
||||
const mockHealth = vi.hoisted(() => vi.fn());
|
||||
const mockGetMetrics = vi.hoisted(() => vi.fn());
|
||||
const mockStreamEvents = vi.hoisted(() => vi.fn());
|
||||
|
||||
vi.mock("../remote-node-client.js", () => ({
|
||||
RemoteNodeClient: vi.fn().mockImplementation((options: unknown) => {
|
||||
mockClientConstructor(options);
|
||||
return {
|
||||
health: mockHealth,
|
||||
getMetrics: mockGetMetrics,
|
||||
streamEvents: mockStreamEvents,
|
||||
};
|
||||
}),
|
||||
}));
|
||||
|
||||
const NOW = "2026-04-08T00:00:00.000Z";
|
||||
|
||||
function createNode(overrides?: Partial<NodeConfig>): NodeConfig {
|
||||
return {
|
||||
id: "node_remote_1",
|
||||
name: "Remote Node",
|
||||
type: "remote",
|
||||
url: "https://remote.example.com",
|
||||
apiKey: "token-123",
|
||||
status: "online",
|
||||
maxConcurrent: 4,
|
||||
createdAt: NOW,
|
||||
updatedAt: NOW,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
async function* idleStream(signal?: AbortSignal): AsyncIterable<unknown> {
|
||||
while (!signal?.aborted) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
}
|
||||
// Yield to satisfy TypeScript/ESLint generator requirements
|
||||
yield;
|
||||
}
|
||||
|
||||
async function* eventStream(events: unknown[], signal?: AbortSignal): AsyncIterable<unknown> {
|
||||
for (const event of events) {
|
||||
yield event;
|
||||
}
|
||||
|
||||
while (!signal?.aborted) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
}
|
||||
}
|
||||
|
||||
describe("RemoteNodeRuntime", () => {
|
||||
beforeEach(() => {
|
||||
mockClientConstructor.mockReset();
|
||||
mockHealth.mockReset();
|
||||
mockGetMetrics.mockReset();
|
||||
mockStreamEvents.mockReset();
|
||||
|
||||
mockHealth.mockResolvedValue({ status: "ok", version: "1.0.0", uptime: 100 });
|
||||
mockGetMetrics.mockResolvedValue({
|
||||
inFlightTasks: 1,
|
||||
activeAgents: 2,
|
||||
lastActivityAt: NOW,
|
||||
} satisfies RuntimeMetrics);
|
||||
mockStreamEvents.mockImplementation(({ signal }: { signal?: AbortSignal } = {}) =>
|
||||
idleStream(signal)
|
||||
);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
vi.clearAllMocks();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("start() transitions stopped -> starting -> active and starts stream", async () => {
|
||||
const runtime = new RemoteNodeRuntime({
|
||||
nodeConfig: createNode(),
|
||||
projectId: "proj_1",
|
||||
projectName: "Project 1",
|
||||
});
|
||||
|
||||
const healthEvents: string[] = [];
|
||||
runtime.on("health-changed", ({ status }) => {
|
||||
healthEvents.push(status);
|
||||
});
|
||||
|
||||
await runtime.start();
|
||||
|
||||
expect(runtime.getStatus()).toBe("active");
|
||||
expect(healthEvents).toEqual(["starting", "active"]);
|
||||
expect(mockHealth).toHaveBeenCalled();
|
||||
expect(mockStreamEvents).toHaveBeenCalled();
|
||||
expect(mockClientConstructor).toHaveBeenCalledWith({
|
||||
baseUrl: "https://remote.example.com",
|
||||
apiKey: "token-123",
|
||||
});
|
||||
|
||||
await runtime.stop();
|
||||
});
|
||||
|
||||
it("stop() transitions to stopped and is idempotent", async () => {
|
||||
const runtime = new RemoteNodeRuntime({
|
||||
nodeConfig: createNode(),
|
||||
projectId: "proj_2",
|
||||
projectName: "Project 2",
|
||||
});
|
||||
|
||||
await runtime.start();
|
||||
await runtime.stop();
|
||||
|
||||
expect(runtime.getStatus()).toBe("stopped");
|
||||
|
||||
await expect(runtime.stop()).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it("getTaskStore() throws descriptive error", () => {
|
||||
const runtime = new RemoteNodeRuntime({
|
||||
nodeConfig: createNode(),
|
||||
projectId: "proj_3",
|
||||
projectName: "Project 3",
|
||||
});
|
||||
|
||||
expect(() => runtime.getTaskStore()).toThrow(
|
||||
"TaskStore not accessible for remote node runtime"
|
||||
);
|
||||
});
|
||||
|
||||
it("getScheduler() throws descriptive error", () => {
|
||||
const runtime = new RemoteNodeRuntime({
|
||||
nodeConfig: createNode(),
|
||||
projectId: "proj_4",
|
||||
projectName: "Project 4",
|
||||
});
|
||||
|
||||
expect(() => runtime.getScheduler()).toThrow("Scheduler not accessible for remote node runtime");
|
||||
});
|
||||
|
||||
it("getMetrics() returns fetched metrics on success and fallback on failure", async () => {
|
||||
const runtime = new RemoteNodeRuntime({
|
||||
nodeConfig: createNode(),
|
||||
projectId: "proj_5",
|
||||
projectName: "Project 5",
|
||||
});
|
||||
|
||||
await runtime.start();
|
||||
|
||||
expect(runtime.getMetrics()).toEqual({
|
||||
inFlightTasks: 1,
|
||||
activeAgents: 2,
|
||||
lastActivityAt: NOW,
|
||||
});
|
||||
|
||||
mockGetMetrics.mockRejectedValueOnce(new Error("metrics unavailable"));
|
||||
|
||||
runtime.getMetrics();
|
||||
await Promise.resolve();
|
||||
|
||||
expect(runtime.getMetrics()).toEqual({
|
||||
inFlightTasks: 0,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: NOW,
|
||||
});
|
||||
|
||||
await runtime.stop();
|
||||
});
|
||||
|
||||
it("forwards remote task and error events", async () => {
|
||||
const createdHandler = vi.fn();
|
||||
const movedHandler = vi.fn();
|
||||
const updatedHandler = vi.fn();
|
||||
const errorHandler = vi.fn();
|
||||
|
||||
mockStreamEvents.mockImplementation(({ signal }: { signal?: AbortSignal } = {}) =>
|
||||
eventStream(
|
||||
[
|
||||
{
|
||||
type: "task:created",
|
||||
payload: { id: "KB-1" },
|
||||
timestamp: NOW,
|
||||
},
|
||||
{
|
||||
type: "task:moved",
|
||||
payload: { task: { id: "KB-1" }, from: "todo", to: "in-progress" },
|
||||
timestamp: NOW,
|
||||
},
|
||||
{
|
||||
type: "task:updated",
|
||||
payload: { id: "KB-1", column: "done" },
|
||||
timestamp: NOW,
|
||||
},
|
||||
{
|
||||
type: "error",
|
||||
payload: { message: "boom" },
|
||||
timestamp: NOW,
|
||||
},
|
||||
],
|
||||
signal
|
||||
)
|
||||
);
|
||||
|
||||
const runtime = new RemoteNodeRuntime({
|
||||
nodeConfig: createNode(),
|
||||
projectId: "proj_6",
|
||||
projectName: "Project 6",
|
||||
});
|
||||
|
||||
runtime.on("task:created", createdHandler);
|
||||
runtime.on("task:moved", movedHandler);
|
||||
runtime.on("task:updated", updatedHandler);
|
||||
runtime.on("error", errorHandler);
|
||||
|
||||
await runtime.start();
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(createdHandler).toHaveBeenCalledWith({ id: "KB-1" });
|
||||
expect(movedHandler).toHaveBeenCalledWith({
|
||||
task: { id: "KB-1" },
|
||||
from: "todo",
|
||||
to: "in-progress",
|
||||
});
|
||||
expect(updatedHandler).toHaveBeenCalledWith({ id: "KB-1", column: "done" });
|
||||
expect(errorHandler).toHaveBeenCalledWith(expect.any(Error));
|
||||
});
|
||||
|
||||
await runtime.stop();
|
||||
});
|
||||
|
||||
it("reconnects when stream ends unexpectedly and transitions to errored after max attempts", async () => {
|
||||
mockStreamEvents.mockImplementation(async function* () {
|
||||
// Immediate end to force reconnect loop.
|
||||
});
|
||||
|
||||
const runtime = new RemoteNodeRuntime({
|
||||
nodeConfig: createNode(),
|
||||
projectId: "proj_7",
|
||||
projectName: "Project 7",
|
||||
});
|
||||
|
||||
(runtime as unknown as { reconnectBaseDelayMs: number }).reconnectBaseDelayMs = 1;
|
||||
(runtime as unknown as { maxReconnectDelayMs: number }).maxReconnectDelayMs = 1;
|
||||
(runtime as unknown as { maxReconnectAttempts: number }).maxReconnectAttempts = 3;
|
||||
|
||||
await runtime.start();
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(runtime.getStatus()).toBe("errored");
|
||||
});
|
||||
|
||||
expect(mockStreamEvents.mock.calls.length).toBeGreaterThanOrEqual(3);
|
||||
|
||||
await runtime.stop();
|
||||
});
|
||||
|
||||
it("validates remote node config on start", async () => {
|
||||
const runtime = new RemoteNodeRuntime({
|
||||
nodeConfig: createNode({ type: "local", url: undefined, apiKey: undefined }),
|
||||
projectId: "proj_8",
|
||||
projectName: "Project 8",
|
||||
});
|
||||
|
||||
await expect(runtime.start()).rejects.toThrow("requires a remote node configuration");
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user