{"version":3,"file":"event-hub.d.ts","sourceRoot":"","sources":["../../src/server/event-hub.ts"],"names":[],"mappings":"AAAA;;;GAGG;AAEH,OAAO,KAAK,EAAE,aAAa,EAAE,MAAM,uBAAuB,CAAC;AAE3D,MAAM,MAAM,YAAY,GAAG,MAAM,GAAG,QAAQ,GAAG,QAAQ,CAAC;AAExD,mEAAmE;AACnE,MAAM,WAAW,gBAAgB;IAChC,IAAI,EAAE,YAAY,CAAC;IACnB,GAAG,EAAE,MAAM,CAAC;IACZ,IAAI,EAAE,MAAM,CAAC;IACb,UAAU,EAAE,MAAM,CAAC;IACnB,2EAA2E;IAC3E,MAAM,CAAC,EAAE,MAAM,CAAC;CAChB;AAED,MAAM,WAAW,gBAAgB;IAChC,IAAI,EAAE,QAAQ,GAAG,QAAQ,CAAC;IAC1B,KAAK,EAAE,MAAM,CAAC;IACd,KAAK,EAAE,MAAM,CAAC;IACd,OAAO,CAAC,EAAE,MAAM,CAAC;IACjB,KAAK,CAAC,EAAE,MAAM,CAAC;IACf,MAAM,CAAC,EAAE,MAAM,CAAC;CAChB;AAED,qEAAqE;AACrE,MAAM,WAAW,SAAS;IACzB,KAAK,CAAC,KAAK,EAAE,MAAM,EAAE,QAAQ,CAAC,EAAE,gBAAgB,GAAG,OAAO,GAAG,SAAS,CAAC;CACvE;AAED,MAAM,WAAW,eAAe;IAC/B,yCAAyC;IACzC,UAAU,CAAC,EAAE,MAAM,CAAC;IACpB,2DAA2D;IAC3D,WAAW,CAAC,EAAE,MAAM,CAAC;IACrB,4DAA4D;IAC5D,WAAW,CAAC,EAAE,MAAM,CAAC;IACrB,oEAAoE;IACpE,UAAU,CAAC,EAAE,MAAM,CAAC;CACpB;AAED,eAAO,MAAM,mBAAmB,OAAO,CAAC;AACxC,eAAO,MAAM,oBAAoB,QAAkB,CAAC;AACpD,gEAAgE;AAChE,eAAO,MAAM,oBAAoB,QAAkB,CAAC;AACpD,eAAO,MAAM,mBAAmB,QAAc,CAAC;AAgB/C;;;;GAIG;AACH,wBAAgB,qBAAqB,CAAC,KAAK,EAAE,MAAM,CAAC,MAAM,EAAE,OAAO,CAAC,GAAG,MAAM,CAAC,MAAM,EAAE,OAAO,CAAC,CA4B7F;AAED,qBAAa,QAAQ;IACpB,OAAO,CAAC,GAAG,CAAK;IAChB,OAAO,CAAC,aAAa,CAAK;IAC1B,OAAO,CAAC,QAAQ,CAAC,MAAM,CAA4B;IACnD,OAAO,CAAC,QAAQ,CAAC,OAAO,CAAwB;IAChD,OAAO,CAAC,QAAQ,CAAC,OAAO,CAA4B;IACpD,OAAO,CAAC,cAAc,CAAC,CAA2E;IAElG,YAAY,OAAO,GAAE,MAAM,GAAG,eAAoB,EASjD;IAED,wFAAwF;IACxF,iBAAiB,CAAC,SAAS,EAAE,CAAC,GAAG,EAAE,MAAM,EAAE,KAAK,EAAE,MAAM,CAAC,MAAM,EAAE,OAAO,CAAC,KAAK,MAAM,CAAC,MAAM,EAAE,OAAO,CAAC,GAAG,IAAI,CAE3G;IAED,+EAA+E;IAC/E,OAAO,CAAC,GAAG,EAAE,MAAM,EAAE,QAAQ,EAAE,MAAM,CAAC,MAAM,EAAE,OAAO,CAAC,GAAG,aAAa,CAWrE;IAED;;;;OAIG;IACH,MAAM,CAAC,MAAM,EAAE,SAAS,EAAE,WAAW,CAAC,EAAE,MAAM,EAAE,QAAQ,CAAC,EAAE,CAAC,UAAU,EAAE,gBAAgB,KAAK,IAAI,GAAG,MAAM,IAAI,CA0B7G;IAED,IAAI,WAAW,IAAI,MAAM,CAExB;IAED,IAAI,YAAY,IAAI,MAAM,CAEzB;IAED,IAAI,YAAY,IAAI,MAAM,CAEzB;IAED,8EAA8E;IAC9E,IAAI,eAAe,IAAI,MAAM,CAE5B;IAED,OAAO,CAAC,SAAS;IAMjB,OAAO,CAAC,MAAM;IAUd,OAAO,CAAC,WAAW;IASnB,OAAO,CAAC,YAAY;IAQpB,iGAAiG;IACjG,OAAO,CAAC,aAAa;IASrB;;;OAGG;IACH,OAAO,CAAC,cAAc;IAOtB,OAAO,CAAC,MAAM;IAMd,OAAO,CAAC,KAAK;CAeb;AAED,+EAA+E;AAC/E,wBAAgB,cAAc,CAAC,QAAQ,EAAE,aAAa,GAAG,MAAM,CAE9D;AAED,+EAA+E;AAC/E,wBAAgB,oBAAoB,IAAI,MAAM,CAE7C","sourcesContent":["/**\n * SSE event hub — projects runtime events for the dashboard, keeps a\n * byte-bounded replay history, and fans serialized frames out to browsers.\n */\n\nimport type { EventEnvelope } from \"../shared/protocol.js\";\n\nexport type SseWriteKind = \"live\" | \"replay\" | \"resync\";\n\n/** Metadata deliberately excludes the serialized event payload. */\nexport interface SseWriteMetadata {\n\tkind: SseWriteKind;\n\tseq: number;\n\ttype: string;\n\tframeBytes: number;\n\t/** Safe synthesized-barrier classification, never runtime payload data. */\n\treason?: string;\n}\n\nexport interface ReplayDiagnostic {\n\tkind: \"replay\" | \"resync\";\n\tcount: number;\n\tbytes: number;\n\tfromSeq?: number;\n\ttoSeq?: number;\n\treason?: string;\n}\n\n/** Return false when this connection can no longer accept frames. */\nexport interface SseClient {\n\twrite(chunk: string, metadata?: SseWriteMetadata): boolean | undefined;\n}\n\nexport interface EventHubOptions {\n\t/** Maximum number of retained frames. */\n\tbufferSize?: number;\n\t/** Maximum encoded bytes retained for reconnect replay. */\n\tbufferBytes?: number;\n\t/** Maximum encoded bytes written during a single replay. */\n\treplayBytes?: number;\n\t/** Largest projected event frame that may be delivered directly. */\n\teventBytes?: number;\n}\n\nexport const DEFAULT_BUFFER_SIZE = 2000;\nexport const DEFAULT_BUFFER_BYTES = 8 * 1024 * 1024;\n/** Kept below the response destruction ceiling in server.ts. */\nexport const DEFAULT_REPLAY_BYTES = 3 * 1024 * 1024;\nexport const DEFAULT_EVENT_BYTES = 1024 * 1024;\n\ninterface SerializedEnvelope {\n\tenvelope: EventEnvelope;\n\tframe: string;\n\tbytes: number;\n\t/** Present only for barriers synthesized by this hub, never runtime data. */\n\tresyncReason?: string;\n}\n\nfunction omit<T extends Record<string, unknown>>(event: T, ...keys: string[]): Record<string, unknown> {\n\tconst copy = { ...event };\n\tfor (const key of keys) delete copy[key];\n\treturn copy;\n}\n\n/**\n * Dashboard-only transport projection. Each removed field is cumulative data\n * that the dashboard reducer does not read. Unknown event types are returned\n * exactly as received so extensions and future runtimes remain forward-safe.\n */\nexport function projectDashboardEvent(event: Record<string, unknown>): Record<string, unknown> {\n\tswitch (event.type) {\n\t\tcase \"agent_end\":\n\t\t\treturn omit(event, \"messages\");\n\t\tcase \"turn_end\":\n\t\t\treturn omit(event, \"message\", \"toolResults\");\n\t\tcase \"message_update\": {\n\t\t\tconst projected = omit(event, \"message\");\n\t\t\tconst streamEvent = event.assistantMessageEvent;\n\t\t\treturn streamEvent && typeof streamEvent === \"object\" && !Array.isArray(streamEvent)\n\t\t\t\t? { ...projected, assistantMessageEvent: omit(streamEvent as Record<string, unknown>, \"partial\") }\n\t\t\t\t: projected;\n\t\t}\n\t\tcase \"tool_execution_update\":\n\t\t\treturn omit(event, \"args\");\n\t\tcase \"stream_retry\":\n\t\t\treturn omit(event, \"discardedPartial\");\n\t\tcase \"length_retry\":\n\t\t\treturn omit(event, \"discardedPartial\");\n\t\tcase \"background_agent_event\": {\n\t\t\tconst child = event.event;\n\t\t\treturn child && typeof child === \"object\" && !Array.isArray(child)\n\t\t\t\t? { ...event, event: projectDashboardEvent(child as Record<string, unknown>) }\n\t\t\t\t: event;\n\t\t}\n\t\tdefault:\n\t\t\treturn event;\n\t}\n}\n\nexport class EventHub {\n\tprivate seq = 0;\n\tprivate bufferedBytes = 0;\n\tprivate readonly buffer: SerializedEnvelope[] = [];\n\tprivate readonly clients = new Set<SseClient>();\n\tprivate readonly options: Required<EventHubOptions>;\n\tprivate eventProjector?: (key: string, event: Record<string, unknown>) => Record<string, unknown>;\n\n\tconstructor(options: number | EventHubOptions = {}) {\n\t\tthis.options = {\n\t\t\tbufferSize: typeof options === \"number\" ? options : (options.bufferSize ?? DEFAULT_BUFFER_SIZE),\n\t\t\tbufferBytes:\n\t\t\t\ttypeof options === \"number\" ? DEFAULT_BUFFER_BYTES : (options.bufferBytes ?? DEFAULT_BUFFER_BYTES),\n\t\t\treplayBytes:\n\t\t\t\ttypeof options === \"number\" ? DEFAULT_REPLAY_BYTES : (options.replayBytes ?? DEFAULT_REPLAY_BYTES),\n\t\t\teventBytes: typeof options === \"number\" ? DEFAULT_EVENT_BYTES : (options.eventBytes ?? DEFAULT_EVENT_BYTES),\n\t\t};\n\t}\n\n\t/** Install the server-owned browser projection before frame sizing/replay retention. */\n\tsetEventProjector(projector: (key: string, event: Record<string, unknown>) => Record<string, unknown>): void {\n\t\tthis.eventProjector = projector;\n\t}\n\n\t/** Publish an event from a runtime; assigns a sequence number and fans out. */\n\tpublish(key: string, rawEvent: Record<string, unknown>): EventEnvelope {\n\t\tconst baseProjection = projectDashboardEvent(rawEvent);\n\t\tconst event = this.eventProjector ? this.eventProjector(key, baseProjection) : baseProjection;\n\t\tconst serialized = this.serialize(this.seq + 1, key, event);\n\t\tif (serialized.bytes > this.options.eventBytes) {\n\t\t\treturn this.publishResync(\"oversized_event\").envelope;\n\t\t}\n\t\tthis.seq += 1;\n\t\tthis.retain(serialized);\n\t\tthis.fanout(serialized, \"live\");\n\t\treturn serialized.envelope;\n\t}\n\n\t/**\n\t * Attach a client. Replays only a complete, bounded range. A gap or replay\n\t * over budget receives a recovery frame only on this connection; it never\n\t * consumes a global sequence or disturbs healthy clients' ordered stream.\n\t */\n\tattach(client: SseClient, lastEventId?: number, onReplay?: (diagnostic: ReplayDiagnostic) => void): () => void {\n\t\tif (lastEventId !== undefined) {\n\t\t\tconst replay = this.replayAfter(lastEventId);\n\t\t\tif (!replay) {\n\t\t\t\tconst reason = this.resyncReason(lastEventId);\n\t\t\t\tconst barrier = this.targetedResync(reason);\n\t\t\t\tonReplay?.({ kind: \"resync\", count: 1, bytes: barrier.bytes, toSeq: barrier.envelope.seq, reason });\n\t\t\t\tif (!this.write(client, barrier, \"resync\")) return () => {};\n\t\t\t} else {\n\t\t\t\tconst bytes = replay.reduce((total, item) => total + item.bytes, 0);\n\t\t\t\tonReplay?.({\n\t\t\t\t\tkind: \"replay\",\n\t\t\t\t\tcount: replay.length,\n\t\t\t\t\tbytes,\n\t\t\t\t\tfromSeq: replay[0]?.envelope.seq,\n\t\t\t\t\ttoSeq: replay[replay.length - 1]?.envelope.seq,\n\t\t\t\t});\n\t\t\t\tfor (const serialized of replay) {\n\t\t\t\t\tif (!this.write(client, serialized, \"replay\")) return () => {};\n\t\t\t\t}\n\t\t\t}\n\t\t}\n\t\tthis.clients.add(client);\n\t\treturn () => {\n\t\t\tthis.clients.delete(client);\n\t\t};\n\t}\n\n\tget clientCount(): number {\n\t\treturn this.clients.size;\n\t}\n\n\tget historyBytes(): number {\n\t\treturn this.bufferedBytes;\n\t}\n\n\tget historyCount(): number {\n\t\treturn this.buffer.length;\n\t}\n\n\t/** Current sequence, captured without emitting, retaining, or fanning out. */\n\tget currentSequence(): number {\n\t\treturn this.seq;\n\t}\n\n\tprivate serialize(seq: number, key: string, event: Record<string, unknown>): SerializedEnvelope {\n\t\tconst envelope: EventEnvelope = { seq, key, event };\n\t\tconst frame = formatSseFrame(envelope);\n\t\treturn { envelope, frame, bytes: Buffer.byteLength(frame) };\n\t}\n\n\tprivate retain(serialized: SerializedEnvelope): void {\n\t\tthis.buffer.push(serialized);\n\t\tthis.bufferedBytes += serialized.bytes;\n\t\twhile (this.buffer.length > this.options.bufferSize || this.bufferedBytes > this.options.bufferBytes) {\n\t\t\tconst evicted = this.buffer.shift();\n\t\t\tif (!evicted) break;\n\t\t\tthis.bufferedBytes -= evicted.bytes;\n\t\t}\n\t}\n\n\tprivate replayAfter(lastEventId: number): SerializedEnvelope[] | undefined {\n\t\tconst oldest = this.buffer[0]?.envelope.seq;\n\t\tconst newest = this.buffer[this.buffer.length - 1]?.envelope.seq;\n\t\tif (oldest === undefined || newest === undefined || lastEventId < oldest - 1 || lastEventId > newest)\n\t\t\treturn undefined;\n\t\tconst replay = this.buffer.filter((item) => item.envelope.seq > lastEventId);\n\t\treturn replay.reduce((bytes, item) => bytes + item.bytes, 0) <= this.options.replayBytes ? replay : undefined;\n\t}\n\n\tprivate resyncReason(lastEventId: number): string {\n\t\tconst oldest = this.buffer[0]?.envelope.seq;\n\t\tconst newest = this.buffer[this.buffer.length - 1]?.envelope.seq;\n\t\tif (oldest === undefined) return \"empty_buffer\";\n\t\tif (lastEventId < oldest - 1 || lastEventId > newest!) return \"buffer_gap\";\n\t\treturn \"replay_over_budget\";\n\t}\n\n\t/** Explicit oversized events are globally unrecoverable, so retain and fan out their barrier. */\n\tprivate publishResync(reason: string): SerializedEnvelope {\n\t\tthis.seq += 1;\n\t\tconst barrier = this.serialize(this.seq, \"\", { type: \"dashboard_resync\", reason });\n\t\tbarrier.resyncReason = reason;\n\t\tthis.retain(barrier);\n\t\tthis.fanout(barrier, \"resync\");\n\t\treturn barrier;\n\t}\n\n\t/**\n\t * A stale reconnect needs the current ordering cursor, not a new global\n\t * event. With no events yet, establish sequence 1 so it has a usable cursor.\n\t */\n\tprivate targetedResync(reason: string): SerializedEnvelope {\n\t\tif (this.seq === 0) this.seq = 1;\n\t\tconst barrier = this.serialize(this.seq, \"\", { type: \"dashboard_resync\", reason });\n\t\tbarrier.resyncReason = reason;\n\t\treturn barrier;\n\t}\n\n\tprivate fanout(serialized: SerializedEnvelope, kind: SseWriteKind): void {\n\t\tfor (const client of this.clients) {\n\t\t\tif (!this.write(client, serialized, kind)) this.clients.delete(client);\n\t\t}\n\t}\n\n\tprivate write(client: SseClient, serialized: SerializedEnvelope, kind: SseWriteKind): boolean {\n\t\ttry {\n\t\t\treturn (\n\t\t\t\tclient.write(serialized.frame, {\n\t\t\t\t\tkind,\n\t\t\t\t\tseq: serialized.envelope.seq,\n\t\t\t\t\ttype: String(serialized.envelope.event.type ?? \"unknown\"),\n\t\t\t\t\tframeBytes: serialized.bytes,\n\t\t\t\t\t...(serialized.resyncReason ? { reason: serialized.resyncReason } : {}),\n\t\t\t\t}) !== false\n\t\t\t);\n\t\t} catch {\n\t\t\treturn false;\n\t\t}\n\t}\n}\n\n/** Format an envelope as an SSE frame with the sequence number as event id. */\nexport function formatSseFrame(envelope: EventEnvelope): string {\n\treturn `id: ${envelope.seq}\\ndata: ${JSON.stringify(envelope)}\\n\\n`;\n}\n\n/** Observable liveness signal; intentionally unnumbered and never buffered. */\nexport function formatHeartbeatFrame(): string {\n\treturn \"event: heartbeat\\ndata: {}\\n\\n\";\n}\n"]}