Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261import { randomUUID } from 'node:crypto';import type { AgentCatalog, AgentManifestV1 } from '@/daemon/agent-catalog';import type { AgentSpawnConfig, ChildHandle, ExitInfo } from '@/daemon/supervisor';
export type RuntimeState = 'stopped' | 'starting' | 'running' | 'failed';
export interface RuntimeInfo { agent_id: string; generation: string; state: RuntimeState; pid?: number; error?: string;}
export interface RuntimeSupervisor { spawn(config: AgentSpawnConfig): ChildHandle; get?(agentId: string): ChildHandle | undefined; kill(agentId: string, force?: boolean): boolean; onExit(listener: (info: ExitInfo) => void): () => void; shutdown(): Promise<void>;}
export interface RuntimeManagerDeps { catalog: AgentCatalog; supervisor: RuntimeSupervisor; /** Resolve the durable manifest to the exact child command line. */ buildSpawnConfig: (manifest: AgentManifestV1) => AgentSpawnConfig; /** Called immediately after spawn, before any readiness wait. */ onSpawn?: (agentId: string, handle: ChildHandle) => void; /** Wait for validated hello + restored sync. */ waitForHello: (handle: ChildHandle, manifest: AgentManifestV1) => Promise<void>; /** Restore the initial sync from the durable JSONL before exposing the runtime. */ restoreInitialSync?: (manifest: AgentManifestV1, handle: ChildHandle) => Promise<void>; startupTimeoutMs?: number; /** Escalate a graceful SIGTERM to SIGKILL after this delay. Set to 0 to disable. */ stopGraceMs?: number; generationId?: () => string; daemonLease?: string;}
type Listener = (runtime: RuntimeInfo) => void;
interface RuntimeEntry { info: RuntimeInfo; handle: ChildHandle | undefined; start: Promise<RuntimeInfo> | undefined; stop: Promise<void> | undefined; /** Set before signaling a child so its callback is classified as intentional. */ stopping: boolean;}
/** * Owns ephemeral child generations. AgentCatalog remains the source of identity and * session files; this class deliberately never removes either on an exit or failure. */export class RuntimeManager { private readonly catalog: AgentCatalog; private readonly supervisor: RuntimeSupervisor; private readonly buildSpawnConfig: RuntimeManagerDeps['buildSpawnConfig']; private readonly waitForHello: RuntimeManagerDeps['waitForHello']; private readonly onSpawn: RuntimeManagerDeps['onSpawn']; private readonly restoreInitialSync: RuntimeManagerDeps['restoreInitialSync']; private readonly startupTimeoutMs: number; private readonly stopGraceMs: number; private readonly generationId: () => string; private readonly daemonLease: string; private readonly runtimes = new Map<string, RuntimeEntry>(); private readonly listeners = new Set<Listener>(); private readonly unsubscribeExit: () => void;
constructor(deps: RuntimeManagerDeps) { this.catalog = deps.catalog; this.supervisor = deps.supervisor; this.buildSpawnConfig = deps.buildSpawnConfig; this.waitForHello = deps.waitForHello; this.onSpawn = deps.onSpawn; this.restoreInitialSync = deps.restoreInitialSync; this.startupTimeoutMs = deps.startupTimeoutMs ?? 10_000; this.stopGraceMs = deps.stopGraceMs ?? 5_000; if (!Number.isFinite(this.stopGraceMs) || this.stopGraceMs < 0) { throw new RangeError('stopGraceMs must be a finite non-negative number'); } this.generationId = deps.generationId ?? randomUUID; this.daemonLease = deps.daemonLease ?? randomUUID(); this.unsubscribeExit = this.supervisor.onExit((exit) => this.onExit(exit)); }
dispose(): void { this.unsubscribeExit(); this.listeners.clear(); }
onChange(listener: Listener): () => void { this.listeners.add(listener); return () => this.listeners.delete(listener); }
get(agentId: string): RuntimeInfo { return this.runtimes.get(agentId)?.info ?? { agent_id: agentId, state: 'stopped', generation: '' }; }
list(): RuntimeInfo[] { return [...this.runtimes.values()].map((entry) => ({ ...entry.info })); }
/** Start once per agent; all callers during startup receive the same promise. */ ensure(agentId: string): Promise<RuntimeInfo> { const existing = this.runtimes.get(agentId); if (existing?.stop !== undefined) return existing.stop.then(() => this.ensure(agentId)); if (existing?.start !== undefined) return existing.start; if (existing?.info.state === 'running') return Promise.resolve({ ...existing.info }); const manifest = this.catalog.get(agentId); if (manifest === undefined) return Promise.reject(new Error(`agent manifest not found: ${agentId}`));
const generation = this.generationId(); const entry: RuntimeEntry = { info: { agent_id: agentId, generation, state: 'starting' }, handle: undefined, start: undefined, stop: undefined, stopping: false, }; this.runtimes.set(agentId, entry); this.emit(entry.info); entry.start = this.start(entry, manifest); return entry.start; }
/** Stop only the current generation. Stopping an already stopped agent is harmless. */ stop(agentId: string, force = false): Promise<void> { const entry = this.runtimes.get(agentId); if (entry === undefined || entry.info.state === 'stopped') return Promise.resolve(); if (entry.stop !== undefined) { // A graceful stop may be waiting on a child that ignored SIGTERM. Allow a later // escalation to signal the same generation instead of merely returning its promise. if (force) this.supervisor.kill(agentId, true); return entry.stop; } const stopping = this.stopEntry(agentId, entry, force); entry.stop = stopping; return stopping; }
private async stopEntry(agentId: string, entry: RuntimeEntry, force: boolean): Promise<void> { const generation = entry.info.generation; entry.stopping = true; this.supervisor.kill(agentId, force); if (entry.handle !== undefined) { if (force || this.stopGraceMs === 0) { await entry.handle.exited.catch(() => undefined); } else { await waitForExitWithEscalation( entry.handle.exited, this.stopGraceMs, () => this.supervisor.kill(agentId, true), ); } } if (!this.isCurrent(agentId, generation)) return; entry.info = { agent_id: agentId, generation, state: 'stopped' }; entry.handle = undefined; entry.start = undefined; entry.stop = undefined; entry.stopping = false; this.emit(entry.info); }
async stopAll(force = false): Promise<void> { await Promise.all([...this.runtimes.keys()].map((agentId) => this.stop(agentId, force))); if (force) return; await this.supervisor.shutdown(); }
private async start(entry: RuntimeEntry, manifest: AgentManifestV1): Promise<RuntimeInfo> { try { const config = this.buildSpawnConfig(manifest); const handle = this.supervisor.spawn({ ...config, runtimeGeneration: entry.info.generation, daemonLease: this.daemonLease }); entry.handle = handle; this.onSpawn?.(manifest.agent_id, handle); entry.info = { ...entry.info, pid: handle.pid }; this.emit(entry.info); await withTimeout(this.waitForHello(handle, manifest), this.startupTimeoutMs, `hello timeout for agent "${manifest.agent_id}"`); if (this.restoreInitialSync !== undefined) { await withTimeout(this.restoreInitialSync(manifest, handle), this.startupTimeoutMs, `initial sync timeout for agent "${manifest.agent_id}"`); } if (!this.isCurrent(manifest.agent_id, entry.info.generation)) throw new Error('runtime generation superseded'); entry.info = { ...entry.info, state: 'running' }; entry.start = undefined; this.emit(entry.info); return { ...entry.info }; } catch (error) { const message = error instanceof Error ? error.message : String(error); this.supervisor.kill(manifest.agent_id, true); if (this.isCurrent(manifest.agent_id, entry.info.generation)) { entry.info = entry.stopping ? { agent_id: manifest.agent_id, generation: entry.info.generation, state: 'stopped' } : { ...entry.info, state: 'failed', error: message }; entry.start = undefined; entry.stop = undefined; entry.stopping = false; this.emit(entry.info); } throw new Error(`agent "${manifest.agent_id}" startup failed: ${message}`, { cause: error }); } }
private onExit(exit: ExitInfo): void { const entry = this.runtimes.get(exit.agentId); if (entry === undefined) return; // Supervisor callbacks do not carry a generation. A stopped/replaced entry is stale. if (entry.info.state === 'stopped' || entry.info.pid !== exit.pid) return; const intentional = entry.stopping; entry.handle = undefined; entry.start = undefined; entry.info = intentional ? { agent_id: exit.agentId, generation: entry.info.generation, state: 'stopped' } : { agent_id: exit.agentId, generation: entry.info.generation, state: 'failed', error: `child exited (${exit.reason}, code ${exit.exitCode ?? 'signal'})`, }; if (!intentional) { entry.stop = undefined; entry.stopping = false; } this.emit(entry.info); }
private isCurrent(agentId: string, generation: string): boolean { return this.runtimes.get(agentId)?.info.generation === generation; }
private emit(info: RuntimeInfo): void { const snapshot = { ...info }; for (const listener of this.listeners) listener(snapshot); }}
function withTimeout<T>(promise: Promise<T>, timeoutMs: number, message: string): Promise<T> { if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) return promise; return new Promise<T>((resolve, reject) => { const timer = setTimeout(() => reject(new Error(message)), timeoutMs); promise.then((value) => { clearTimeout(timer); resolve(value); }, (error) => { clearTimeout(timer); reject(error); }); });}
/** Wait for graceful exit, escalating the same generation to SIGKILL at the deadline. */function waitForExitWithEscalation( exited: Promise<ExitInfo>, graceMs: number, escalate: () => void,): Promise<void> { return new Promise<void>((resolve) => { const timer = setTimeout(escalate, graceMs); timer.unref?.(); exited.then( () => { clearTimeout(timer); resolve(); }, () => { clearTimeout(timer); resolve(); }, ); });}