diff --git a/packages/orchestrator/src/handler.ts b/packages/orchestrator/src/handler.ts index dab6ee6b..35ed49b9 100644 --- a/packages/orchestrator/src/handler.ts +++ b/packages/orchestrator/src/handler.ts @@ -24,6 +24,7 @@ function toInstanceSummary(instance: InstanceRecord): InstanceSummary { cwd: instance.cwd, label: instance.label, sessionId: instance.sessionId, + sessionFile: instance.sessionFile, }; } diff --git a/packages/orchestrator/src/ipc/protocol.ts b/packages/orchestrator/src/ipc/protocol.ts index b4313783..8c194c32 100644 --- a/packages/orchestrator/src/ipc/protocol.ts +++ b/packages/orchestrator/src/ipc/protocol.ts @@ -45,6 +45,7 @@ export interface InstanceSummary { cwd: string; label?: string; sessionId?: string; + sessionFile?: string; } export interface ResponseBase { diff --git a/packages/orchestrator/src/radius.ts b/packages/orchestrator/src/radius.ts new file mode 100644 index 00000000..f3aaacb5 --- /dev/null +++ b/packages/orchestrator/src/radius.ts @@ -0,0 +1,157 @@ +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"; + +const DEFAULT_RADIUS_URL = "https://radius.pi.dev/"; +const DEFAULT_ORCHESTRATOR_BASE_PATH = "/v1/"; + +interface RegisterMachineResponse { + id: string; + heartbeatIntervalMs: number; + expiresInMs: number; +} + +interface RegisterPiResponse { + id: string; + heartbeatIntervalMs: number; + expiresInMs: number; +} + +async function post(path: string, body: unknown): Promise { + const response = await fetch(new URL(path, getRadiusOrchestratorBaseUrl()), { + method: "POST", + headers: { + Authorization: `Bearer ${getRadiusApiKey()}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(body), + }); + + if (!response.ok) { + throw new Error(`Radius request failed: ${response.status} ${await response.text()}`); + } + + return (await response.json()) as T; +} + +function maybePost(path: string, body: unknown): Promise { + return fetch(new URL(path, getRadiusOrchestratorBaseUrl()), { + method: "POST", + headers: { + Authorization: `Bearer ${getRadiusApiKey()}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(body), + }); +} + +export function getRadiusUrl(): string { + return process.env.PI_RADIUS_URL || DEFAULT_RADIUS_URL; +} + +export function getRadiusOrchestratorBaseUrl(): string { + const explicitUrl = process.env.PI_RADIUS_ORCHESTRATOR_URL; + if (explicitUrl) { + return explicitUrl; + } + + return new URL(DEFAULT_ORCHESTRATOR_BASE_PATH, getRadiusUrl()).toString(); +} + +export function getRadiusApiKey(): string { + const apiKey = process.env.PI_RADIUS_API_KEY; + if (!apiKey) { + throw new Error("PI_RADIUS_API_KEY is required for Radius integration"); + } + return apiKey; +} + +export function isRadiusEnabled(): boolean { + return !!process.env.PI_RADIUS_API_KEY; +} + +export class RadiusPresence { + private heartbeatTimer?: NodeJS.Timeout; + private machine?: MachineRecord; + + async start(label?: string): Promise { + if (!isRadiusEnabled()) { + return undefined; + } + + const registered = await post("/v1/machines/register", { + label, + hostname: hostname(), + platform: platform(), + arch: process.arch, + version: "0.79.6", + capabilities: { spawn: true, relay: false, iroh: false }, + }); + + const now = new Date().toISOString(); + this.machine = { + id: registered.id, + createdAt: now, + lastSeenAt: now, + label, + }; + saveMachine(this.machine); + this.heartbeatTimer = setInterval(() => { + void this.heartbeatMachine(); + }, registered.heartbeatIntervalMs); + return this.machine; + } + + async stop(): Promise { + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = undefined; + } + if (!this.machine || !isRadiusEnabled()) { + return; + } + await maybePost(`/v1/machines/${this.machine.id}/disconnect`, {}); + this.machine = undefined; + } + + async registerPi(instance: InstanceRecord): Promise { + if (!isRadiusEnabled()) { + return instance; + } + const machine = this.machine ?? loadMachine(); + if (!machine) { + throw new Error("No registered machine available for Pi registration"); + } + const registered = await post("/v1/pis/register", { + machineId: machine.id, + label: instance.label, + cwd: instance.cwd, + hostname: hostname(), + pid: process.pid, + transport: "local-rpc", + capabilities: { rpc: true, relay: false, iroh: false }, + sessionId: instance.sessionId, + }); + return { ...instance, radiusPiId: registered.id }; + } + + async disconnectPi(instance: InstanceRecord): Promise { + if (!isRadiusEnabled() || !instance.radiusPiId) { + return; + } + await maybePost(`/v1/pis/${instance.radiusPiId}/disconnect`, {}); + } + + private async heartbeatMachine(): Promise { + if (!this.machine || !isRadiusEnabled()) { + return; + } + await maybePost(`/v1/machines/${this.machine.id}/heartbeat`, { + cwd: getOrchestratorDir(), + socketPath: getSocketPath(), + }); + } +} + +export const radiusPresence = new RadiusPresence(); diff --git a/packages/orchestrator/src/serve.ts b/packages/orchestrator/src/serve.ts index 5183ca60..97b07676 100644 --- a/packages/orchestrator/src/serve.ts +++ b/packages/orchestrator/src/serve.ts @@ -3,10 +3,17 @@ import { dirname } from "node:path"; import { getSocketPath } from "./config.ts"; import { handleIpcRequest } from "./handler.ts"; import { startIpcServer } from "./ipc/server.ts"; +import { getRadiusOrchestratorBaseUrl, isRadiusEnabled, radiusPresence } from "./radius.ts"; export async function serve(): Promise { const socketPath = getSocketPath(); mkdirSync(dirname(socketPath), { recursive: true }); + if (isRadiusEnabled()) { + await radiusPresence.start(); + console.log(`radius integration enabled: ${socketPath} -> ${getRadiusOrchestratorBaseUrl()}`); + } else { + console.log("radius integration disabled: set PI_RADIUS_API_KEY to enable"); + } const server = await startIpcServer(handleIpcRequest); console.log(`orchestrator listening on ${socketPath}`); @@ -17,6 +24,7 @@ export async function serve(): Promise { } cleanedUp = true; server.close(); + void radiusPresence.stop(); if (existsSync(socketPath)) { unlinkSync(socketPath); } diff --git a/packages/orchestrator/src/supervisor.ts b/packages/orchestrator/src/supervisor.ts index 24bcca57..d99db09a 100644 --- a/packages/orchestrator/src/supervisor.ts +++ b/packages/orchestrator/src/supervisor.ts @@ -10,6 +10,7 @@ import { type RpcResponse, SessionManager, } 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 type { InstanceRecord } from "./types.ts"; @@ -25,7 +26,7 @@ function cloneInstance(record: InstanceRecord): InstanceRecord { async function createRuntime(cwd: string): Promise { const agentDir = getAgentDir(); - const sessionManager = SessionManager.inMemory(cwd); + const sessionManager = SessionManager.create(cwd); const runtimeFactory: CreateAgentSessionRuntimeFactory = async ({ cwd, agentDir, @@ -79,11 +80,13 @@ export class OrchestratorSupervisor { lastSeenAt: now, label: options.label, sessionId: runtime.session.sessionId, + sessionFile: runtime.session.sessionFile, }; - this.liveInstances.set(record.id, { runtime, record }); - upsertInstance(record); - return cloneInstance(record); + const registeredRecord = await radiusPresence.registerPi(record); + this.liveInstances.set(registeredRecord.id, { runtime, record: registeredRecord }); + upsertInstance(registeredRecord); + return cloneInstance(registeredRecord); } async stopInstance(instanceId: string): Promise { @@ -92,6 +95,7 @@ export class OrchestratorSupervisor { return undefined; } + await radiusPresence.disconnectPi(live.record); await live.runtime.dispose(); this.liveInstances.delete(instanceId); removeInstance(instanceId); diff --git a/packages/orchestrator/src/types.ts b/packages/orchestrator/src/types.ts index d2e4d06e..c4738ccc 100644 --- a/packages/orchestrator/src/types.ts +++ b/packages/orchestrator/src/types.ts @@ -15,4 +15,6 @@ export interface InstanceRecord { lastSeenAt?: string; label?: string; sessionId?: string; + sessionFile?: string; + radiusPiId?: string; }