Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
4.9 kB · 135 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136import type { DaemonCommandBody } from "./daemon-client.js";import { DaemonClient, DaemonSocketClosedError } from "./daemon-client.js";import type { DaemonResponse } from "./daemon-protocol.js";
/** * Long-lived supervisor connection for a daemon worker. * * Holds one persistent connection and multiplexes cross-worker requests * (agent messages, roster reads, root-session creation, renames) over * it. Socket death is expected during supervisor restarts: teardown * happens on close or handshake failure, and the next request * reconnects. Requests are never retried (daemon commands are not * idempotent). */
export interface DaemonClientLike { connect(timeoutMs: number): Promise<void>; waitForHello(timeoutMs?: number): Promise<unknown>; request(command: DaemonCommandBody, timeoutMs: number): Promise<DaemonResponse>; onClose(listener: () => void): () => void; close(): void;}
export interface SupervisorLinkOptions { socketPath: string; connectTimeoutMs?: number; /** Request-level default; call sites may override per request. */ requestTimeoutMs?: number; /** Test seam: client factory. */ factory?: (socketPath: string) => DaemonClientLike;}
export class SupervisorLink { private client?: DaemonClientLike; /** Client mid-handshake; teardown must close it too (close() during connect). */ private pendingClient?: DaemonClientLike; private connecting?: Promise<DaemonClientLike>; private closed = false; private readonly disposers = new Set<() => void>();
constructor(private readonly options: SupervisorLinkOptions) {}
/** * Send one request over the persistent link. Never retries: daemon * commands like create or send_message are not idempotent, so a * failed request surfaces its error to the caller. Only * DaemonSocketClosedError tears the link down; the NEXT request * reconnects. Command-level failures (timeouts, rejections) leave * the shared connection serving other in-flight requests. Call * sites that need an establishment window use ensureConnected in * their own retry loop. */ async request( command: DaemonCommandBody, timeoutMs: number = this.options.requestTimeoutMs ?? 30_000, ): Promise<DaemonResponse> { if (this.closed) throw new Error("Supervisor link is closed"); const client = await this.ensureConnected(); try { return await client.request(command, timeoutMs); } catch (error) { // A command-level failure (timeout, rejection) leaves a healthy // socket serving other in-flight requests; only socket death // invalidates the shared connection. if (error instanceof DaemonSocketClosedError) this.teardown(); throw error; } }
/** Establish (or reuse) the authenticated connection. */ async ensureConnected(): Promise<DaemonClientLike> { if (this.closed) throw new Error("Supervisor link is closed"); return this.ensureConnectedInternal(); }
private async ensureConnectedInternal(): Promise<DaemonClientLike> { if (this.client) return this.client; this.connecting ??= (async () => { const client = ( this.options.factory ?? ((socketPath: string) => new DaemonClient(socketPath) as DaemonClientLike) )(this.options.socketPath); this.pendingClient = client; const detach = client.onClose(() => this.teardown()); this.disposers.add(detach); const cleanup = () => { // DaemonClient fires onClose only after a successful connect, // so a failed handshake must close its own socket here. this.disposers.delete(detach); detach(); if (this.pendingClient === client) this.pendingClient = undefined; client.close(); }; try { await client.connect(this.options.connectTimeoutMs ?? 1000); await client.waitForHello(); } catch (error) { cleanup(); throw error; } if (this.closed || this.client !== undefined) { // Teardown ran mid-handshake or another coroutine won the race. cleanup(); throw new Error("Supervisor link was closed during handshake"); } this.pendingClient = undefined; this.client = client; return client; })(); const connecting = this.connecting; try { return await connecting; } finally { // A teardown during the handshake lets a later caller install a // newer in-flight promise; this awaiter must never clear that one. if (this.connecting === connecting) this.connecting = undefined; } }
/** Drop the cached connection; the next request reconnects. */ teardown(): void { const clients = [this.client, this.pendingClient]; this.client = undefined; this.pendingClient = undefined; this.connecting = undefined; for (const detach of this.disposers) detach(); this.disposers.clear(); for (const client of clients) client?.close(); }
/** Stop the link permanently; further requests fail fast without reconnecting. */ close(): void { this.closed = true; this.teardown(); }}