Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443/** * Unix-socket server — PROTOCOL §7.1 transport + §8 socketPath + §9.1 hello-on-connect * (saga tb4593.3). * * Listens on a unix-domain socket (default `~/.shard/daemon.sock`), accepts connections, * frames them with the EXISTING `JsonlDecoder`/`encodeJsonl` (LF-only, §2.1 — never * `readline`, which splits on U+2028/U+2029), and emits `hello` immediately on connect * (`PROTOCOL_VERSION` from lifecycle.ts, `server_version` from package.json — §9.1). * * Owns connection IO ONLY: accept, frame, hello, track, close. **No routing, no proxy, no * message dispatch** — that's tb4593.4 wired on top via {@link SocketServer.onConnect}. * The connection hands the caller a `JsonlDecoder`-fed `messages()` async iterable (inbound * commands) + a `send` write (outbound events), so .4 owns the multiplexing semantics and * this module owns the bytes. * * ## Socket lifecycle (§8, §13) * - **Boot:** if `socketPath` exists but no listener answers, it's a stale socket from a * crashed daemon — unlink it before bind (defensive; failure to bind then surfaces). The * caller (tb4593.6 daemon wiring) is expected to pre-ensure the parent dir exists. * - **Shutdown:** `stop()` closes the server and `unlink`s the socket file. Best-effort: * a missing file on unlink is swallowed (a concurrent `daemon.shutdown` race is fine). * * §14.1: the daemon binds a unix socket for local clients; this module is that surface. No * network exposure, no token handshake here (auth is implicit filesystem perms per tb4593 * description; token+origin gate land with the web surface, S15). */import { createConnection, createServer, type Socket, type Server } from 'node:net';import { lstat, unlink } from 'node:fs/promises';import { dirname } from 'node:path';import { mkdir } from 'node:fs/promises';import { homedir } from 'node:os';import { JsonlDecoder, encodeJsonl, type JsonlRecord } from '@/protocol/framing/jsonl';import { PROTOCOL_VERSION } from '@/protocol/lifecycle';import type { AuthContext } from './auth';import pkg from '../../package.json' with { type: 'json' };
/** `hello` capabilities advertised by the daemon socket (§9.1). `streaming-deltas` matches * the stdio agent server (src/transports/stdio.ts). */const DAEMON_CAPABILITIES: readonly string[] = ['streaming-deltas'];
/** * A server-side handle to one accepted client connection. The socket IO is delegated to * framing helpers; `messages()` yields inbound JSONL records (each a `JsonlRecord` so the * caller decides how to handle parse errors per §8), `send()` writes an outbound message. */export interface Connection { /** A stable id for this connection (UUID; used by .4 for attach-set bookkeeping). */ readonly id: string; /** Local filesystem-authenticated connection context. */ readonly auth: AuthContext; /** Underlying socket (for close/error introspection). */ readonly socket: Socket; /** Queue a message to the client (JSONL-encoded, §2.1). Returns once the OS has it. */ send(message: unknown): Promise<void>; /** Inbound JSONL records, parsed per-record (parse failures carry `parseError`). */ messages(): AsyncGenerator<JsonlRecord>; /** Whether the connection is still open. */ readonly closed: boolean;}
/** Listener invoked for every accepted connection, AFTER `hello` is sent (§9.1). */export type ConnectListener = (conn: Connection) => void;
/** Internal: a connection's mutable state shared across the `Connection` view + iteration. */interface ConnState { id: string; socket: Socket; decoder: JsonlDecoder; closed: boolean; /** Pending inbound records awaiting a `messages()` consumer. */ queue: JsonlRecord[]; /** Resolved when either a record is queued or the socket ends (so the iterator unblocks). */ drain: () => void; /** Set when the connection has been torn down (idempotent). */ teardown: () => void;}
/** Constructor options for {@link SocketServer}. All optional; sane defaults for §13. */export interface SocketServerOptions { /** Socket path (default `~/.shard/daemon.sock`). */ socketPath?: string; /** Server version string for `hello` (default: `shard <pkg.version>`). */ serverVersion?: string; /** Backlog for `listen()` (default 128). */ backlog?: number;}
/** * Unix-socket server for the daemon. Call {@link SocketServer.start} to bind, then either * subscribe via {@link SocketServer.onConnect} or iterate {@link SocketServer.connections}. * Call {@link SocketServer.stop} to close + unlink. */export class SocketServer { private readonly socketPath: string; private readonly serverVersion: string; private readonly backlog: number; private server: Server | undefined; private boundSocketIdentity: { dev: number; ino: number } | undefined; private readonly listeners = new Set<ConnectListener>(); private readonly conns = new Map<string, ConnState>(); /** Pending connections not yet consumed by an `onConnect` listener. */ private pending: Connection[] = []; /** Resolvers for an outstanding `connections()` async iterator `next()`. */ private waiters: Array<(conn: Connection) => void> = [];
constructor(opts: SocketServerOptions = {}) { this.socketPath = opts.socketPath ?? defaultSocketPath(); this.serverVersion = opts.serverVersion ?? `shard ${pkg.version ?? '0.0.0-unknown'}`; this.backlog = opts.backlog ?? 128; }
/** The configured socket path. */ get path(): string { return this.socketPath; }
/** * Bind + listen. Removes a stale socket file at `socketPath` defensively (a crashed * daemon may leave one; `EADDRINUSE` would block the bind). Ensures the parent dir * exists. Resolves once listening; never rejects (bind errors throw — caller handles). */ async start(): Promise<void> { // Ensure the parent dir exists (~/.shard/ in production). await mkdir(dirname(this.socketPath), { recursive: true });
// Never unlink a socket owned by a live daemon. Unlinking a pathname does not close // the existing listener: a second daemon can otherwise bind the same name while the // first remains alive, splitting web traffic and managed-agent callbacks across two // brokers. Only remove the path after proving that no listener accepts connections. if (await socketAcceptsConnections(this.socketPath)) { const error = new Error(`daemon socket is already in use: ${this.socketPath}`) as NodeJS.ErrnoException; error.code = 'EADDRINUSE'; throw error; } await safeUnlink(this.socketPath);
return new Promise<void>((resolve, reject) => { const server = createServer((socket) => this.handleConnection(socket)); server.on('error', (err) => { // Surface bind-time errors to the caller via the start() promise; runtime errors // (a dropped client) are handled per-socket in handleConnection. if (this.server === undefined) { reject(err); } else { // Post-start error: log to the daemon's stderr; server stays up unless fatal. process.stderr.write(`shardd: socket server error: ${err.message}\n`); } }); server.listen({ path: this.socketPath, backlog: this.backlog }, () => { this.server = server; void lstat(this.socketPath).then((stat) => { if (this.server === server) this.boundSocketIdentity = { dev: stat.dev, ino: stat.ino }; resolve(); }, (error) => { if (this.server === server) this.server = undefined; server.close(() => reject(error)); }); }); }); }
/** * Register a listener fired for every accepted connection (after `hello` is sent). * Returns an unsubscribe function. Listeners are the integration point for tb4593.4 * (proxy subscribe) and tb4593.5 (orchestrator command dispatch). */ onConnect(listener: ConnectListener): () => void { this.listeners.add(listener); // Replay pending connections a late subscriber missed (mirrors the EventBus pattern). for (const conn of this.pending) { queueMicrotask(() => listener(conn)); } this.pending = []; return () => { this.listeners.delete(listener); }; }
/** * Async-iterable of accepted connections (alternative to {@link onConnect}). Consumed * by the daemon's main accept loop if it prefers an iterator-driven shape. */ async *connections(): AsyncGenerator<Connection> { while (this.server !== undefined) { const conn = await new Promise<Connection | undefined>((resolve) => { if (this.pending.length > 0) { resolve(this.pending.shift()); return; } this.waiters.push((c) => resolve(c)); }); if (conn === undefined) { // Server stopped mid-await (a stop()-flushed waiter). End the iterator. break; } yield conn; } }
/** * Snapshot of currently-open connections (for .4 fan-out bookkeeping / `agent.attach` * resolution). Returns the live `Connection` views; callers must not hold beyond the * connection's lifetime. */ liveConnections(): Connection[] { return [...this.conns.values()].map((s) => this.toConnection(s)); }
/** * Graceful shutdown: stop accepting, destroy live sockets, unlink the socket file. * Resolves once the server's `close` callback fires. Idempotent (subsequent calls no-op). */ async stop(): Promise<void> { const server = this.server; if (server === undefined) { return; } this.server = undefined;
// Flush any outstanding `connections()` waiters so they unblock + end their loop. while (this.waiters.length > 0) { this.waiters.shift()?.(undefined as unknown as Connection); } this.pending = [];
// Destroy every live connection (tears down per-conn decoder + resolves iterators). for (const state of [...this.conns.values()]) { state.teardown(); }
await new Promise<void>((resolve) => { server.close(() => resolve()); }); const identity = this.boundSocketIdentity; this.boundSocketIdentity = undefined; if (identity !== undefined) await unlinkIfOwned(this.socketPath, identity); }
// ─── Internal ──────────────────────────────────────────────────────────────────
/** Wire up one accepted socket: hello-on-connect, decoder, inbound queue, teardown. */ private handleConnection(socket: Socket): void { if (this.server === undefined) { // Server stopped between the `listen` callback and this `connection` event. Refuse. socket.destroy(); return; } const id = crypto.randomUUID(); const decoder = new JsonlDecoder(); let waiter: (() => void) | undefined;
const state: ConnState = { id, socket, decoder, closed: false, queue: [], drain: () => { waiter?.(); }, teardown: () => { if (state.closed) return; state.closed = true; try { socket.destroy(); } catch { // Already destroyed. } // Drop all server-owned references as soon as the socket closes. Without this, // every historical client socket remained reachable for the daemon's lifetime. this.conns.delete(id); this.pending = this.pending.filter((conn) => conn.id !== id); // Unblock a `messages()` consumer waiting for the next record (so it observes EOF). state.drain(); }, }; this.conns.set(id, state);
// §9.1 hello — first on connect, before any inbound command can arrive. Validate // against HelloSchema shape (mirrors the stdio server's sendRaw). Caller checks the // protocol_version per §7. const hello = { type: 'hello' as const, protocol_version: PROTOCOL_VERSION, server_version: this.serverVersion, capabilities: [...DAEMON_CAPABILITIES], }; socket.write(encodeJsonl(hello));
// Feed inbound bytes to the decoder; each completed record is queued + drains the // `messages()` iterator. A socket `error` (drop, reset) tears the conn down. socket.on('data', (chunk: Buffer) => { decoder.write(chunk); const records = decoder.parse(); if (records.length > 0) { state.queue.push(...records); state.drain(); } }); socket.on('error', (err) => { if (!state.closed) { process.stderr.write(`shardd: connection ${id} error: ${err.message}\n`); } state.teardown(); }); socket.on('close', () => { state.teardown(); });
const conn = this.toConnection(state);
// Distribute to subscribers + the async iterator queue. if (this.listeners.size > 0) { for (const listener of this.listeners) { queueMicrotask(() => listener(conn)); } } else { // No subscribers yet — buffer for a late `onConnect`/`connections()` consumer. this.pending.push(conn); } if (this.waiters.length > 0) { const resolve = this.waiters.shift(); if (resolve !== undefined) { // Don't re-buffer — hand straight to the waiter, skipping `pending` entirely. const idx = this.pending.indexOf(conn); if (idx >= 0) this.pending.splice(idx, 1); queueMicrotask(() => resolve(conn)); } } }
/** Build the public `Connection` view from internal `ConnState`. */ private toConnection(state: ConnState): Connection { return { id: state.id, auth: { kind: 'local' }, socket: state.socket, get closed() { return state.closed; }, send: (message: unknown) => writeEnqueued(state.socket, message), messages: () => connMessages(state), }; }}
// ─── Module-private helpers ──────────────────────────────────────────────────────
/** Resolve the default socket path (~/.shard/daemon.sock, §13). */function defaultSocketPath(): string { return `${homedir()}/.shard/daemon.sock`;}
/** * Write a JSONL-encoded message to a socket, resolving once the bytes are queued. Uses * `socket.write`'s callback to detect backpressure-friendly completion. Rejects if the * socket is closed before the write flushes (caller decides whether to drop or requeue). */function writeEnqueued(socket: Socket, message: unknown): Promise<void> { return new Promise<void>((resolve, reject) => { if (socket.destroyed) { reject(new Error('socket closed')); return; } const ok = socket.write(encodeJsonl(message), (err) => { if (err !== undefined && err !== null) { reject(err); } else { resolve(); } }); void ok; // backpressure handled by node's highWaterMark; v0.1 doesn't cork on `false`. });}
/** The inbound messages async-iterable for one connection. */async function* connMessages(state: ConnState): AsyncGenerator<JsonlRecord> { while (!state.closed) { // Drain whatever is buffered synchronously first (no await if the queue has records). if (state.queue.length > 0) { yield state.queue.shift() as JsonlRecord; continue; } // Wait for either new records or teardown (EOF). One promise per stall. await new Promise<void>((resolve) => { // `drain` is overwritten per stall so only the current waiter unblocks. state.drain = () => resolve(); }); if (state.queue.length === 0 && state.closed) { // Teardown woke us with nothing left to yield — EOF. return; } }}
/** * Probe an existing Unix socket without mutating its pathname. A successful connect proves * that another daemon owns it. Only ENOENT/ECONNREFUSED prove the path is stale; timeouts * and unusual errors fail closed so startup never destroys a listener it could not verify. */async function socketAcceptsConnections(path: string): Promise<boolean> { return new Promise<boolean>((resolve) => { const socket = createConnection({ path }); let settled = false; const finish = (live: boolean) => { if (settled) return; settled = true; clearTimeout(timer); socket.destroy(); resolve(live); }; const timer = setTimeout(() => finish(true), 250); timer.unref?.(); socket.once('connect', () => finish(true)); socket.once('error', (error: NodeJS.ErrnoException) => { finish(error.code !== 'ENOENT' && error.code !== 'ECONNREFUSED'); }); });}
/** Remove the pathname only when it still names the socket this server originally bound. */async function unlinkIfOwned(path: string, identity: { dev: number; ino: number }): Promise<void> { try { const stat = await lstat(path); if (stat.dev === identity.dev && stat.ino === identity.ino) await unlink(path); } catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; }}
/** Unlink, swallowing "not found" (a stale socket is the normal pre-bind state). */async function safeUnlink(path: string): Promise<void> { try { await unlink(path); } catch (err) { const code = (err as NodeJS.ErrnoException).code; if (code !== 'ENOENT') { // A different error (EACCES, EISDIR) is a real bind-blocker — let it surface. throw err; } }}