Something went wrong. Try again.
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459/** * Proxy + multiplexer — ARCHITECTURE §8 responsibility 2 + PROTOCOL §6 multi-client * semantics (saga tb4593.4). * * Routes per-agent protocol messages between daemon connections (§5.2 daemon mode) and * agent child stdio (§5.2 DIRECT mode — the `shard agent` stdio server strips `agent_id` * outbound, so the proxy is the ONLY place that adds the routing key on fan-out and * strips it on inbound routing). Owns: * * - **Attachment sets** — per-agent `Set<Connection>` (§6.1: subscription is implicit in * `agent.attach`; the set a client watches = what it has attached & not detached). * - **Inbound routing** — per-agent command from a daemon client → strip routing * `agent_id` → write to the agent child's stdin (direct form). * - **Outbound fan-out** — agent event from child stdout → tag `agent_id` → send to every * attached client (§6.2). * - **First-response-wins** (§6.3) — `permission_request`/`ui_request` broadcast to all * attached clients; the FIRST `permission_response`/`ui_response` is forwarded to the * agent, late ones dropped silently; the agent's `permission_resolved`/`ui_resolved` * fans out to all (so clients dismiss dialogs). * - **Orchestrator dispatch seam** — orchestrator commands (§10.1) are NOT routed; they're * handed to an injectable handler (tb4593.5 plugs in). `agent.attach`/`detach`/`kill` * carry `agent_id` as a PAYLOAD target arg, not a routing key — so the proxy validates * routing `agent_id` ONLY on per-agent commands (see `requiresAgentIdOnDaemon`). * * ## Decoupled seams (wired by tb4593.6 daemon wiring) * - `getStream(agentId)` → `{stdin, stdout} | undefined` — backed by `Supervisor.get` * (tb4593.1). Keeps the broker free of restart/promise-identity concerns. * - `getSync(agentId)` → `{status, mode, live_window, pending_requests} | undefined` — backed * by `Registry` (tb4593.2) for status + the session store for `live_window`. v0.1 * minimal sync returns empty `live_window` (the real window needs the agent child's * SessionStore; enhancement for .6). * - `setOrchestratorHandler(fn)` — tb4593.5 plugs in `agent.*` + `daemon.shutdown`. * * All routing/fan-out logic is transport-neutral: the broker speaks to `Connection` * (from tb4593.3's SocketServer) and to agent stdio streams. No new message types — reuses * `Envelope`, `validateMessage`, `requiresAgentIdOnDaemon`, `omitRoutingAgentId`, the * `JsonlDecoder`, and `encodeJsonl`. */import { JsonlDecoder, encodeJsonl, type JsonlRecord } from '@/protocol/framing/jsonl';import { omitRoutingAgentId, validateAgentId, type Envelope,} from '@/protocol/envelope';import type { AttachmentRole } from '@/protocol/commands';import { requiresAgentIdOnDaemon, validateMessage, isOrchestratorMessage,} from '@/protocol/catalog';import type { AgentStatus } from '@/contracts/agent-info';import type { ErrorCode } from '@/protocol/error-codes';import { PermissionRequestSchema, UiRequestSchema,} from '@/contracts/event';import type { Static } from '@sinclair/typebox';
/** A pending permission or UI request held by the broker for an in-flight turn. */export type PendingRequest = | Static<typeof PermissionRequestSchema> | Static<typeof UiRequestSchema>;
/** * The minimal connection view the broker needs from tb4593.3's `SocketServer.Connection` * (a structural subset — duck-typed so a fake stands in for tests). */export interface ProxyConnection { /** Stable per-connection id (used as the attachment-set identity). */ readonly id: string; /** Immutable transport authentication/capability context. */ readonly auth?: import('./auth.js').AuthContext; /** Send a JSONL-encoded message to the client. Rejects if the connection is closed. */ send(message: unknown): Promise<void>; /** Whether the connection has been torn down. */ readonly closed: boolean;}
/** A readable/writable pair backing one agent child's direct-connection stdio. */export interface AgentStream { /** Where the proxy writes (agent_id-stripped) commands — the child's stdin. */ stdin: NodeJS.WritableStream; /** Where the proxy reads (agent_id-less) events — the child's stdout. */ stdout: NodeJS.ReadableStream;}
/** Sync payload the broker emits on attach (§9.1 `sync`; mirrors `SyncSchema`). */export interface SyncPayload { status: AgentStatus; mode: string; live_window: { messages: unknown[]; compaction_index?: number }; pending_requests: PendingRequest[];}
/** Injectable seam: resolve an agent's stdio stream (undefined = not running). */export type StreamProvider = (agentId: string) => AgentStream | undefined;
/** Lazily restore an idle-stopped runtime before forwarding a new command. */export type RuntimeEnsurer = (agentId: string) => Promise<void>;
/** Injectable seam: resolve an agent's sync payload for the `sync` event (§9.1). */export type SyncProvider = (agentId: string) => SyncPayload | undefined;
/** * Durable parent lookup (saga u4uo9m.2). Resolves the durable parent agent id for a child * (backed by `AgentCatalog.spawn_options.parent_agent_id`), or `undefined` for a root / * unknown agent. Used to route a delegated child's permission/UI requests up the hierarchy * to the nearest interactive ancestor when the child has no interactive responder. */export type ParentResolver = (agentId: string) => string | undefined;
/** * Internal tagged event subscriber (saga u4uo9m.2). The `SubagentJobCoordinator` (.4) * subscribes to per-agent events through this seam — it is NOT an attachment and NEVER * counts as a permission/UI responder. Events are delivered synchronously in stdout-pump * order (one at a time, before fan-out) so the coordinator's lease state stays consistent * with event dispatch. The `tag` lets one coordinator register under a stable key and * replace itself cleanly. */export type InternalEventSubscriber = (agentId: string, event: Envelope) => void;
/** A lease-guard decision returned by {@link SubagentLeaseGuard.inspect}. */export type LeaseAction = 'allow' | 'conflict' | 'delegate';
/** Outcome of {@link SubagentLeaseGuard.inspect} for one inbound per-agent command. */export interface LeaseInspection { /** `allow` = forward normally (unleased, or a response that always passes); `conflict` * = reject with a CONFLICT error and do NOT forward; `delegate` = hand the command to the * coordinator's strict active-turn / cancellation path (the coordinator replies itself). */ action: LeaseAction; /** When `action === 'conflict'`: the user-facing error message. */ message?: string;}
/** * Inbound lease guard (saga u4uo9m.2). The broker consults it in `routeInbound` for every * per-agent command BEFORE forwarding to the child. For a leased agent: * - `prompt` → `conflict` (one Agent may have at most one queued/active run; direct prompt * would misattribute results); * - `permission_response` / `ui_response` → `allow` (responses resolve the child gate; * first-response-wins already dropped late ones); * - `steer` → `delegate` (the coordinator's strict active-turn handshake — bound to the * leased `run_id`/`turn_id` — never the idle queue); * - `abort` → `delegate` (the coordinator's cancellation path, recorded against the run). * Unleased agents: `inspect` returns `allow` for everything (backward parity). The default * guard (no coordinator wired) always allows. Ordering (saga u4uo9m.2 item 6): the broker * delivers events to the internal subscriber synchronously in pump order, so the * coordinator's lease state (turn_start binding / turn_end clearing) is fresh before any * concurrent `inspect`; the coordinator's `delegate` does the atomic turn-binding so a * matching `turn_end` can never race an accepted strict steer into a future run. */export interface SubagentLeaseGuard { /** Inspect one inbound per-agent command for a leased agent. */ inspect(agentId: string, msg: Envelope): LeaseInspection; /** Delegate a direct human `steer`/`abort` on a leased child through the coordinator's * strict active-turn / cancellation path. Resolves after the coordinator has replied to * `conn` (injected/`conflict` ack) or forwarded the strict steer to the child stdin. */ delegate(agentId: string, msg: Envelope, conn: ProxyConnection): Promise<void>;}
/** * Status-change bridge (tb4593.6 wiring). Invoked from the stdout pump when the agent * emits a per-agent `status` event (§9.1) — the daemon plugs in `orchestrator.reportStatusChange` * so the registry + the §10.2 `agent_status_changed` broadcast stay in sync with the live * agent status. The per-agent `status` event STILL fans out to attached clients (§6.2) — * this hook is the additional registry-level mirror, not a replacement. */export type AgentStatusListener = ( agentId: string, status: AgentStatus, detail?: string,) => void;
/** Signals accepted work before the child emits its first turn event. */export type AgentActivityListener = (agentId: string) => void;
/** Emits a sanitized global activity summary; never receives the source event payload. */export type AgentOverviewActivityListener = ( agentId: string, phase: 'idle' | 'thinking' | 'streaming' | 'tool' | 'waiting' | 'compacting' | 'retrying' | 'error', toolName?: string,) => void;
/** Mirrors a successfully changed agent mode into daemon-owned bootstrap state. */export type AgentModeListener = (agentId: string, mode: string) => void;
/** * Mirrors a successfully changed agent model selection into daemon-owned bootstrap state * (saga vr7kgx.6). Fired from the child's `model_changed` event — only real changes broadcast — * so the registry's `model`/`provider` projection stays authoritative for reconnect sync * without exposing secrets or changing the private routing model. */export type AgentModelListener = ( agentId: string, model: { id: string; provider: string; model: string },) => void;
/** * Mirrors a successfully changed agent reasoning effort into daemon-owned bootstrap state * (saga p7m8jp.3). Fired from the child's `reasoning_level_changed` event — only genuine * changes broadcast — so the registry's `reasoning_effort` projection stays authoritative * for reconnect sync without exposing secrets. */export type AgentReasoningEffortListener = (agentId: string, effort: string) => void;
/** * Mirrors a per-agent `usage` event into daemon-owned runtime state (saga vr7kgx.7). Fired * from the child's `usage` event — the secret-free token/cost accounting the provider * reported for the last call. The orchestrator persists it onto the durable `AgentInfo` * so a later `session.status` / reconnect surfaces the last-known context size, and * re-emits the sanitized session runtime projection so attached clients see live usage * (the `usage` event itself folds into the transcript; this is the runtime-status mirror). * Sanitized to the `Usage` shape (numbers only); never carries secrets or routing ids. */export type AgentUsageListener = ( agentId: string, usage: { input: number; output: number; cache_read?: number; cache_write?: number; cost?: number; context_tokens?: number; model?: string; provider?: string },) => void;
/** * Orchestrator command handler (tb4593.5). Receives a validated orchestrator command + * the connection it arrived on; resolves once the handler has replied (or streamed an * error). The broker does NOT interpret orchestrator commands — it only routes them here. */export type OrchestratorHandler = (cmd: Envelope, conn: ProxyConnection) => Promise<void>;
/** Broker constructor options. */export interface BrokerOptions { /** Resolve an agent's stdio stream (backed by `Supervisor.get`, tb4593.1). */ getStream: StreamProvider; /** Restore a stopped runtime on demand; omitted by isolated proxy tests. */ ensureRuntime?: RuntimeEnsurer; /** Resolve an agent's sync payload (backed by `Registry` + session store). */ getSync?: SyncProvider; /** Durable parent lookup for hierarchical permission/UI routing (saga u4uo9m.2). * Backed by `AgentCatalog.spawn_options.parent_agent_id`; omitted in isolated tests. */ parentOf?: ParentResolver; /** Inbound lease guard for daemon-managed runs (saga u4uo9m.2). Default = always allow. */ leaseGuard?: SubagentLeaseGuard; /** Bridge effective agent execution status to the orchestrator. */ onAgentStatus?: AgentStatusListener; /** Cancel idle cleanup as soon as turn work is accepted. */ onAgentActivity?: AgentActivityListener; /** Publish sanitized activity summaries to all daemon clients. */ onAgentOverviewActivity?: AgentOverviewActivityListener; /** Mirror successful `set_mode` responses into agent bootstrap state. */ onAgentMode?: AgentModeListener; /** Mirror successful `model_changed` events into agent bootstrap state (saga vr7kgx.6). */ onAgentModel?: AgentModelListener; /** Mirror per-agent `usage` events into durable runtime state (saga vr7kgx.7). */ onAgentUsage?: AgentUsageListener; /** Injectable clock (default `Date.now`) — for deterministic `sync` timestamps if needed. */ now?: () => number; /** Headless fallback delay in milliseconds (default: approximately 60 seconds). */ headlessTimeoutMs?: number; /** Injectable timer seams for deterministic headless-fallback tests. */ setTimeout?: (callback: () => void, delay: number) => ReturnType<typeof globalThis.setTimeout>; clearTimeout?: (timer: ReturnType<typeof globalThis.setTimeout>) => void;}
/** Per-agent broker state: role-aware attachments + the stdout pump + forwarded-request ids. */interface AgentState { /** Connections attached to this agent keyed by role (fan-out set, §6.2 + saga u4uo9m.2). * `interactive` conns answer permission/UI; `observer` conns only receive events. */ readonly attached: Map<ProxyConnection, AttachmentRole>; /** Connections reserved during runtime startup; they count as clients but receive no live events yet. */ readonly pendingAttach: Map<ProxyConnection, Envelope[]>; /** request_ids already forwarded to the agent (first-response-wins, §6.3). */ readonly forwarded: Set<string>; /** In-flight `permission_request`/`ui_request` records awaiting a response (njhows). */ readonly pending: Map<string, PendingRequest>; /** Timers for pending requests, cancelled on response, resolution, or generation exit. */ readonly pendingTimers: Map<string, ReturnType<typeof globalThis.setTimeout>>; /** Accepted prompts not yet completed; keeps queued work from looking idle. */ pendingTurns: number; /** Whether a stdout pump is bound for the current runtime generation. */ pumping: boolean; stream: AgentStream | undefined; ready: Promise<void> | undefined; resolveReady: (() => void) | undefined; rejectReady: ((error: Error) => void) | undefined; helloSeen: boolean; syncSeen: boolean; /** * The child runtime's validated `hello.capabilities` (saga vr7kgx.6), projected into the * daemon-authored `sync`. Empty until the generation's hello is observed; reset to empty * on every generation replacement so an old runtime's capabilities cannot leak. */ capabilities: string[]; /** The current generation's authoritative bootstrap mode from the child `sync`. Reset on * stream replacement so a previous generation's registry projection cannot leak into the * replacement sync while the new child boots with a different default/restored mode. */ runtimeMode: string | undefined; /** * The authoritative child-runtime generation epoch (saga vr7kgx.6 hardening). Increments * on every same-`agent_id` stream replacement in {@link Broker.ensurePump}. Carried in * every daemon-authored `sync` so the browser can fence catalogs/responses from a * superseded generation. The stdout-pump `data`/`end` closures capture the epoch they * were bound for and ignore events from any other epoch — the authoritative fence that * stops a superseded child stream from mutating/fanning-out current-generation state. */ runtimeEpoch: number; /** Whether the post-replacement re-sync has been emitted for the current generation. * Reset to `false` on every replacement; set `true` once the new generation reaches * hello+sync readiness and the daemon-authored `sync` has been fanned to attachments. */ replacementSyncEmitted: boolean;}
/** * The proxy/multiplexer. Construct with `getStream` (+ optional `getSync`); wire the * orchestrator handler via {@link Broker.setOrchestratorHandler}; then feed every inbound * command through {@link Broker.routeInbound} and call {@link Broker.attach}/ * {@link Broker.detach} from the orchestrator handler on `agent.attach`/`agent.detach`. * Call {@link Broker.onConnectionClosed} when a socket drops (cleanup attachment sets). */export class Broker { private readonly getStream: StreamProvider; private readonly ensureRuntime: RuntimeEnsurer | undefined; private readonly getSync: SyncProvider | undefined; private readonly parentOf: ParentResolver | undefined; private leaseGuard: SubagentLeaseGuard | undefined; private readonly onAgentStatus: AgentStatusListener | undefined; private readonly onAgentActivity: AgentActivityListener | undefined; private readonly onAgentOverviewActivity: AgentOverviewActivityListener | undefined; private readonly onAgentMode: AgentModeListener | undefined; private readonly onAgentModel: AgentModelListener | undefined; private readonly onAgentUsage: AgentUsageListener | undefined; private readonly now: () => number; private readonly headlessTimeoutMs: number; private readonly setTimer: (callback: () => void, delay: number) => ReturnType<typeof globalThis.setTimeout>; private readonly clearTimer: (timer: ReturnType<typeof globalThis.setTimeout>) => void; private readonly states = new Map<string, AgentState>(); /** Internal tagged event subscribers (saga u4uo9m.2) — NOT attachments, never responders. */ private readonly internalSubscribers = new Map<string, InternalEventSubscriber>(); /** Delegated pending controls routed to an ancestor (saga u4uo9m.2 item 3): * `request_id` → `{ child, ancestor }`. Cleared on resolution / generation exit. */ private readonly delegatedPending = new Map<string, { child: string; ancestor: string }>(); private orchestratorHandler: OrchestratorHandler | undefined;
constructor(opts: BrokerOptions) { this.getStream = opts.getStream; this.ensureRuntime = opts.ensureRuntime; this.getSync = opts.getSync; this.parentOf = opts.parentOf; this.leaseGuard = opts.leaseGuard; this.onAgentStatus = opts.onAgentStatus; this.onAgentActivity = opts.onAgentActivity; this.onAgentOverviewActivity = opts.onAgentOverviewActivity; this.onAgentMode = opts.onAgentMode; this.onAgentModel = opts.onAgentModel; this.onAgentUsage = opts.onAgentUsage; this.now = opts.now ?? (() => Date.now()); this.headlessTimeoutMs = opts.headlessTimeoutMs ?? 60_000; this.setTimer = opts.setTimeout ?? ((callback, delay) => globalThis.setTimeout(callback, delay)); this.clearTimer = opts.clearTimeout ?? ((timer) => globalThis.clearTimeout(timer)); }
/** Whether the daemon owns a live stdin/stdout stream for an agent. */ hasLiveStream(agentId: string): boolean { return this.getStream(agentId) !== undefined; }
/** Plug in the orchestrator command handler (tb4593.5). */ setOrchestratorHandler(handler: OrchestratorHandler): void { this.orchestratorHandler = handler; }
/** Plug in the inbound lease guard (saga u4uo9m.2). The `SubagentJobCoordinator` (.4) * installs this once it owns lease state; before then the default (always-allow) * preserves unleased behavior. */ setLeaseGuard(guard: SubagentLeaseGuard): void { this.leaseGuard = guard; }
/** * Inject a per-agent command DIRECTLY to a child's stdin, BYPASSING the inbound lease * guard + the first-response-wins duplicate check (saga u4uo9m.4). The `SubagentJobCoordinator` * is the sole legitimate driver of a leased Agent's prompt/steer/abort — it must not be * blocked by its own guard. The command is in ROUTING form (carries `agent_id`); it is * stripped to direct form before writing, exactly like {@link Broker.forwardToAgent}. * Restores an idle-stopped runtime for `prompt` (so a coordinator can drive a stopped * child); notifies `onAgentActivity` so idle cleanup is cancelled. Returns `true` on a * successful write or `false` when no live stream is available / the runtime failed to * restore (the caller surfaces a run failure; this method never throws). */ async inject(agentId: string, msg: Envelope): Promise<boolean> { const startsTurn = msg.type === 'prompt' && typeof msg.message === 'string' && msg.message.length > 0; let stream = this.getStream(agentId); if (this.ensureRuntime !== undefined && shouldRestoreRuntime(msg.type)) { try { await this.ensureRuntime(agentId); stream = this.getStream(agentId); } catch { return false; } } if (stream === undefined) return false; if (startsTurn) this.onAgentActivity?.(agentId); const state = startsTurn ? this.getOrCreate(agentId) : undefined; if (state !== undefined) state.pendingTurns += 1; const direct = omitRoutingAgentId(msg); const written = this.writeToStream(stream.stdin, direct); if (!written && state !== undefined) state.pendingTurns = Math.max(0, state.pendingTurns - 1); return written; }
/** * Deliver an unsolicited background result to an agent. Active parents receive a steer so * the result joins their current turn at the next provider boundary; idle/stopped parents * receive a prompt, which starts a fresh turn and therefore wakes them without polling. */ async deliverBackgroundMessage(agentId: string, message: string): Promise<boolean> { if (message.length === 0) return false; const busy = (this.states.get(agentId)?.pendingTurns ?? 0) > 0; if (busy) { const steered = await this.inject(agentId, { type: 'steer', agent_id: agentId, message, } as Envelope); if (steered) return true; } return this.inject(agentId, { type: 'prompt', agent_id: agentId, message, } as Envelope); }
/** * Subscribe to per-agent events through an INTERNAL tagged channel (saga u4uo9m.2). The * `SubagentJobCoordinator` (.4) uses this to bind `turn_id` at `turn_start` and clear it at * `turn_end`. It is NOT an attachment: it never counts as a permission/UI responder, never * receives `sync`, and never changes the headless-deny predicate. Events are delivered * synchronously in stdout-pump order, one at a time, BEFORE fan-out (so lease state is * fresh before any concurrent inbound control is inspected). Re-registering under the same * `tag` replaces the prior subscriber. Returns an unsubscribe function. */ subscribeInternal(tag: string, fn: InternalEventSubscriber): () => void { this.internalSubscribers.set(tag, fn); return () => { if (this.internalSubscribers.get(tag) === fn) this.internalSubscribers.delete(tag); }; }
/** * Route one inbound command from a daemon connection. Validates the catalog, then: * - orchestrator command → delegate to the orchestrator handler (§10; not routed). * - per-agent command → validate routing `agent_id` (§5.2), enforce * first-response-wins for `permission_response`/`ui_response`, strip `agent_id`, * write to the agent child's stdin. * Never throws — invalid/unroutable messages produce an `error` event (§9.4) sent back * to the client, and the stream continues (mirrors the stdio server's discipline). */ async routeInbound(cmd: Envelope, conn: ProxyConnection): Promise<void> { const checked = validateMessage(cmd); if (!checked.ok) { await this.sendError(conn, 'INVALID_REQUEST', checked.error, cmd); return; } const msg = checked.message as Envelope;
// Web administration is deliberately local-only, regardless of web authentication. if (msg.type.startsWith('web.') && (conn.auth?.kind ?? 'local') !== 'local') { await conn.send({ type: 'response', id: String(msg.id ?? ''), command: msg.type, success: false, error: { code: 'PERMISSION_DENIED', message: 'web administration requires a local connection' } }); return; }
// Orchestrator commands carry agent_id as PAYLOAD where applicable (agent.attach/detach/ // kill/status) — NOT a routing key. Dispatch to the handler BEFORE any routing-agent_id // validation so the payload field isn't mistaken for a routing key (§5.2 / conformance). if (!requiresAgentIdOnDaemon(msg.type) && isCommandType(msg.type)) { await this.delegateOrchestrator(msg, conn); return; }
// Per-agent command: validate routing agent_id (§5.2 daemon rule). const agentIdError = validateAgentId(msg, 'daemon', true); if (agentIdError !== null) { await this.sendError(conn, 'PROTOCOL_ERROR', agentIdError, msg); return; } const agentId = msg.agent_id as string;
// First-response-wins (§6.3): drop late permission/ui responses silently. if (isBroadcastResponse(msg.type) && this.isAlreadyForwarded(agentId, msg)) { // Late — discard silently (§6.3). Not an error; the first response already won. return; } // Structured permission choices are opaque IDs, but only IDs advertised by the // matching pending request may win. Do this before forwarding/marking the request; // an invalid first attempt must leave the request available to another responder. if (msg.type === 'permission_response' && !this.hasValidPermissionChoice(agentId, msg)) return;
// Inbound lease guard (saga u4uo9m.2 item 5): for a daemon-managed (leased) agent, // direct prompt is a CONFLICT (one run at a time); direct steer/abort traverse the // coordinator's strict active-turn / cancellation path; responses always pass. The // default guard (no coordinator) always allows — unleased behavior is unchanged. const guard = this.leaseGuard; if (guard !== undefined) { const decision = guard.inspect(agentId, msg); if (decision.action === 'conflict') { await this.sendError( conn, 'CONFLICT', decision.message ?? `agent "${agentId}" is leased by a daemon-managed run`, msg, ); return; } if (decision.action === 'delegate') { try { await guard.delegate(agentId, msg, conn); } catch (err) { const message = err instanceof Error ? err.message : String(err); await this.sendError(conn, 'INTERNAL_ERROR', message, msg); } return; } }
await this.forwardToAgent(agentId, msg, conn); }
/** * Attach a connection to an agent (§6.1). Adds `conn` to the agent's fan-out set and * emits a `sync` event to `conn` (§9.1 / §6.4). Returns the sync payload (so the * orchestrator handler can build the `agent.attach` response data) or `undefined` if no * sync provider is wired. Idempotent: re-attaching is a no-op (no duplicate sync). */ /** Start the stdout→fan-out pump for an agent at spawn time (ARCHITECTURE §6.2, * saga umckla). Without this, a headless agent (zero clients ever attached) would * never have its events processed — including `permission_request`, which the * proxy auto-denies at zero clients. Idempotent: a second call (e.g. on `attach`) * is a no-op once `pumping` is set. Safe to call before the stream exists — returns * silently; `attach` re-runs `ensurePump` when the stream appears. */ startPump(agentId: string): void { this.ensurePump(agentId, this.getOrCreate(agentId)); }
/** Resolve only after the pump has observed the generation's hello and sync. */ waitForReady(agentId: string): Promise<void> { const state = this.getOrCreate(agentId); if (state.helloSeen && state.syncSeen) return Promise.resolve(); if (state.ready === undefined) { state.ready = new Promise<void>((resolve, reject) => { state.resolveReady = resolve; state.rejectReady = reject; }); } this.ensurePump(agentId, state); return state.ready; }
/** Reserve an attachment before RuntimeManager startup; commit after readiness. */ reserveAttach(conn: ProxyConnection, agentId: string): () => void { const state = this.getOrCreate(agentId); if (state.attached.has(conn)) return () => {}; const buffer: Envelope[] = []; state.pendingAttach.set(conn, buffer); this.ensurePump(agentId, state); return () => { // Do not cancel a later reservation for the same connection. if (state.pendingAttach.get(conn) === buffer) { state.pendingAttach.delete(conn); this.reevaluatePendingFallbacks(); } }; }
async attach( conn: ProxyConnection, agentId: string, role: AttachmentRole = 'interactive', ): Promise<SyncPayload | undefined> { const state = this.getOrCreate(agentId); const prior = state.attached.get(conn); if (prior !== undefined) { // Idempotent re-attach: update the role in place (a client may promote/demote itself) // but do NOT re-emit sync (§6.1). A no-op when the role is unchanged. if (prior !== role) state.attached.set(conn, role); return this.getSync?.(agentId); } // Keep the reservation in place until sync and every event buffered during the // asynchronous sends have been delivered. This prevents a live event from overtaking // authoritative sync, and keeps a reserved client eligible for permission prompts. if (!state.pendingAttach.has(conn)) state.pendingAttach.set(conn, []); this.ensurePump(agentId, state); const sync = this.getSync?.(agentId); if (sync !== undefined) { await this.sendTo(conn, { type: 'sync', agent_id: agentId, status: sync.status, mode: state.runtimeMode ?? sync.mode, capabilities: state.capabilities, runtime_epoch: state.runtimeEpoch, live_window: sync.live_window, pending_requests: sync.pending_requests }); } while (true) { const buffered = state.pendingAttach.get(conn); if (buffered === undefined || buffered.length === 0) break; const batch = buffered.splice(0, buffered.length); for (const event of batch) await this.sendTo(conn, event); } // A startup cancellation or connection close may have removed this reservation while // sync was being sent. Never promote a cancelled reservation. if (!state.pendingAttach.has(conn) || conn.closed) return sync; state.pendingAttach.delete(conn); state.attached.set(conn, role); this.reevaluatePendingFallbacks(); return sync; }
/** Detach a connection from an agent (§6.1). No-op if not attached. */ detach(conn: ProxyConnection, agentId: string): boolean { const state = this.states.get(agentId); if (state === undefined) return false; const detached = state.attached.delete(conn); if (detached) this.reevaluatePendingFallbacks(); return detached; }
/** Tear down all of a connection's attachments (call on socket close). */ onConnectionClosed(conn: ProxyConnection): void { for (const state of this.states.values()) { state.attached.delete(conn); state.pendingAttach.delete(conn); } this.reevaluatePendingFallbacks(); // (forwarded request_ids stay — they're keyed by agent, not conn; the resolved event // from the agent still fans out, and the set is cleared on resolved.) }
/** * Drop all broker state for an agent (call from the supervisor's onExit — tb4593.5 also * subscribes to emit `agent_exited`). Attached clients simply stop receiving events; * their next `agent.attach` to a different agent works normally. */ forgetAgent(agentId: string): void { // Generation exit / termination (saga u4uo9m.2 item 3): drop delegated-pending records // for which this agent is the child OR the ancestor so they can't leak into a future // generation or strand a reconnecting ancestor's sync. for (const [rid, info] of this.delegatedPending) { if (info.child === agentId || info.ancestor === agentId) this.delegatedPending.delete(rid); } const state = this.states.get(agentId); if (state !== undefined) this.clearPendingTimers(state); this.states.delete(agentId); }
/** The set of attached connections for an agent (test/diagnostic introspection). */ attachedConnections(agentId: string): readonly ProxyConnection[] { const state = this.states.get(agentId); return state === undefined ? [] : [...state.attached.keys()]; }
/** Connections attached as `interactive` to an agent — the permission/UI responder set * (saga u4uo9m.2). Observers are excluded. */ interactiveConnections(agentId: string): readonly ProxyConnection[] { const state = this.states.get(agentId); if (state === undefined) return []; const out: ProxyConnection[] = []; for (const [conn, role] of state.attached) if (role === 'interactive') out.push(conn); return out; }
/** The role of an attached connection, or `undefined` if not attached (saga u4uo9m.2). */ attachmentRole(conn: ProxyConnection, agentId: string): AttachmentRole | undefined { return this.states.get(agentId)?.attached.get(conn); }
/** * In-flight `permission_request`/`ui_request` records for an agent awaiting a response * (saga njhows). Drives `sync.pending_requests` on attach so a reconnecting client can * re-surface an outstanding prompt mid-turn (§6.4). Returns a snapshot copy. */ getPending(agentId: string): PendingRequest[] { const state = this.states.get(agentId); return state === undefined ? [] : [...state.pending.values()]; }
/** * Delegated pending controls routed FROM a child TO this ancestor for sync/reconnect * (saga u4uo9m.2 item 3). Returns the child pending-request records that were routed up * to `ancestorAgentId` because the child had no interactive responder of its own. Used by * the daemon's sync composer so an ancestor's reconnecting client re-surfaces a delegated * child prompt. Cleared on resolution / generation exit / termination. */ delegatedPendingFor(ancestorAgentId: string): PendingRequest[] { // The delegated set is small (one per in-flight child prompt); walk the child pending // maps and keep the records whose delegation target is this ancestor. const out: PendingRequest[] = []; for (const [rid, { child, ancestor }] of this.delegatedPending) { if (ancestor !== ancestorAgentId) continue; const req = this.states.get(child)?.pending.get(rid); if (req !== undefined) out.push(req); } return out; }
// ─── Internal: routing ──────────────────────────────────────────────────────────
/** Forward a per-agent command to the agent child's stdin (stripped of routing agent_id). */ private async forwardToAgent( agentId: string, msg: Envelope, conn: ProxyConnection, ): Promise<void> { const startsTurn = msg.type === 'prompt' && typeof msg.message === 'string' && msg.message.length > 0; let stream = this.getStream(agentId); if (this.ensureRuntime !== undefined && shouldRestoreRuntime(msg.type)) { try { // Restoring commands always pass through the lifecycle coordinator, even when a // stream still exists: an idle timeout may have signaled it without reaping it yet. await this.ensureRuntime(agentId); stream = this.getStream(agentId); } catch (error) { const message = error instanceof Error ? error.message : String(error); this.onAgentStatus?.(agentId, 'error', message); await this.sendError( conn, 'INTERNAL_ERROR', `agent "${agentId}" runtime startup failed: ${message}`, msg, ); return; } } if (stream === undefined) { await this.sendError( conn, 'NOT_FOUND', `agent "${agentId}" has no live stream (not running)`, msg, ); return; } if (startsTurn) this.onAgentActivity?.(agentId); if (isBroadcastResponse(msg.type)) { // A real response wins immediately, including while the child is emitting its resolved // event. Cancel the fallback now so it cannot race that response. if (typeof msg.request_id === 'string') this.cancelPendingTimer(agentId, msg.request_id); // Record the request_id as forwarded so late responses are dropped (§6.3). // Choice validation has already run in routeInbound, so an invalid choice can // never consume first-response-wins. const rid = msg.request_id; if (typeof rid === 'string') this.getOrCreate(agentId).forwarded.add(rid); } if (startsTurn) this.getOrCreate(agentId).pendingTurns += 1; const direct = omitRoutingAgentId(msg); // strip routing key → direct-conn form this.writeToStream(stream.stdin, direct); }
/** Validate a structured permission response against the advertised pending choices. */ private hasValidPermissionChoice(agentId: string, msg: Envelope): boolean { const choiceId = (msg as { choice_id?: unknown }).choice_id; if (choiceId === undefined) return true; // legacy decision form if (typeof choiceId !== 'string' || choiceId.length === 0) return false; const requestId = msg.request_id; if (typeof requestId !== 'string') return false; const pending = this.states.get(agentId)?.pending.get(requestId); if (pending?.type !== 'permission_request') return false; const choices = (pending as { choices?: Array<{ id?: unknown }> }).choices; return Array.isArray(choices) && choices.some((choice) => choice.id === choiceId); }
/** Whether a permission/ui response's request_id has already been forwarded (§6.3). */ private isAlreadyForwarded(agentId: string, msg: Envelope): boolean { const rid = msg.request_id; if (typeof rid !== 'string') return false; return this.getOrCreate(agentId).forwarded.has(rid); }
/** Delegate an orchestrator command to the registered handler (tb4593.5). */ private async delegateOrchestrator( msg: Envelope, conn: ProxyConnection, ): Promise<void> { if (this.orchestratorHandler === undefined) { // No handler wired (pre-.5). Surface a NOT_FOUND error so the client isn't left // hanging; the daemon wiring (.6) always plugs .5 in before serving. Uses // NOT_FOUND (§8 has no NOT_IMPLEMENTED code; mirrors the stdio server's // "session_op X not implemented" → NOT_FOUND precedent). await this.sendError( conn, 'NOT_FOUND', `orchestrator command "${msg.type}" has no handler`, msg, ); return; } try { await this.orchestratorHandler(msg, conn); } catch (err) { // A handler throw is a bug — surface INTERNAL_ERROR, keep the broker up. const message = err instanceof Error ? err.message : String(err); await this.sendError(conn, 'INTERNAL_ERROR', message, msg); } }
// ─── Internal: agent stdout pump (fan-out + first-response tracking) ────────────
/** Lazily start the stdout → fan-out pump for an agent (once per agent). */ private ensurePump(agentId: string, state: AgentState): void { const stream = this.getStream(agentId); if (stream === undefined) return; // no stream yet; starts when one exists if (state.pumping && sameStream(state.stream, stream)) return; const streamReplaced = state.stream !== undefined && !sameStream(state.stream, stream); if (streamReplaced) { // Same-agent runtime replacement (saga vr7kgx.6 hardening): a brand-new stream under // the same `agent_id` supersedes the prior generation, even when the old stream ended // before the replacement was installed (`state.pumping` is then already false). Reset // all generation-scoped // capability/request state and bump the authoritative epoch so (a) the old stream's // `data`/`end` closures — which capture the prior epoch — stop mutating/fanning-out // current-generation state, and (b) the daemon can include a fresh epoch in the // post-readiness re-sync so the browser fences stale catalogs/responses. this.clearPendingTimers(state); state.forwarded.clear(); state.pending.clear(); state.runtimeEpoch += 1; state.replacementSyncEmitted = false; // New runtime generation: drop delegated-pending records for this child so a stale // ancestor sync can't re-surface a prompt from a prior generation (saga u4uo9m.2). for (const [rid, info] of this.delegatedPending) { if (info.child === agentId) this.delegatedPending.delete(rid); } } state.pendingTurns = 0; state.pumping = true; state.stream = stream; state.capabilities = []; state.runtimeMode = undefined; const decoder = new JsonlDecoder(); state.helloSeen = false; state.syncSeen = false; state.ready = undefined; state.resolveReady = undefined; state.rejectReady = undefined; // Authoritative generation fence: a `data`/`end` callback bound for `epoch` is ignored // the moment the epoch advances (replacement) OR the stream/state identity changes. // Belt-and-suspenders with `sameStream`: the epoch is the authoritative token the // browser also consumes, so the runtime and the wire agree on a single fence. const epoch = state.runtimeEpoch; const current = () => state.runtimeEpoch === epoch && sameStream(this.getStream(agentId), stream) && this.states.get(agentId) === state; stream.stdout.on('data', (chunk: Buffer) => { if (!current()) return; decoder.write(chunk); for (const record of decoder.parse()) this.onAgentEvent(agentId, record); }); stream.stdout.on('end', () => { if (current() && !(state.helloSeen && state.syncSeen)) state.rejectReady?.(new Error(`agent "${agentId}" exited before readiness`)); if (!current()) return; decoder.end(); for (const record of decoder.parse()) this.onAgentEvent(agentId, record); state.pumping = false; }); stream.stdout.on('error', (err: Error) => { if (current()) process.stderr.write(`shardd: agent ${agentId} stdout: ${err.message}\n`); }); }
/** One parsed record from the agent child's stdout — tag agent_id + fan out / track. */ private async onAgentEvent(agentId: string, record: JsonlRecord): Promise<void> { if (record.value === undefined) { // Malformed JSON from the agent — log; the agent's own stderr is the real signal. process.stderr.write( `shardd: agent ${agentId} sent malformed JSON: ${record.parseError ?? 'parse error'}\n`, ); return; } const event = record.value as Envelope; if (event.type === 'hello') { stateReady(this.states.get(agentId), 'hello'); // saga vr7kgx.6: capture the child runtime's validated capabilities for projection // into the daemon-authored `sync`. HelloSchema (§9.1) guarantees `capabilities` is a // string[]; a malformed hello never reaches here (validateMessage ran in the child's // stdio server before it was written, and the daemon re-validates inbound records). // Stored BEFORE the readiness-early-return below so a `sync` requested during the // hello→sync bootstrap window still carries it. const caps = (event as { capabilities?: unknown }).capabilities; const state = this.states.get(agentId); if (state !== undefined && Array.isArray(caps)) { state.capabilities = caps.filter((c): c is string => typeof c === 'string'); } } if (event.type === 'sync') { const state = this.states.get(agentId); const mode = (event as { mode?: unknown }).mode; if (state !== undefined && typeof mode === 'string' && mode !== '') { state.runtimeMode = mode; // Keep the durable reconnect/bootstrap projection aligned with the child generation // before composing the daemon-authored replacement sync. this.onAgentMode?.(agentId, mode); } stateReady(state, 'sync'); // saga vr7kgx.6 hardening: once a REPLACEMENT generation reaches hello+sync readiness, // emit a fresh daemon-authored `sync` (carrying the new epoch + capabilities) to every // existing attachment so the browser can fence stale catalogs/responses from the // superseded generation. The initial generation (epoch 0) is owned by `attach`; only // replacements re-sync existing clients here. Initiated synchronously (before the // readiness early-return below) so it lands ahead of the new generation's first live // event in per-connection send order. void this.maybeEmitReplacementSync(agentId); } // Readiness records are consumed by the generation coordinator. Attach emits the // daemon-authoritative sync exactly once; never replay bootstrap records as live events. if (event.type === 'hello' || event.type === 'sync') return; // §6.3 note: a resolved event fans out to all clients (below). We do NOT clear the // forwarded id here — a late `permission_response` arriving AFTER resolved must still // be dropped (the request is done; forwarding a second answer would re-trigger the // agent's gate). The forwarded set accumulates request_ids for the agent's lifetime; // v0.1 accepts the growth (bounded by per-session request count; ids are UUIDs, never // reused). A cap+eviction can land if it ever matters. // Tag routing agent_id (the child's direct-conn form omits it; §5.2 daemon fan-out). const tagged: Envelope = { ...event, agent_id: agentId }; // §10.2 mirror: also bridge per-agent `status` events to the orchestrator so the // registry + `agent_status_changed` broadcast stay in sync (tb4593.6 wiring). The // tagged per-agent `status` event below still fans out to attached clients (§6.2) — // this is the additional registry-level mirror, fired BEFORE fan-out so order is // registry-update-then-broadcast. let overviewPhase: Parameters<AgentOverviewActivityListener>[1] = 'idle'; let overviewTool: string | undefined; if (event.type === 'status') { const status = (event as { status?: AgentStatus }).status; const detail = (event as { detail?: string }).detail; if (typeof status === 'string') this.onAgentStatus?.(agentId, status, detail); if (status !== undefined) { overviewPhase = status === 'compacting' ? 'compacting' : status === 'streaming' ? 'streaming' : status === 'error' ? 'error' : status.startsWith('waiting_') ? 'waiting' : 'idle'; } } else { const inferred = this.inferStatus(agentId, event); if (inferred !== undefined) this.onAgentStatus?.(agentId, inferred); const activityPhase = this.activityPhase(event); if (activityPhase !== undefined) overviewPhase = activityPhase; if (event.type === 'tool_call' && typeof event.name === 'string' && isSafeToolName(event.name)) overviewTool = event.name; } if (this.onAgentOverviewActivity !== undefined && (event.type === 'status' || this.activityPhase(event) !== undefined)) { this.onAgentOverviewActivity(agentId, overviewPhase, overviewTool); } if (event.type === 'response' && event.command === 'set_mode' && event.success === true) { const mode = (event.data as { mode?: unknown } | undefined)?.mode; if (typeof mode === 'string' && mode !== '') this.onAgentMode?.(agentId, mode); } // saga vr7kgx.6: mirror a successful model change into daemon runtime metadata from // the child's `model_changed` broadcast (only real changes broadcast — an idempotent // `set_model` is an ack with no event). The active descriptor's id/provider/model update // the registry projection without exposing secrets or changing private routing. if (event.type === 'model_changed') { const active = (event as { active?: unknown }).active; const capabilities = (event as { capabilities?: unknown }).capabilities; if ( active !== null && typeof active === 'object' && typeof (active as { id?: unknown }).id === 'string' && typeof (active as { provider?: unknown }).provider === 'string' && typeof (active as { model?: unknown }).model === 'string' ) { this.onAgentModel?.(agentId, active as { id: string; provider: string; model: string }); } // Model-specific capabilities change with the active selection. Persist the child's // post-switch snapshot so a later/reopened attachment receives the same capabilities // as clients that observed this live broadcast. Older runtimes omit the field. const state = this.states.get(agentId); if (state !== undefined && Array.isArray(capabilities) && capabilities.every((cap) => typeof cap === 'string')) { state.capabilities = [...capabilities]; } } // saga vr7kgx.7: mirror a per-agent `usage` event into durable runtime state so the // session runtime projection (status / context size) carries the last-known accounting. // Sanitized to the secret-free `Usage` shape (numbers + optional model name); a malformed // payload (missing input/output) is dropped — the bridge never throws. Mirrors the // status/model bridges: fired BEFORE fan-out so the registry projection is fresh, and the // `usage` event still fans out to attached clients below (folded into the transcript). if (event.type === 'usage') { const u = event as { input?: unknown; output?: unknown; cache_read?: unknown; cache_write?: unknown; cost?: unknown; context_tokens?: unknown; model?: unknown; provider?: unknown }; const input = u.input; const output = u.output; if (typeof input === 'number' && typeof output === 'number') { const usage: { input: number; output: number; cache_read?: number; cache_write?: number; cost?: number; context_tokens?: number; model?: string; provider?: string } = { input, output }; if (typeof u.cache_read === 'number') usage.cache_read = u.cache_read; if (typeof u.cache_write === 'number') usage.cache_write = u.cache_write; if (typeof u.cost === 'number') usage.cost = u.cost; if (typeof u.context_tokens === 'number') usage.context_tokens = u.context_tokens; if (typeof u.model === 'string' && u.model !== '') usage.model = u.model; if (typeof u.provider === 'string' && u.provider !== '') usage.provider = u.provider; this.onAgentUsage?.(agentId, usage); } } // Track in-flight permission/ui requests so a client that attaches MID-turn gets them // in `sync.pending_requests` (§6.4, saga njhows). Added on request, removed on the // matching *_resolved (the request is answered — first-response-wins already applied). this.trackPending(agentId, tagged); // Internal tagged event subscription (saga u4uo9m.2 item 4): deliver to the // SubagentJobCoordinator synchronously, in pump order, BEFORE fan-out / headless // synthesis, so the coordinator's lease state (turn_start binding / turn_end clearing) // is fresh before any concurrent inbound control is inspected. Internal subscribers // are NOT attachments and never count as permission/UI responders. for (const fn of this.internalSubscribers.values()) { try { fn(agentId, tagged); } catch (err) { // A subscriber throw is a coordinator bug — log and keep the broker up. The broker // never lets an internal subscriber break fan-out / headless synthesis. process.stderr.write(`shardd: internal subscriber for ${agentId} threw: ${err instanceof Error ? err.message : String(err)}\n`); } } // Hierarchical permission/UI routing + headless fallback (ARCHITECTURE §6.2, // saga umckla + u4uo9m.2 items 2/7). A delegated child's request goes to the child's // own interactive clients first; if the child has NONE, it is routed up to the // nearest interactive ancestor (retaining the child agent_id so a response routes // back to the child). Only when there is no interactive responder ANYWHERE in the // hierarchy does the proxy synthesize a bottom-case answer so no request hangs: // - permission_request → permission_response{source:'headless', decision:'deny'} // (the gate maps it to a terminal deny; umckla parity for root/no-parent agents); // - ui_request → ui_response{value:null} (the stdio bridge turns it into a // ui_resolved{value:null} so a future UI gate never hangs). if (event.type === 'permission_request' || event.type === 'ui_request') { const rid = (event as { request_id?: string }).request_id; const childHasResponder = this.hasInteractiveResponder(agentId); if (!childHasResponder) { const ancestor = this.nearestInteractiveAncestor(agentId); if (ancestor !== undefined && typeof rid === 'string') { // Delegate up the hierarchy: fan the request (still tagged with the CHILD // agent_id) to the ancestor's interactive responders and record the delegation // so an ancestor reconnect can re-surface it (saga u4uo9m.2 item 3). A // response carries the child agent_id as the routing key, so it routes back to // the child stdin through the normal per-agent path. this.delegatedPending.set(rid, { child: agentId, ancestor }); await this.fanoutToInteractive(ancestor, tagged); } else if (typeof rid === 'string') { // No responder currently exists. Keep the request pending: a client may attach // before the safety timeout and answer it (including after a client swap). this.schedulePendingFallback(agentId, rid, event.type); } } } await this.fanout(agentId, tagged); // Clear a delegated-pending record on resolution (saga u4uo9m.2 item 3): the request // is answered, so an ancestor reconnect must not re-surface a stale prompt. if (event.type === 'permission_resolved' || event.type === 'ui_resolved') { const rid = (event as { request_id?: string }).request_id; if (typeof rid === 'string') this.delegatedPending.delete(rid); } }
private activityPhase(event: Envelope): 'idle' | 'thinking' | 'streaming' | 'tool' | 'waiting' | 'compacting' | 'retrying' | 'error' | undefined { switch (event.type) { case 'turn_start': case 'message_complete': case 'tool_result': return 'thinking'; case 'message_delta': return 'streaming'; case 'tool_call': return 'tool'; case 'permission_request': case 'ui_request': return 'waiting'; case 'compaction': return 'compacting'; case 'provider_retry': return 'retrying'; case 'turn_end': return 'idle'; case 'error': return event.fatal === true || event.code === 'aborted' ? 'error' : undefined; default: return undefined; } }
private inferStatus(agentId: string, event: Envelope): AgentStatus | undefined { const state = this.getOrCreate(agentId); switch (event.type) { case 'turn_start': // Most turns were pre-counted when a prompt was forwarded. An idle `steer` may also // start a fresh turn inside the Runtime, so observe that case here and keep background // delivery in steer mode until its matching terminal event. state.pendingTurns = Math.max(1, state.pendingTurns); return 'streaming'; case 'permission_resolved': case 'ui_resolved': return 'streaming'; case 'permission_request': return 'waiting_for_permission'; case 'ui_request': return 'waiting_for_ui'; case 'turn_end': state.pendingTurns = Math.max(0, state.pendingTurns - 1); return state.pendingTurns > 0 ? 'streaming' : 'idle'; case 'error': { const code = (event as { code?: unknown }).code; const fatal = (event as { fatal?: unknown }).fatal; if (fatal !== true && code !== 'aborted') return undefined; state.pendingTurns = Math.max(0, state.pendingTurns - 1); return state.pendingTurns > 0 ? 'streaming' : 'error'; } default: return undefined; } }
/** Add a `permission_request`/`ui_request` to the agent's pending set, or remove it on * the matching `permission_resolved`/`ui_resolved`. No-op for other event types. */ private trackPending(agentId: string, event: Envelope): void { const state = this.states.get(agentId); if (state === undefined) return; // no per-agent state yet — nothing to track against const rid = (event as { request_id?: string }).request_id; if (typeof rid !== 'string') return; if (event.type === 'permission_request' || event.type === 'ui_request') { // Pending tracking is universal; the timer is deliberately scheduled only after routing. state.pending.set(rid, event as unknown as PendingRequest); } else if (event.type === 'permission_resolved' || event.type === 'ui_resolved') { state.pending.delete(rid); this.cancelPendingTimer(agentId, rid); } }
/** Send an event to every attached connection for an agent (§6.2 fan-out). Both roles * receive the event stream; only `interactive` conns answer prompts (saga u4uo9m.2). */ private async fanout(agentId: string, event: Envelope): Promise<void> { const state = this.states.get(agentId); if (state === undefined) return; for (const events of state.pendingAttach.values()) events.push(event); const conns = [...state.attached.keys()].filter((conn) => !state.pendingAttach.has(conn)); await Promise.all(conns.map((conn) => conn.closed ? Promise.resolve() : this.sendTo(conn, event).catch(() => {}))); }
/** * Emit a fresh daemon-authored `sync` to existing attachments after a same-`agent_id` * runtime replacement reaches hello+sync readiness (saga vr7kgx.6 hardening). The re-sync * carries the new `runtime_epoch` + projected `capabilities` so a browser can fence stale * catalogs/responses emitted by the superseded generation. Only fully-attached clients * receive it: a pending reservation is still buffering and gets its own authoritative * `sync` (with the fresh epoch) when its `attach` commits, preserving attach ordering / * buffering guarantees. Idempotent per generation (`replacementSyncEmitted`); a no-op for * the initial generation (epoch 0 — `attach` owns its sync) and before readiness. */ private async maybeEmitReplacementSync(agentId: string): Promise<void> { const state = this.states.get(agentId); if (state === undefined) return; if (state.runtimeEpoch === 0) return; // initial generation: attach owns the sync if (state.replacementSyncEmitted) return; if (!(state.helloSeen && state.syncSeen)) return; // not ready yet state.replacementSyncEmitted = true; const sync = this.getSync?.(agentId); if (sync === undefined) return; // no sync provider — nothing authoritative to re-emit const payload: Envelope = { type: 'sync', agent_id: agentId, status: sync.status, mode: state.runtimeMode ?? sync.mode, capabilities: state.capabilities, runtime_epoch: state.runtimeEpoch, live_window: sync.live_window, pending_requests: sync.pending_requests, }; // Fan to fully-attached connections only. Pending reservations continue buffering // live events and receive their own daemon-authored sync (with the fresh epoch) when // their `attach` commits — so the re-sync never overtakes a reservation's initial sync. const conns = [...state.attached.keys()].filter( (conn) => !state.pendingAttach.has(conn) && !conn.closed, ); await Promise.all(conns.map((conn) => this.sendTo(conn, payload).catch(() => {}))); }
/** * Whether an agent has any `interactive` responder that can answer a permission/UI * request (saga u4uo9m.2). Reserved (pending-attach) connections count as interactive * (they are real clients mid-attach); `observer` attachments do not. */ private hasInteractiveResponder(agentId: string): boolean { const state = this.states.get(agentId); if (state === undefined) return false; if (state.pendingAttach.size > 0) return true; for (const role of state.attached.values()) if (role === 'interactive') return true; return false; }
/** * Walk the durable parent chain to the nearest ancestor with an `interactive` responder * (saga u4uo9m.2 item 2). Returns the ancestor agent_id, or `undefined` for a root / * unknown parent / no interactive ancestor. Cycle-safe: a visited set + depth cap (64) * prevent a corrupt hierarchy from looping; a detected cycle logs a warning and yields * `undefined` (the broker never throws on bad durable state). */ private nearestInteractiveAncestor(agentId: string): string | undefined { if (this.parentOf === undefined) return undefined; const visited = new Set<string>([agentId]); let current = agentId; for (let depth = 0; depth < 64; depth++) { const parent = this.parentOf(current); if (parent === undefined || parent === '') return undefined; if (visited.has(parent)) { // Corrupt hierarchy cycle — fail loudly to stderr but keep serving. process.stderr.write(`shardd: parent-of cycle detected at ${parent} (walking from ${agentId})\n`); return undefined; } visited.add(parent); if (this.hasInteractiveResponder(parent)) return parent; current = parent; } return undefined; }
/** Fan an event out to an agent's `interactive` responders only (saga u4uo9m.2). Used to * route a delegated child's permission/ui request up to the nearest interactive * ancestor without also waking its observers. The event retains the CHILD agent_id. */ private async fanoutToInteractive(agentId: string, event: Envelope): Promise<void> { const state = this.states.get(agentId); if (state === undefined) return; const conns: ProxyConnection[] = []; for (const [conn, role] of state.attached) { if (role === 'interactive' && !state.pendingAttach.has(conn) && !conn.closed) conns.push(conn); } await Promise.all(conns.map((conn) => this.sendTo(conn, event).catch(() => {}))); }
/** Synthesize the bottom-case response after the headless safety timeout. Marks the id * forwarded so a late client response is dropped by first-response-wins. */ private synthesizeHeadless( agentId: string, rid: string, requestType: 'permission_request' | 'ui_request', stream: AgentStream, ): void { const state = this.getOrCreate(agentId); this.cancelPendingTimer(agentId, rid); if (state.forwarded.has(rid)) return; state.forwarded.add(rid); if (requestType === 'permission_request') { this.writeToStream(stream.stdin, { type: 'permission_response', request_id: rid, decision: 'deny', source: 'headless', } as Envelope); } else { // ui_request → null value. The child's stdio bridge converts the ui_response into a // ui_resolved{value:null} event so a future UI gate resolves (never hangs). this.writeToStream(stream.stdin, { type: 'ui_response', request_id: rid, value: null, source: 'headless', } as Envelope); } }
private schedulePendingFallback( agentId: string, rid: string, requestType: 'permission_request' | 'ui_request', ): void { const state = this.states.get(agentId); if (state === undefined || state.pendingTimers.has(rid)) return; const timer = this.setTimer(() => { state.pendingTimers.delete(rid); if (this.states.get(agentId) !== state || !state.pending.has(rid) || state.forwarded.has(rid)) return; const stream = this.getStream(agentId); if (stream === undefined) { // The generation disappeared before the callback fired. Do not wake it merely to // synthesize a response, and do not retain a permanently stale pending record. state.pending.delete(rid); return; } this.synthesizeHeadless(agentId, rid, requestType, stream); }, this.headlessTimeoutMs); state.pendingTimers.set(rid, timer); }
/** Schedule fallback for pending requests that still have no eligible responder. */ private reevaluatePendingFallbacks(): void { for (const [agentId, state] of this.states) { const hasResponder = this.hasInteractiveResponder(agentId) || this.nearestInteractiveAncestor(agentId) !== undefined; if (hasResponder) { for (const rid of state.pendingTimers.keys()) this.cancelPendingTimer(agentId, rid); continue; } for (const [rid, request] of state.pending) { if (!state.forwarded.has(rid) && !state.pendingTimers.has(rid)) { this.schedulePendingFallback(agentId, rid, request.type); } } } }
private cancelPendingTimer(agentId: string, rid: string): void { const state = this.states.get(agentId); const timer = state?.pendingTimers.get(rid); if (timer === undefined) return; this.clearTimer(timer); state?.pendingTimers.delete(rid); }
private clearPendingTimers(state: AgentState): void { for (const timer of state.pendingTimers.values()) this.clearTimer(timer); state.pendingTimers.clear(); }
// ─── Internal: send helpers ─────────────────────────────────────────────────────
/** Encode + send a message to one connection; swallows send failures on a dead conn. */ private async sendTo(conn: ProxyConnection, message: Envelope): Promise<void> { if (conn.closed) return; try { await conn.send(message); } catch { // The socket died mid-send; onConnectionClosed will drop it from attachment sets. } }
/** * Surface a recoverable `error` event to the client (§9.4 — fatal:false). Used for * routing/validation failures that don't correlate to a command id (events carry no id). */ private async sendError( conn: ProxyConnection, code: ErrorCode, message: string, _orig: Envelope, ): Promise<void> { await this.sendTo(conn, { type: 'error', code, message, fatal: false, }); }
/** Write a JSONL-encoded message to a writable stream; false means the stream was closed. */ private writeToStream(stdin: NodeJS.WritableStream, msg: Envelope): boolean { const writable = stdin as NodeJS.WritableStream & { destroyed?: boolean; writable?: boolean; writableEnded?: boolean; }; if (writable.destroyed === true || writable.writable === false || writable.writableEnded === true) { return false; } try { writable.write(encodeJsonl(msg)); return true; } catch { // The child's stdin closed (agent dying). Its exit will tear down the pump; callers // that need reliable wake-up may restore the runtime and retry as a prompt. return false; } }
// ─── Internal: state ────────────────────────────────────────────────────────────
/** Get or create the per-agent state (attachment set + forwarded ids + pump flag). */ private getOrCreate(agentId: string): AgentState { let state = this.states.get(agentId); if (state === undefined) { const created: AgentState = { attached: new Map(), pendingAttach: new Map(), forwarded: new Set(), pending: new Map(), pendingTimers: new Map(), pendingTurns: 0, pumping: false, stream: undefined, ready: undefined, resolveReady: undefined, rejectReady: undefined, helloSeen: false, syncSeen: false, capabilities: [], runtimeMode: undefined, runtimeEpoch: 0, replacementSyncEmitted: false, }; this.states.set(agentId, created); return created; } return state; }}
// ─── Pure predicates ──────────────────────────────────────────────────────────────
/** Whether a `type` is any command (client→server) — distinguishes orchestrator commands * (which `requiresAgentIdOnDaemon` reports as false) from orchestrator EVENTS (also false * but NOT commands). Used to route orchestrator commands to the handler. */function isCommandType(type: string): boolean { return ( type === 'prompt' || type === 'steer' || type === 'abort' || type === 'set_mode' || type === 'list_modes' || type === 'list_models' || type === 'set_model' || type === 'list_reasoning_levels' || type === 'set_reasoning_level' || type === 'session_op' || type === 'permission_response' || type === 'ui_response' || type === 'agent.list' || type === 'agent.create' || type === 'agent.attach' || type === 'agent.detach' || type === 'agent.terminate' || type === 'agent.kill' || type === 'agent.status' || type === 'daemon.shutdown' || type === 'session.create' || type === 'session.status' || type === 'session.terminate' || type === 'session.kill' || type === 'session.activate' || type === 'session.list' || type === 'session.replay' || type === 'stats.tool_usage' || type === 'stats.token_usage' || type === 'web.approve' || type === 'web.token.create' || type === 'web.clients' || type === 'web.revoke' || type === 'web.token.rotate' || // Workspace lifecycle commands (saga j8dtju.6, §10.1 additive). These are orchestrator // commands — they name no routing `agent_id` (the daemon resolves `cwd` / // `workspace_id` and generates the Agent + Session itself). The catalog's // `requiresAgentIdOnDaemon` already reports `false` for them, and the protocol schemas // (`Workspace*CommandSchema`, `additionalProperties: false`) forbid a client from // injecting one. Without these entries the broker silently fell through to the // per-agent path and PROTOCOL_ERROR'd a valid `workspace.create` (it demanded a // routing `agent_id` the schema never permits) — the same class of regression the // u4uo9m.7 real-socket E2E caught for `subagent.*`. Listing them here routes them to // `delegateOrchestrator` like `agent.create` / `session.create`. type === 'workspace.create' || type === 'workspace.attach' || type === 'workspace.list' || type === 'workspace.status' || type === 'workspace.retire' || // Subagent job orchestrator commands (saga u4uo9m.1). These carry `parent_agent_id` as // a payload field (the owning durable parent), NOT a routing `agent_id`, so they must // be dispatched to the orchestrator BEFORE the routing-agent_id validation (§5.2) — the // same path as `agent.create`/`agent.list`. Discovered by the u4uo9m.7 real-socket E2E: // without these entries the broker silently dropped `subagent.*` commands (no handler, // no routing agent_id) instead of delegating to `Orchestrator.onSubagent`. type === 'subagent.create' || type === 'subagent.wait' || type === 'subagent.steer' || type === 'subagent.relaunch' || type === 'subagent.status' || type === 'subagent.stop' );}
/** Whether a type is a first-response-wins response command (§6.3). */function sameStream(left: AgentStream | undefined, right: AgentStream): boolean { return left !== undefined && left.stdin === right.stdin && left.stdout === right.stdout;}
function isBroadcastResponse(type: string): boolean { return type === 'permission_response' || type === 'ui_response';}
function isSafeToolName(name: string): boolean { return name.length > 0 && name.length <= 128 && /^[A-Za-z0-9_.:-]+$/.test(name);}
function shouldRestoreRuntime(type: string): boolean { // Read-only discovery commands (`list_modes`, `list_models`, `list_reasoning_levels`) // are NOT restorers: they must not wake a stopped runtime (saga vr7kgx.6 hardening). They // route only when a live stream already exists and otherwise surface the normal routed // error (NOT_FOUND) without reviving the runtime — parity with `list_modes`. Only // work-carrying / mutating commands restore. `set_reasoning_level` follows the same // restore/wake policy as `set_model` (saga p7m8jp.3). return type === 'prompt' || type === 'set_mode' || type === 'set_model' || type === 'set_reasoning_level' || type === 'session_op';}
function stateReady(state: AgentState | undefined, kind: 'hello' | 'sync'): void { if (state === undefined) return; if (kind === 'hello') state.helloSeen = true; else state.syncSeen = true; if (state.helloSeen && state.syncSeen) state.resolveReady?.();}
// `isOrchestratorMessage` is re-exported for the orchestrator handler (tb4593.5) to identify// orchestrator events when fanning them out. Kept as a re-export to avoid a second import// site drifting from the catalog's definition.export { isOrchestratorMessage };