Repository Public
1/**2 * Fan-out of kernel events (`AosEvent`) to the authenticated `/api/v1/events` WebSockets.3 *4 * One kernel subscription per process (`listenEvents`). Each socket first replays persisted5 * events after its cursor from Postgres while live events are buffered, then switches to live;6 * ids are monotonic per socket, so the hand-over neither loses nor duplicates an event.7 * Events are delivered only to the owner of their mission. Consumers that cannot keep up are8 * dropped, not buffered. Every frame passes through the secret scrubber.9 */10import { eventForBrowser } from './browser-view.js';11import type { FastifyBaseLogger } from 'fastify';12import type WebSocket from 'ws';13import type { AosEvent } from '@aos/shared';14import type { ServerConfig } from './config.js';15import type { Db } from './db.js';16import { WS_CLOSE } from './dto-auth.js';17import type { KernelPort } from './kernel-port.js';18import type { Notifier } from './modules/notifications/service.js';1920const TERMINAL = new Set(['done', 'partial', 'blocked', 'failed', 'cancelled', 'budget_exhausted']);21const MAX_BUFFERED_EVENTS = 20_000;22const OWNER_CACHE_MAX = 5000;2324interface Sub {25 socket: WebSocket;26 userId: string;27 missionId: string | null;28 /** Live events held back while the replay runs; null once live. */29 buffer: AosEvent[] | null;30 lastId: number;31}32interface EventRow {33 id: number; mission_id: string; run_id: string | null; seq: number | null; ts: Date; kind: string; trust: string; payload: Record<string, unknown>;34}3536export const notice = (code: string, message: string, extra: Record<string, unknown> = {}): AosEvent => ({37 id: -1, missionId: '', runId: null, seq: null, ts: new Date().toISOString(), kind: 'system.notice', trust: 'system', payload: { code, message, ...extra },38});3940export class EventStream {41 private readonly subs = new Set<Sub>();42 private readonly owners = new Map<string, string>();43 private chain: Promise<void> = Promise.resolve();44 private stopListening: (() => Promise<void>) | null = null;4546 constructor(47 private readonly d: { db: Db; kernel: KernelPort; log: FastifyBaseLogger; notifier: Notifier; config: ServerConfig },48 ) {}4950 get size(): number {51 return this.subs.size;52 }53 get listening(): boolean {54 return this.stopListening !== null;55 }5657 async start(): Promise<void> {58 if (this.stopListening) return;59 this.stopListening = await this.d.kernel.listenEvents((ev) => {60 // Serialised: the owner lookup is asynchronous and order must be preserved.61 this.chain = this.chain.then(() => this.dispatch(ev)).catch((err: unknown) => this.d.log.error({ err }, 'event dispatch failed'));62 });63 }64 /** Resolves when every event received so far has been dispatched. */65 idle(): Promise<void> {66 return this.chain;67 }68 async stop(): Promise<void> {69 const stop = this.stopListening;70 this.stopListening = null;71 if (stop) await stop().catch(() => undefined);72 await this.chain;73 }7475 /** Frames are text JSON, one `AosEvent` each. */76 sendTo(socket: WebSocket, ev: AosEvent): void {77 if (socket.readyState === socket.OPEN) socket.send(this.d.config.secrets.scrub(JSON.stringify(eventForBrowser(ev))));78 }7980 private async ownerOf(missionId: string): Promise<string | null> {81 const hit = this.owners.get(missionId);82 if (hit) return hit;83 if (!/^[0-9a-f-]{36}$/i.test(missionId)) return null;84 const row = await this.d.db.one<{ owner_id: string }>('SELECT owner_id FROM missions WHERE id = $1', [missionId]);85 if (!row) return null;86 if (this.owners.size >= OWNER_CACHE_MAX) this.owners.clear();87 this.owners.set(missionId, row.owner_id);88 return row.owner_id;89 }9091 private deliver(sub: Sub, ev: AosEvent): void {92 if (!this.subs.has(sub)) return;93 if (ev.id >= 0) {94 if (ev.id <= sub.lastId) return; // already replayed95 sub.lastId = ev.id;96 }97 if (sub.socket.bufferedAmount > this.d.config.eventsSlowBytes) {98 this.subs.delete(sub);99 sub.socket.close(WS_CLOSE.BACKPRESSURE, 'slow consumer');100 // close() waits for the peer; a consumer that is not reading must not hold the socket open.101 setTimeout(() => sub.socket.terminate(), 1000).unref();102 return;103 }104 this.sendTo(sub.socket, ev);105 }106107 private async dispatch(ev: AosEvent): Promise<void> {108 const owner = await this.ownerOf(ev.missionId);109 if (!owner) return;110 for (const sub of [...this.subs]) {111 if (sub.userId !== owner || (sub.missionId !== null && sub.missionId !== ev.missionId)) continue;112 if (sub.buffer) {113 if (sub.buffer.length >= MAX_BUFFERED_EVENTS) {114 this.subs.delete(sub);115 sub.socket.terminate();116 } else sub.buffer.push(ev);117 } else this.deliver(sub, ev);118 }119 if (ev.id >= 0) await this.notifyOwner(owner, ev).catch((err: unknown) => this.d.log.error({ err }, 'notification failed'));120 }121122 /** "Something needs you" / "a mission ended": titles only, never model or tool output. */123 private async notifyOwner(owner: string, ev: AosEvent): Promise<void> {124 let kind: 'wait.approval' | 'wait.question' | 'mission.finished';125 let title: string;126 if (ev.kind === 'wait.created' && (ev.payload.kind === 'approval' || ev.payload.kind === 'question')) {127 kind = ev.payload.kind === 'approval' ? 'wait.approval' : 'wait.question';128 title = ev.payload.kind === 'approval' ? 'Approval needed' : 'The agent has a question';129 } else if (ev.kind === 'mission.status' && typeof ev.payload.status === 'string' && TERMINAL.has(ev.payload.status)) {130 kind = 'mission.finished';131 title = `Mission ${ev.payload.status.replace('_', ' ')}`;132 } else return;133 const m = await this.d.db.one<{ title: string }>('SELECT title FROM missions WHERE id = $1', [ev.missionId]);134 await this.d.notifier.notify(owner, kind, title, m?.title ?? null, `/mission/?id=${ev.missionId}`);135 }136137 /**138 * Registers the socket. `cursor` = last event id the browser has (null ⇒ live only). Returns139 * the unregister function; the replay runs in the background.140 */141 attach(socket: WebSocket, userId: string, opts: { cursor: number | null; missionId: string | null }): () => void {142 const sub: Sub = { socket, userId, missionId: opts.missionId, buffer: [], lastId: opts.cursor ?? -1 };143 this.subs.add(sub);144 void this.replay(sub, opts.cursor).catch((err: unknown) => {145 this.d.log.error({ err }, 'event replay failed');146 this.subs.delete(sub);147 socket.close(1011, 'replay failed');148 });149 return () => this.subs.delete(sub);150 }151152 private async replay(sub: Sub, cursor: number | null): Promise<void> {153 if (cursor !== null) {154 const max = this.d.config.eventsReplayMax;155 const rows = await this.d.db.q<EventRow>(156 `SELECT e.id, e.mission_id, e.run_id, e.seq, e.ts, e.kind, e.trust, e.payload157 FROM events e JOIN missions m ON m.id = e.mission_id158 WHERE m.owner_id = $1 AND e.id > $2 AND ($3::uuid IS NULL OR e.mission_id = $3::uuid)159 ORDER BY e.id LIMIT $4`,160 [sub.userId, cursor, sub.missionId, max + 1],161 );162 for (const r of rows.slice(0, max)) {163 this.deliver(sub, {164 id: r.id, missionId: r.mission_id, runId: r.run_id, seq: r.seq, ts: r.ts.toISOString(),165 kind: r.kind as AosEvent['kind'], trust: r.trust as AosEvent['trust'], payload: r.payload ?? {},166 });167 }168 if (rows.length > max) {169 // More than the bound is missing: the browser must refetch its state instead of replaying.170 const head = await this.d.db.one<{ id: number | null }>('SELECT max(id) AS id FROM events');171 this.sendTo(sub.socket, notice('gap', 'gap, refetch', { latestId: head?.id ?? sub.lastId }));172 }173 }174 // Hand-over: synchronous from here, so no live event can slip between the flush and live mode.175 const held = sub.buffer ?? [];176 sub.buffer = null;177 if (this.subs.has(sub)) for (const ev of held) this.deliver(sub, ev);178 }179}180