fix: use raius pi id for persistence

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,
 	};
 }
 

packages/orchestrator/src/ipc/protocol.ts

@@ -46,6 +46,7 @@ export interface InstanceSummary {
 	label?: string;
 	sessionId?: string;
 	sessionFile?: string;
+	radiusPiId?: string;
 }
 
 export interface ResponseBase {

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<T>(path: string, body: unknown): Promise<T> {
@@ -35,8 +32,8 @@ async function post<T>(path: string, body: unknown): Promise<T> {
 	return (await response.json()) as T;
 }
 
-function maybePost(path: string, body: unknown): Promise<Response> {
+async function maybePost(path: string, body: unknown): Promise<void> {
-	return fetch(new URL(path, getRadiusOrchestratorBaseUrl()), {
+	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<Response> {
 		},
 		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<string, NodeJS.Timeout>();
 	private machine?: MachineRecord;
 
 	async start(label?: string): Promise<MachineRecord | undefined> {
@@ -80,39 +81,44 @@ export class RadiusPresence {
 			return undefined;
 		}
 
-		const registered = await post<RegisterMachineResponse>("/v1/machines/register", {
+		const existingMachine = loadMachine();
+		const registered = await post<RegisterMachineResponse>("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,
+			createdAt: existingMachine?.createdAt ?? timestamp,
-			lastSeenAt: now,
+			lastSeenAt: timestamp,
 			label,
 		};
 		saveMachine(this.machine);
-		this.heartbeatTimer = setInterval(() => {
+		this.machineHeartbeatTimer = setInterval(() => {
 			void this.heartbeatMachine();
 		}, registered.heartbeatIntervalMs);
 		return this.machine;
 	}
 
 	async stop(): Promise<void> {
-		if (this.heartbeatTimer) {
+		if (this.machineHeartbeatTimer) {
-			clearInterval(this.heartbeatTimer);
+			clearInterval(this.machineHeartbeatTimer);
-			this.heartbeatTimer = undefined;
+			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`, {});
+		await maybePost(`machines/${this.machine.id}/disconnect`, {});
-		this.machine = undefined;
 	}
 
 	async registerPi(instance: InstanceRecord): Promise<InstanceRecord> {
@@ -123,7 +129,7 @@ export class RadiusPresence {
 		if (!machine) {
 			throw new Error("No registered machine available for Pi registration");
 		}
-		const registered = await post<RegisterPiResponse>("/v1/pis/register", {
+		const registered = await post<RegisterPiResponse>("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<void> {
+		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<void> {
 		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<void> {
+		if (!isRadiusEnabled()) {
+			return;
+		}
+		await maybePost(`pis/${radiusPiId}/heartbeat`, {});
+	}
 }
 
 export const radiusPresence = new RadiusPresence();

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<void> {
 	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;
+	let shutdownPromise: Promise<void> | undefined;
-	const cleanup = () => {
+	const shutdown = async (exitCode: number) => {
-		if (cleanedUp) {
+		if (shutdownPromise) {
-			return;
+			await shutdownPromise;
+			process.exit(exitCode);
 		}
-		cleanedUp = true;
-		server.close();
-		void radiusPresence.stop();
-		if (existsSync(socketPath)) {
-			unlinkSync(socketPath);
-		}
-	};
 
-	const shutdown = (exitCode: number) => {
+		shutdownPromise = (async () => {
-		cleanup();
+			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("SIGINT", () => {
-	process.on("SIGTERM", () => shutdown(0));
+		void shutdown(0);
-	process.on("exit", cleanup);
+	});
+	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<void>(() => {

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)) {

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<AgentSessionRuntime> {
 export class OrchestratorSupervisor {
 	private readonly liveInstances = new Map<string, LiveInstance>();
 
+	async recoverAfterRestart(): Promise<void> {
+		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<void> {
+		for (const instanceId of [...this.liveInstances.keys()]) {
+			await this.stopInstance(instanceId);
+		}
+	}
 }
 
 export const supervisor = new OrchestratorSupervisor();

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;