diff --git a/packages/orchestrator/src/handler.ts b/packages/orchestrator/src/handler.ts index 35ed49b9..bfac58a3 100644 --- a/packages/orchestrator/src/handler.ts +++ b/packages/orchestrator/src/handler.ts @@ -25,6 +25,7 @@ function toInstanceSummary(instance: InstanceRecord): InstanceSummary { label: instance.label, sessionId: instance.sessionId, sessionFile: instance.sessionFile, + radiusPiId: instance.radiusPiId, }; } diff --git a/packages/orchestrator/src/ipc/protocol.ts b/packages/orchestrator/src/ipc/protocol.ts index 8c194c32..796418ab 100644 --- a/packages/orchestrator/src/ipc/protocol.ts +++ b/packages/orchestrator/src/ipc/protocol.ts @@ -46,6 +46,7 @@ export interface InstanceSummary { label?: string; sessionId?: string; sessionFile?: string; + radiusPiId?: string; } export interface ResponseBase { diff --git a/packages/orchestrator/src/radius.ts b/packages/orchestrator/src/radius.ts index f3aaacb5..39a9237a 100644 --- a/packages/orchestrator/src/radius.ts +++ b/packages/orchestrator/src/radius.ts @@ -1,21 +1,18 @@ import { hostname, platform } from "node:os"; import { getOrchestratorDir, getSocketPath } from "./config.ts"; import { loadMachine, saveMachine } from "./storage.ts"; -import type { InstanceRecord, MachineRecord } from "./types.ts"; +import type { InstanceRecord, MachineRecord, RadiusRegistration } from "./types.ts"; const DEFAULT_RADIUS_URL = "https://radius.pi.dev/"; const DEFAULT_ORCHESTRATOR_BASE_PATH = "/v1/"; +const ORCHESTRATOR_VERSION = "0.79.6"; -interface RegisterMachineResponse { +interface RegisterMachineResponse extends RadiusRegistration { id: string; - heartbeatIntervalMs: number; - expiresInMs: number; } -interface RegisterPiResponse { +interface RegisterPiResponse extends RadiusRegistration { id: string; - heartbeatIntervalMs: number; - expiresInMs: number; } async function post(path: string, body: unknown): Promise { @@ -35,8 +32,8 @@ async function post(path: string, body: unknown): Promise { return (await response.json()) as T; } -function maybePost(path: string, body: unknown): Promise { - return fetch(new URL(path, getRadiusOrchestratorBaseUrl()), { +async function maybePost(path: string, body: unknown): Promise { + const response = await fetch(new URL(path, getRadiusOrchestratorBaseUrl()), { method: "POST", headers: { Authorization: `Bearer ${getRadiusApiKey()}`, @@ -44,6 +41,9 @@ function maybePost(path: string, body: unknown): Promise { }, body: JSON.stringify(body), }); + if (!response.ok) { + throw new Error(`Radius request failed: ${response.status} ${await response.text()}`); + } } export function getRadiusUrl(): string { @@ -72,7 +72,8 @@ export function isRadiusEnabled(): boolean { } export class RadiusPresence { - private heartbeatTimer?: NodeJS.Timeout; + private machineHeartbeatTimer?: NodeJS.Timeout; + private readonly piHeartbeatTimers = new Map(); private machine?: MachineRecord; async start(label?: string): Promise { @@ -80,39 +81,44 @@ export class RadiusPresence { return undefined; } - const registered = await post("/v1/machines/register", { + const existingMachine = loadMachine(); + const registered = await post("machines/register", { + machineId: existingMachine?.id, label, hostname: hostname(), platform: platform(), arch: process.arch, - version: "0.79.6", + version: ORCHESTRATOR_VERSION, capabilities: { spawn: true, relay: false, iroh: false }, }); - const now = new Date().toISOString(); + const timestamp = new Date().toISOString(); this.machine = { id: registered.id, - createdAt: now, - lastSeenAt: now, + createdAt: existingMachine?.createdAt ?? timestamp, + lastSeenAt: timestamp, label, }; saveMachine(this.machine); - this.heartbeatTimer = setInterval(() => { + this.machineHeartbeatTimer = setInterval(() => { void this.heartbeatMachine(); }, registered.heartbeatIntervalMs); return this.machine; } async stop(): Promise { - if (this.heartbeatTimer) { - clearInterval(this.heartbeatTimer); - this.heartbeatTimer = undefined; + if (this.machineHeartbeatTimer) { + clearInterval(this.machineHeartbeatTimer); + this.machineHeartbeatTimer = undefined; + } + for (const [instanceId, timer] of this.piHeartbeatTimers) { + clearInterval(timer); + this.piHeartbeatTimers.delete(instanceId); } if (!this.machine || !isRadiusEnabled()) { return; } - await maybePost(`/v1/machines/${this.machine.id}/disconnect`, {}); - this.machine = undefined; + await maybePost(`machines/${this.machine.id}/disconnect`, {}); } async registerPi(instance: InstanceRecord): Promise { @@ -123,7 +129,7 @@ export class RadiusPresence { if (!machine) { throw new Error("No registered machine available for Pi registration"); } - const registered = await post("/v1/pis/register", { + const registered = await post("pis/register", { machineId: machine.id, label: instance.label, cwd: instance.cwd, @@ -133,25 +139,49 @@ export class RadiusPresence { capabilities: { rpc: true, relay: false, iroh: false }, sessionId: instance.sessionId, }); + this.startPiHeartbeat(instance.id, registered.heartbeatIntervalMs, registered.id); return { ...instance, radiusPiId: registered.id }; } async disconnectPi(instance: InstanceRecord): Promise { + const timer = this.piHeartbeatTimers.get(instance.id); + if (timer) { + clearInterval(timer); + this.piHeartbeatTimers.delete(instance.id); + } if (!isRadiusEnabled() || !instance.radiusPiId) { return; } - await maybePost(`/v1/pis/${instance.radiusPiId}/disconnect`, {}); + await maybePost(`pis/${instance.radiusPiId}/disconnect`, {}); + } + + private startPiHeartbeat(instanceId: string, intervalMs: number, radiusPiId: string): void { + const existingTimer = this.piHeartbeatTimers.get(instanceId); + if (existingTimer) { + clearInterval(existingTimer); + } + const timer = setInterval(() => { + void this.heartbeatPi(radiusPiId); + }, intervalMs); + this.piHeartbeatTimers.set(instanceId, timer); } private async heartbeatMachine(): Promise { if (!this.machine || !isRadiusEnabled()) { return; } - await maybePost(`/v1/machines/${this.machine.id}/heartbeat`, { + await maybePost(`machines/${this.machine.id}/heartbeat`, { cwd: getOrchestratorDir(), socketPath: getSocketPath(), }); } + + private async heartbeatPi(radiusPiId: string): Promise { + if (!isRadiusEnabled()) { + return; + } + await maybePost(`pis/${radiusPiId}/heartbeat`, {}); + } } export const radiusPresence = new RadiusPresence(); diff --git a/packages/orchestrator/src/serve.ts b/packages/orchestrator/src/serve.ts index 97b07676..42d785bc 100644 --- a/packages/orchestrator/src/serve.ts +++ b/packages/orchestrator/src/serve.ts @@ -4,47 +4,57 @@ import { getSocketPath } from "./config.ts"; import { handleIpcRequest } from "./handler.ts"; import { startIpcServer } from "./ipc/server.ts"; import { getRadiusOrchestratorBaseUrl, isRadiusEnabled, radiusPresence } from "./radius.ts"; +import { supervisor } from "./supervisor.ts"; export async function serve(): Promise { const socketPath = getSocketPath(); mkdirSync(dirname(socketPath), { recursive: true }); + await supervisor.recoverAfterRestart(); if (isRadiusEnabled()) { - await radiusPresence.start(); + const machine = await radiusPresence.start(); console.log(`radius integration enabled: ${socketPath} -> ${getRadiusOrchestratorBaseUrl()}`); + if (machine) { + console.log(`radius machine id: ${machine.id}`); + } } else { console.log("radius integration disabled: set PI_RADIUS_API_KEY to enable"); } const server = await startIpcServer(handleIpcRequest); console.log(`orchestrator listening on ${socketPath}`); - let cleanedUp = false; - const cleanup = () => { - if (cleanedUp) { - return; + let shutdownPromise: Promise | undefined; + const shutdown = async (exitCode: number) => { + if (shutdownPromise) { + await shutdownPromise; + process.exit(exitCode); } - cleanedUp = true; - server.close(); - void radiusPresence.stop(); - if (existsSync(socketPath)) { - unlinkSync(socketPath); - } - }; - const shutdown = (exitCode: number) => { - cleanup(); + shutdownPromise = (async () => { + server.close(); + await supervisor.shutdown(); + await radiusPresence.stop(); + if (existsSync(socketPath)) { + unlinkSync(socketPath); + } + })(); + + await shutdownPromise; process.exit(exitCode); }; - process.on("SIGINT", () => shutdown(0)); - process.on("SIGTERM", () => shutdown(0)); - process.on("exit", cleanup); + process.on("SIGINT", () => { + void shutdown(0); + }); + process.on("SIGTERM", () => { + void shutdown(0); + }); process.on("uncaughtException", (error) => { console.error(error); - shutdown(1); + void shutdown(1); }); process.on("unhandledRejection", (reason) => { console.error(reason); - shutdown(1); + void shutdown(1); }); await new Promise(() => { diff --git a/packages/orchestrator/src/storage.ts b/packages/orchestrator/src/storage.ts index 29eb3213..7c75f185 100644 --- a/packages/orchestrator/src/storage.ts +++ b/packages/orchestrator/src/storage.ts @@ -1,4 +1,4 @@ -import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { getInstancesPath, getMachinePath, getOrchestratorDir } from "./config.ts"; import type { InstanceRecord, MachineRecord } from "./types.ts"; @@ -24,6 +24,14 @@ export function saveMachine(machine: MachineRecord): void { writeFileSync(getMachinePath(), JSON.stringify(machine, null, 2)); } +export function deleteMachine(): void { + const machinePath = getMachinePath(); + if (!existsSync(machinePath)) { + return; + } + rmSync(machinePath); +} + export function loadInstances(): InstanceRecord[] { const instancesPath = getInstancesPath(); if (!existsSync(instancesPath)) { diff --git a/packages/orchestrator/src/supervisor.ts b/packages/orchestrator/src/supervisor.ts index d99db09a..38b4025a 100644 --- a/packages/orchestrator/src/supervisor.ts +++ b/packages/orchestrator/src/supervisor.ts @@ -12,7 +12,7 @@ import { } from "@earendil-works/pi-coding-agent"; import { radiusPresence } from "./radius.ts"; import { handleRpcCommand } from "./rpc-bridge.ts"; -import { getInstance, loadInstances, removeInstance, upsertInstance } from "./storage.ts"; +import { getInstance, loadInstances, removeInstance, saveInstances, upsertInstance } from "./storage.ts"; import type { InstanceRecord } from "./types.ts"; interface LiveInstance { @@ -56,6 +56,19 @@ async function createRuntime(cwd: string): Promise { export class OrchestratorSupervisor { private readonly liveInstances = new Map(); + async recoverAfterRestart(): Promise { + const recoveredAt = new Date().toISOString(); + const instances = loadInstances().map((instance) => ({ + ...instance, + status: instance.status === "online" || instance.status === "starting" ? "stopped" : instance.status, + lastSeenAt: recoveredAt, + })); + for (const instance of instances) { + await radiusPresence.disconnectPi(instance); + } + saveInstances(instances); + } + listInstances(): InstanceRecord[] { return loadInstances().map(cloneInstance); } @@ -110,6 +123,12 @@ export class OrchestratorSupervisor { return handleRpcCommand(live.runtime, command); } + + async shutdown(): Promise { + for (const instanceId of [...this.liveInstances.keys()]) { + await this.stopInstance(instanceId); + } + } } export const supervisor = new OrchestratorSupervisor(); diff --git a/packages/orchestrator/src/types.ts b/packages/orchestrator/src/types.ts index c4738ccc..847b2943 100644 --- a/packages/orchestrator/src/types.ts +++ b/packages/orchestrator/src/types.ts @@ -7,6 +7,11 @@ export interface MachineRecord { label?: string; } +export interface RadiusRegistration { + heartbeatIntervalMs: number; + expiresInMs: number; +} + export interface InstanceRecord { id: string; status: InstanceStatus;