feat(FN-1079): add remote node runtime orchestration
- Add RemoteNodeClient and RemoteNodeRuntime with comprehensive lifecycle, status, and metrics test coverage - Route projects by node assignment in ProjectManager and integrate remote runtime handling in HybridExecutor - Introduce NodeHealthMonitor and wire diagnostic logging/export updates for remote node health tracking - Harden remote runtime shutdown checks and type-safety paths surfaced during review feedback
This commit is contained in:
gsxdsm
323
packages/engine/src/runtimes/remote-node-client.test.ts
Normal file
323
packages/engine/src/runtimes/remote-node-client.test.ts
Normal file
@@ -0,0 +1,323 @@
|
||||
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 () => {
|
||||
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 });
|
||||
|
||||
await expect(client.health()).resolves.toEqual({
|
||||
status: "ok",
|
||||
version: "1.0.0",
|
||||
uptime: 123,
|
||||
});
|
||||
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 () => {
|
||||
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 });
|
||||
|
||||
await expect(client.health()).resolves.toEqual({
|
||||
status: "ok",
|
||||
version: "1.0.0",
|
||||
uptime: 999,
|
||||
});
|
||||
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}`,
|
||||
})
|
||||
);
|
||||
}
|
||||
});
|
||||
});
|
||||
441
packages/engine/src/runtimes/remote-node-client.ts
Normal file
441
packages/engine/src/runtimes/remote-node-client.ts
Normal file
@@ -0,0 +1,441 @@
|
||||
import type { Task, TaskCreateInput } from "@fusion/core";
|
||||
import type { RuntimeMetrics } from "../project-runtime.js";
|
||||
import { remoteNodeLog } from "../logger.js";
|
||||
|
||||
export interface RemoteNodeEvent {
|
||||
type: string;
|
||||
payload: unknown;
|
||||
timestamp: string;
|
||||
}
|
||||
|
||||
export interface RemoteNodeClientOptions {
|
||||
baseUrl: string;
|
||||
apiKey: string;
|
||||
timeoutMs?: number;
|
||||
}
|
||||
|
||||
export type RemoteTaskListFilter = Record<string, string | number | boolean | undefined | null>;
|
||||
|
||||
class RemoteNodeRequestError extends Error {
|
||||
constructor(
|
||||
message: string,
|
||||
readonly retryable: boolean,
|
||||
readonly status?: number
|
||||
) {
|
||||
super(message);
|
||||
this.name = "RemoteNodeRequestError";
|
||||
}
|
||||
}
|
||||
|
||||
const RETRY_BASE_DELAY_MS = 1000;
|
||||
const DEFAULT_MAX_RETRIES = 3;
|
||||
|
||||
export class RemoteNodeClient {
|
||||
private readonly baseUrl: string;
|
||||
private readonly apiKey: string;
|
||||
private readonly timeoutMs: number;
|
||||
|
||||
constructor(options: RemoteNodeClientOptions) {
|
||||
this.baseUrl = options.baseUrl.replace(/\/+$/, "");
|
||||
this.apiKey = options.apiKey;
|
||||
this.timeoutMs = options.timeoutMs ?? 30_000;
|
||||
}
|
||||
|
||||
async health(): Promise<{ status: string; version: string; uptime: number }> {
|
||||
return this.withRetry(() =>
|
||||
this.requestJson<{ status: string; version: string; uptime: number }>("/api/health", {
|
||||
method: "GET",
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
async getMetrics(): Promise<RuntimeMetrics> {
|
||||
return this.withRetry(() =>
|
||||
this.requestJson<RuntimeMetrics>("/api/metrics", {
|
||||
method: "GET",
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
async createTask(input: TaskCreateInput): Promise<Task> {
|
||||
return this.withRetry(() =>
|
||||
this.requestJson<Task>("/api/tasks", {
|
||||
method: "POST",
|
||||
body: JSON.stringify(input),
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
async listTasks(filter?: RemoteTaskListFilter): Promise<Task[]> {
|
||||
const query = new URLSearchParams();
|
||||
if (filter) {
|
||||
for (const [key, value] of Object.entries(filter)) {
|
||||
if (value !== undefined && value !== null) {
|
||||
query.set(key, String(value));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const path = query.toString().length > 0 ? `/api/tasks?${query.toString()}` : "/api/tasks";
|
||||
return this.withRetry(() =>
|
||||
this.requestJson<Task[]>(path, {
|
||||
method: "GET",
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
async executeTask(taskId: string): Promise<{ acknowledged: boolean; [key: string]: unknown }> {
|
||||
return this.withRetry(() =>
|
||||
this.requestJson<{ acknowledged: boolean; [key: string]: unknown }>(
|
||||
`/api/tasks/${encodeURIComponent(taskId)}/execute`,
|
||||
{
|
||||
method: "POST",
|
||||
}
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
async *streamEvents(options?: { signal?: AbortSignal }): AsyncIterable<RemoteNodeEvent> {
|
||||
const response = await this.withRetry(
|
||||
() => this.openStream("/api/events/stream", options?.signal),
|
||||
DEFAULT_MAX_RETRIES
|
||||
);
|
||||
|
||||
const contentType = response.headers.get("content-type") ?? "";
|
||||
if (!response.body) {
|
||||
throw new Error("Remote node event stream opened without a body");
|
||||
}
|
||||
|
||||
if (contentType.includes("text/event-stream")) {
|
||||
yield* this.parseSseStream(response.body, options?.signal);
|
||||
return;
|
||||
}
|
||||
|
||||
// Fallback for long-polling endpoints that return JSON payloads.
|
||||
if (contentType.includes("application/json")) {
|
||||
const payload = (await response.json()) as unknown;
|
||||
if (Array.isArray(payload)) {
|
||||
for (const rawEvent of payload) {
|
||||
yield this.normalizeEvent(rawEvent, "message");
|
||||
}
|
||||
} else {
|
||||
yield this.normalizeEvent(payload, "message");
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Generic fallback: treat each line as one JSON event.
|
||||
yield* this.parseJsonLines(response.body, options?.signal);
|
||||
}
|
||||
|
||||
private async requestJson<T>(path: string, init: RequestInit): Promise<T> {
|
||||
const response = await this.fetchWithTimeout(path, {
|
||||
...init,
|
||||
headers: {
|
||||
...this.getAuthHeaders(),
|
||||
Accept: "application/json",
|
||||
...(init.body ? { "Content-Type": "application/json" } : {}),
|
||||
...(init.headers ?? {}),
|
||||
},
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
await this.throwHttpError(path, response);
|
||||
}
|
||||
|
||||
try {
|
||||
return (await response.json()) as T;
|
||||
} catch (error) {
|
||||
throw new RemoteNodeRequestError(
|
||||
`Failed to parse JSON response for ${path}: ${error instanceof Error ? error.message : String(error)}`,
|
||||
false
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async openStream(path: string, signal?: AbortSignal): Promise<Response> {
|
||||
const response = await this.fetchWithTimeout(
|
||||
path,
|
||||
{
|
||||
method: "GET",
|
||||
headers: {
|
||||
...this.getAuthHeaders(),
|
||||
Accept: "text/event-stream, application/json",
|
||||
},
|
||||
},
|
||||
signal
|
||||
);
|
||||
|
||||
if (!response.ok) {
|
||||
await this.throwHttpError(path, response);
|
||||
}
|
||||
|
||||
return response;
|
||||
}
|
||||
|
||||
private async throwHttpError(path: string, response: Response): Promise<never> {
|
||||
const responseBody = (await response.text()).trim();
|
||||
const snippet = responseBody.length > 0 ? ` — ${responseBody.slice(0, 300)}` : "";
|
||||
const retryable = response.status >= 500;
|
||||
|
||||
throw new RemoteNodeRequestError(
|
||||
`Remote node request failed (${response.status} ${response.statusText}) for ${path}${snippet}`,
|
||||
retryable,
|
||||
response.status
|
||||
);
|
||||
}
|
||||
|
||||
private getAuthHeaders(): Record<string, string> {
|
||||
return {
|
||||
Authorization: `Bearer ${this.apiKey}`,
|
||||
};
|
||||
}
|
||||
|
||||
private async fetchWithTimeout(
|
||||
path: string,
|
||||
init: RequestInit,
|
||||
externalSignal?: AbortSignal
|
||||
): Promise<Response> {
|
||||
const controller = new AbortController();
|
||||
let timedOut = false;
|
||||
|
||||
const timeout = setTimeout(() => {
|
||||
timedOut = true;
|
||||
controller.abort();
|
||||
}, this.timeoutMs);
|
||||
|
||||
const onAbort = () => controller.abort(externalSignal?.reason);
|
||||
if (externalSignal) {
|
||||
if (externalSignal.aborted) {
|
||||
clearTimeout(timeout);
|
||||
throw new RemoteNodeRequestError("Request aborted", false);
|
||||
}
|
||||
externalSignal.addEventListener("abort", onAbort, { once: true });
|
||||
}
|
||||
|
||||
try {
|
||||
return await fetch(`${this.baseUrl}${path}`, {
|
||||
...init,
|
||||
signal: controller.signal,
|
||||
});
|
||||
} catch (error) {
|
||||
if (error instanceof RemoteNodeRequestError) {
|
||||
throw error;
|
||||
}
|
||||
|
||||
if (timedOut) {
|
||||
throw new RemoteNodeRequestError(
|
||||
`Remote node request timed out after ${this.timeoutMs}ms (${path})`,
|
||||
true
|
||||
);
|
||||
}
|
||||
|
||||
if (error instanceof Error && error.name === "AbortError") {
|
||||
throw new RemoteNodeRequestError(`Remote node request aborted (${path})`, false);
|
||||
}
|
||||
|
||||
throw new RemoteNodeRequestError(
|
||||
`Remote node network error (${path}): ${error instanceof Error ? error.message : String(error)}`,
|
||||
true
|
||||
);
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
if (externalSignal) {
|
||||
externalSignal.removeEventListener("abort", onAbort);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async *parseSseStream(
|
||||
stream: ReadableStream<Uint8Array>,
|
||||
signal?: AbortSignal
|
||||
): AsyncIterable<RemoteNodeEvent> {
|
||||
const reader = stream.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
|
||||
let buffer = "";
|
||||
let eventType = "message";
|
||||
let dataLines: string[] = [];
|
||||
|
||||
const flushEvent = (): RemoteNodeEvent | null => {
|
||||
if (dataLines.length === 0) {
|
||||
eventType = "message";
|
||||
return null;
|
||||
}
|
||||
|
||||
const data = dataLines.join("\n");
|
||||
dataLines = [];
|
||||
|
||||
const normalized = this.normalizeEvent(data, eventType);
|
||||
eventType = "message";
|
||||
return normalized;
|
||||
};
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
if (signal?.aborted) {
|
||||
break;
|
||||
}
|
||||
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
const lines = buffer.split(/\r?\n/);
|
||||
buffer = lines.pop() ?? "";
|
||||
|
||||
for (const line of lines) {
|
||||
if (line.length === 0) {
|
||||
const event = flushEvent();
|
||||
if (event) {
|
||||
yield event;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (line.startsWith(":")) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const separator = line.indexOf(":");
|
||||
const field = separator === -1 ? line : line.slice(0, separator);
|
||||
const valuePart = separator === -1 ? "" : line.slice(separator + 1).trimStart();
|
||||
|
||||
if (field === "event") {
|
||||
eventType = valuePart || "message";
|
||||
} else if (field === "data") {
|
||||
dataLines.push(valuePart);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (buffer.trim().length > 0) {
|
||||
dataLines.push(buffer.trim());
|
||||
}
|
||||
|
||||
const trailingEvent = flushEvent();
|
||||
if (trailingEvent) {
|
||||
yield trailingEvent;
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
}
|
||||
|
||||
private async *parseJsonLines(
|
||||
stream: ReadableStream<Uint8Array>,
|
||||
signal?: AbortSignal
|
||||
): AsyncIterable<RemoteNodeEvent> {
|
||||
const reader = stream.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = "";
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
if (signal?.aborted) {
|
||||
break;
|
||||
}
|
||||
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
const lines = buffer.split(/\r?\n/);
|
||||
buffer = lines.pop() ?? "";
|
||||
|
||||
for (const line of lines) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
yield this.normalizeEvent(trimmed, "message");
|
||||
}
|
||||
}
|
||||
|
||||
const trailing = buffer.trim();
|
||||
if (trailing.length > 0) {
|
||||
yield this.normalizeEvent(trailing, "message");
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
}
|
||||
|
||||
private normalizeEvent(raw: unknown, fallbackType: string): RemoteNodeEvent {
|
||||
const parsed = this.tryParseJson(raw);
|
||||
|
||||
if (
|
||||
parsed &&
|
||||
typeof parsed === "object" &&
|
||||
"type" in parsed &&
|
||||
"timestamp" in parsed
|
||||
) {
|
||||
return {
|
||||
type: String((parsed as { type: unknown }).type),
|
||||
payload: (parsed as { payload?: unknown }).payload,
|
||||
timestamp: String((parsed as { timestamp: unknown }).timestamp),
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
type: fallbackType,
|
||||
payload: parsed,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
private tryParseJson(value: unknown): unknown {
|
||||
if (typeof value !== "string") {
|
||||
return value;
|
||||
}
|
||||
|
||||
try {
|
||||
return JSON.parse(value);
|
||||
} catch {
|
||||
return value;
|
||||
}
|
||||
}
|
||||
|
||||
private async withRetry<T>(fn: () => Promise<T>, maxRetries = DEFAULT_MAX_RETRIES): Promise<T> {
|
||||
let attempt = 0;
|
||||
|
||||
while (true) {
|
||||
try {
|
||||
return await fn();
|
||||
} catch (error) {
|
||||
const isRetryable =
|
||||
error instanceof RemoteNodeRequestError
|
||||
? error.retryable
|
||||
: this.isLikelyNetworkError(error);
|
||||
|
||||
if (!isRetryable || attempt >= maxRetries) {
|
||||
throw error;
|
||||
}
|
||||
|
||||
const delayMs = RETRY_BASE_DELAY_MS * 2 ** attempt;
|
||||
attempt += 1;
|
||||
remoteNodeLog.warn(
|
||||
`Request failed, retrying in ${delayMs}ms (attempt ${attempt}/${maxRetries})`,
|
||||
error
|
||||
);
|
||||
await this.sleep(delayMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private isLikelyNetworkError(error: unknown): boolean {
|
||||
if (!(error instanceof Error)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (error.name === "AbortError") {
|
||||
return true;
|
||||
}
|
||||
|
||||
return error instanceof TypeError;
|
||||
}
|
||||
|
||||
private async sleep(ms: number): Promise<void> {
|
||||
await new Promise<void>((resolve) => {
|
||||
setTimeout(resolve, ms);
|
||||
});
|
||||
}
|
||||
}
|
||||
266
packages/engine/src/runtimes/remote-node-runtime.test.ts
Normal file
266
packages/engine/src/runtimes/remote-node-runtime.test.ts
Normal file
@@ -0,0 +1,266 @@
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
||||
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");
|
||||
});
|
||||
});
|
||||
343
packages/engine/src/runtimes/remote-node-runtime.ts
Normal file
343
packages/engine/src/runtimes/remote-node-runtime.ts
Normal file
@@ -0,0 +1,343 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { NodeConfig, Task, TaskStore } from "@fusion/core";
|
||||
import type { Scheduler } from "../scheduler.js";
|
||||
import type {
|
||||
ProjectRuntime,
|
||||
ProjectRuntimeEvents,
|
||||
RuntimeMetrics,
|
||||
RuntimeStatus,
|
||||
} from "../project-runtime.js";
|
||||
import { remoteNodeLog } from "../logger.js";
|
||||
import { RemoteNodeClient, type RemoteNodeEvent } from "./remote-node-client.js";
|
||||
|
||||
export interface RemoteNodeRuntimeConfig {
|
||||
nodeConfig: NodeConfig;
|
||||
projectId: string;
|
||||
projectName: string;
|
||||
}
|
||||
|
||||
export class RemoteNodeRuntime
|
||||
extends EventEmitter<ProjectRuntimeEvents>
|
||||
implements ProjectRuntime
|
||||
{
|
||||
private status: RuntimeStatus = "stopped";
|
||||
private client: RemoteNodeClient;
|
||||
private healthInterval: ReturnType<typeof setInterval> | null = null;
|
||||
private streamLoopAbortController: AbortController | null = null;
|
||||
private streamLoopPromise: Promise<void> | null = null;
|
||||
private lastSuccessfulMetricsAt: string;
|
||||
private cachedMetrics: RuntimeMetrics;
|
||||
|
||||
// Kept as mutable fields for testability.
|
||||
private reconnectBaseDelayMs = 5_000;
|
||||
private maxReconnectDelayMs = 60_000;
|
||||
private maxReconnectAttempts = 10;
|
||||
|
||||
constructor(private config: RemoteNodeRuntimeConfig) {
|
||||
super();
|
||||
this.setMaxListeners(100);
|
||||
|
||||
this.cachedMetrics = {
|
||||
inFlightTasks: 0,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
};
|
||||
this.lastSuccessfulMetricsAt = this.cachedMetrics.lastActivityAt;
|
||||
|
||||
this.client = new RemoteNodeClient({
|
||||
baseUrl: config.nodeConfig.url ?? "",
|
||||
apiKey: config.nodeConfig.apiKey ?? "",
|
||||
});
|
||||
|
||||
remoteNodeLog.log(
|
||||
`Created RemoteNodeRuntime for project ${config.projectId} on node ${config.nodeConfig.name}`
|
||||
);
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
if (this.status !== "stopped") {
|
||||
throw new Error(`Cannot start runtime: current status is ${this.status}`);
|
||||
}
|
||||
|
||||
this.validateRemoteNodeConfig();
|
||||
|
||||
this.setStatus("starting");
|
||||
try {
|
||||
await this.client.health();
|
||||
await this.refreshMetrics();
|
||||
|
||||
this.setStatus("active");
|
||||
this.startHealthChecks();
|
||||
this.startEventStreamLoop();
|
||||
|
||||
remoteNodeLog.log(
|
||||
`RemoteNodeRuntime started for ${this.config.projectId} (${this.config.projectName})`
|
||||
);
|
||||
} catch (error) {
|
||||
const err = this.toError(error);
|
||||
this.setStatus("errored");
|
||||
this.emitRuntimeError(err);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
if (this.status === "stopped" || this.status === "stopping") {
|
||||
return;
|
||||
}
|
||||
|
||||
this.setStatus("stopping");
|
||||
|
||||
if (this.healthInterval) {
|
||||
clearInterval(this.healthInterval);
|
||||
this.healthInterval = null;
|
||||
}
|
||||
|
||||
if (this.streamLoopAbortController) {
|
||||
this.streamLoopAbortController.abort();
|
||||
}
|
||||
|
||||
if (this.streamLoopPromise) {
|
||||
try {
|
||||
await this.streamLoopPromise;
|
||||
} catch {
|
||||
// Best-effort shutdown. Errors are already emitted via runtime events.
|
||||
}
|
||||
}
|
||||
|
||||
this.streamLoopAbortController = null;
|
||||
this.streamLoopPromise = null;
|
||||
|
||||
this.setStatus("stopped");
|
||||
remoteNodeLog.log(`RemoteNodeRuntime stopped for ${this.config.projectId}`);
|
||||
}
|
||||
|
||||
getStatus(): RuntimeStatus {
|
||||
return this.status;
|
||||
}
|
||||
|
||||
getTaskStore(): TaskStore {
|
||||
throw new Error(
|
||||
"TaskStore not accessible for remote node runtime. Use the remote Fusion API directly."
|
||||
);
|
||||
}
|
||||
|
||||
getScheduler(): Scheduler {
|
||||
throw new Error("Scheduler not accessible for remote node runtime.");
|
||||
}
|
||||
|
||||
getMetrics(): RuntimeMetrics {
|
||||
void this.refreshMetrics();
|
||||
return { ...this.cachedMetrics };
|
||||
}
|
||||
|
||||
private validateRemoteNodeConfig(): void {
|
||||
if (this.config.nodeConfig.type !== "remote") {
|
||||
throw new Error(
|
||||
`RemoteNodeRuntime requires a remote node configuration (received: ${this.config.nodeConfig.type})`
|
||||
);
|
||||
}
|
||||
|
||||
if (!this.config.nodeConfig.url) {
|
||||
throw new Error("Remote node runtime requires nodeConfig.url for remote nodes.");
|
||||
}
|
||||
|
||||
if (!this.config.nodeConfig.apiKey) {
|
||||
throw new Error("Remote node runtime requires nodeConfig.apiKey for authentication.");
|
||||
}
|
||||
}
|
||||
|
||||
private startHealthChecks(): void {
|
||||
if (this.healthInterval) {
|
||||
clearInterval(this.healthInterval);
|
||||
}
|
||||
|
||||
this.healthInterval = setInterval(() => {
|
||||
void this.client.health().catch((error) => {
|
||||
this.emitRuntimeError(this.toError(error));
|
||||
});
|
||||
}, 30_000);
|
||||
}
|
||||
|
||||
private startEventStreamLoop(): void {
|
||||
if (this.streamLoopPromise) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.streamLoopAbortController = new AbortController();
|
||||
this.streamLoopPromise = this.runEventStreamLoop(this.streamLoopAbortController.signal)
|
||||
.catch((error) => {
|
||||
this.emitRuntimeError(this.toError(error));
|
||||
})
|
||||
.finally(() => {
|
||||
this.streamLoopPromise = null;
|
||||
});
|
||||
}
|
||||
|
||||
private async runEventStreamLoop(signal: AbortSignal): Promise<void> {
|
||||
let reconnectAttempts = 0;
|
||||
|
||||
while (!signal.aborted && !this.isShuttingDown()) {
|
||||
let sawAnyEvent = false;
|
||||
|
||||
try {
|
||||
for await (const event of this.client.streamEvents({ signal })) {
|
||||
if (signal.aborted || this.isShuttingDown()) {
|
||||
return;
|
||||
}
|
||||
|
||||
sawAnyEvent = true;
|
||||
this.forwardRemoteEvent(event);
|
||||
}
|
||||
|
||||
if (signal.aborted || this.isShuttingDown()) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (sawAnyEvent) {
|
||||
reconnectAttempts = 0;
|
||||
}
|
||||
|
||||
throw new Error("Remote event stream ended unexpectedly");
|
||||
} catch (error) {
|
||||
if (signal.aborted || this.isShuttingDown()) {
|
||||
return;
|
||||
}
|
||||
|
||||
reconnectAttempts += 1;
|
||||
this.emitRuntimeError(this.toError(error));
|
||||
|
||||
if (reconnectAttempts >= this.maxReconnectAttempts) {
|
||||
this.setStatus("errored");
|
||||
return;
|
||||
}
|
||||
|
||||
const delayMs = Math.min(
|
||||
this.reconnectBaseDelayMs * 2 ** (reconnectAttempts - 1),
|
||||
this.maxReconnectDelayMs
|
||||
);
|
||||
|
||||
remoteNodeLog.warn(
|
||||
`Remote event stream disconnected for ${this.config.projectId}; reconnecting in ${delayMs}ms ` +
|
||||
`(attempt ${reconnectAttempts}/${this.maxReconnectAttempts})`
|
||||
);
|
||||
|
||||
await this.sleep(delayMs, signal);
|
||||
if (signal.aborted) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await this.client.health();
|
||||
} catch (healthError) {
|
||||
this.emitRuntimeError(this.toError(healthError));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private forwardRemoteEvent(event: RemoteNodeEvent): void {
|
||||
switch (event.type) {
|
||||
case "task:created":
|
||||
this.emit("task:created", event.payload as Task);
|
||||
break;
|
||||
case "task:moved": {
|
||||
const payload = event.payload as { task: Task; from: string; to: string };
|
||||
this.emit("task:moved", {
|
||||
task: payload.task,
|
||||
from: payload.from,
|
||||
to: payload.to,
|
||||
});
|
||||
break;
|
||||
}
|
||||
case "task:updated":
|
||||
this.emit("task:updated", event.payload as Task);
|
||||
break;
|
||||
case "error": {
|
||||
const payload = event.payload;
|
||||
if (payload instanceof Error) {
|
||||
this.emitRuntimeError(payload);
|
||||
} else if (typeof payload === "object" && payload && "message" in payload) {
|
||||
const message = String((payload as { message: unknown }).message);
|
||||
this.emitRuntimeError(new Error(message));
|
||||
} else {
|
||||
this.emitRuntimeError(new Error(String(payload)));
|
||||
}
|
||||
break;
|
||||
}
|
||||
default:
|
||||
remoteNodeLog.warn(
|
||||
`Ignoring unsupported remote event type "${event.type}" for ${this.config.projectId}`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async refreshMetrics(): Promise<RuntimeMetrics> {
|
||||
try {
|
||||
const metrics = await this.client.getMetrics();
|
||||
this.cachedMetrics = { ...metrics };
|
||||
this.lastSuccessfulMetricsAt = metrics.lastActivityAt;
|
||||
return metrics;
|
||||
} catch {
|
||||
const fallback: RuntimeMetrics = {
|
||||
inFlightTasks: 0,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: this.lastSuccessfulMetricsAt,
|
||||
};
|
||||
this.cachedMetrics = fallback;
|
||||
return fallback;
|
||||
}
|
||||
}
|
||||
|
||||
private setStatus(newStatus: RuntimeStatus): void {
|
||||
const previous = this.status;
|
||||
this.status = newStatus;
|
||||
|
||||
if (previous !== newStatus) {
|
||||
this.emit("health-changed", {
|
||||
status: newStatus,
|
||||
previous,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private emitRuntimeError(error: Error): void {
|
||||
if (this.listenerCount("error") > 0) {
|
||||
this.emit("error", error);
|
||||
return;
|
||||
}
|
||||
|
||||
remoteNodeLog.error(
|
||||
`Unhandled remote runtime error for ${this.config.projectId}: ${error.message}`
|
||||
);
|
||||
}
|
||||
|
||||
private isShuttingDown(): boolean {
|
||||
return this.status === "stopping" || this.status === "stopped";
|
||||
}
|
||||
|
||||
private toError(error: unknown): Error {
|
||||
return error instanceof Error ? error : new Error(String(error));
|
||||
}
|
||||
|
||||
private async sleep(ms: number, signal?: AbortSignal): Promise<void> {
|
||||
await new Promise<void>((resolve) => {
|
||||
const timeout = setTimeout(() => {
|
||||
cleanup();
|
||||
resolve();
|
||||
}, ms);
|
||||
|
||||
const onAbort = () => {
|
||||
cleanup();
|
||||
resolve();
|
||||
};
|
||||
|
||||
const cleanup = () => {
|
||||
clearTimeout(timeout);
|
||||
signal?.removeEventListener("abort", onAbort);
|
||||
};
|
||||
|
||||
signal?.addEventListener("abort", onAbort, { once: true });
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user