spb/khaelor Public

1/**2 * KHAELOR3 * File: src/session/bus.ts4 * Description: Typed in-process event bus (write-ahead durable publish) and the ~16 ms delta coalescer.5 *6 * Author: Simon-Pierre Boucher7 * Contact: contact@spboucher.ai8 */910import { KhaelorError, ulid } from "../shared/index.js";11import type { Unsubscribe } from "../shared/index.js";12import {13  EVENT_SCHEMA_VERSION,14  isDurableEventType,15  isEphemeralEventType,16} from "./events.js";17import type {18  DurableEvent,19  DurableEventInput,20  EphemeralEvent,21  EphemeralEventInput,22  KhaelorEvent,23} from "./events.js";2425// ───────────────────────────── event bus ─────────────────────────────2627/**28 * Write-ahead sink for durable events. The SessionLog implements this:29 * it assigns the envelope (v/id/sessionId/seq/ts) and enqueues the30 * serialized append BEFORE the bus fans the event out (EVENT_MODEL.md §5.2).31 */32export interface DurableAppender {33  append(input: DurableEventInput): DurableEvent;34}3536/** Bus API (EVENT_MODEL.md §8). */37export interface EventBus {38  /** Append (write-ahead) then publish. Returns the full envelope. */39  publishDurable(input: DurableEventInput): DurableEvent;40  /** Fire-and-forget to live subscribers. Never persisted. */41  publishEphemeral(input: EphemeralEventInput): void;42  /** Typed subscription — handler parameter narrows by `type`. */43  on<T extends KhaelorEvent["type"]>(44    type: T,45    handler: (e: Extract<KhaelorEvent, { type: T }>) => void,46  ): Unsubscribe;47  /** Wildcard (projections, logging, coalescer). */48  onAny(handler: (e: KhaelorEvent) => void): Unsubscribe;49}5051export interface SessionEventBusOptions {52  sessionId: string;53  /** Write-ahead durable sink. When absent (tests, ephemeral-only usage) the bus self-assigns envelopes. */54  appender?: DurableAppender;55  /** Handler exceptions are caught and reported here — never rethrown, never block other subscribers. */56  onHandlerError?: (error: unknown, event: KhaelorEvent) => void;57}5859type AnyHandler = (e: KhaelorEvent) => void;6061/**62 * Synchronous, in-process, per-session event bus. Fan-out is synchronous and63 * in `seq` order; handlers must be fast and non-throwing (EVENT_MODEL.md §8).64 */65export class SessionEventBus implements EventBus {66  private readonly sessionId: string;67  private readonly appender: DurableAppender | undefined;68  private readonly onHandlerError: ((error: unknown, event: KhaelorEvent) => void) | undefined;69  private readonly byType = new Map<string, Set<AnyHandler>>();70  private readonly anyHandlers = new Set<AnyHandler>();71  /** Fallback seq counter used only when no appender is attached. */72  private selfSeq = 0;7374  constructor(options: SessionEventBusOptions) {75    this.sessionId = options.sessionId;76    this.appender = options.appender;77    this.onHandlerError = options.onHandlerError;78  }7980  publishDurable(input: DurableEventInput): DurableEvent {81    if (!isDurableEventType(input.type)) {82      throw new KhaelorError("invalid-event", `publishDurable: not a durable event type: ${String(input.type)}`);83    }84    const event: DurableEvent = this.appender85      ? this.appender.append(input)86      : ({87          v: EVENT_SCHEMA_VERSION,88          id: ulid(),89          sessionId: this.sessionId,90          seq: ++this.selfSeq,91          ts: Date.now(),92          type: input.type,93          payload: input.payload,94        } as DurableEvent);95    this.dispatch(event);96    return event;97  }9899  publishEphemeral(input: EphemeralEventInput): void {100    if (!isEphemeralEventType(input.type)) {101      throw new KhaelorError("invalid-event", `publishEphemeral: not an ephemeral event type: ${String(input.type)}`);102    }103    const event: EphemeralEvent = {104      id: ulid(),105      sessionId: this.sessionId,106      ts: Date.now(),107      type: input.type,108      payload: input.payload,109    } as EphemeralEvent;110    this.dispatch(event);111  }112113  on<T extends KhaelorEvent["type"]>(114    type: T,115    handler: (e: Extract<KhaelorEvent, { type: T }>) => void,116  ): Unsubscribe {117    let set = this.byType.get(type);118    if (!set) {119      set = new Set();120      this.byType.set(type, set);121    }122    const anyHandler = handler as AnyHandler;123    set.add(anyHandler);124    return () => {125      set.delete(anyHandler);126    };127  }128129  onAny(handler: (e: KhaelorEvent) => void): Unsubscribe {130    this.anyHandlers.add(handler);131    return () => {132      this.anyHandlers.delete(handler);133    };134  }135136  /** Number of live subscriptions (leak checks in tests). */137  subscriberCount(): number {138    let n = this.anyHandlers.size;139    for (const set of this.byType.values()) n += set.size;140    return n;141  }142143  private dispatch(event: KhaelorEvent): void {144    const typed = this.byType.get(event.type);145    // Snapshot so handlers may unsubscribe during delivery.146    const handlers: AnyHandler[] = [];147    if (typed) handlers.push(...typed);148    handlers.push(...this.anyHandlers);149    for (const handler of handlers) {150      try {151        handler(event);152      } catch (error) {153        this.onHandlerError?.(error, event);154      }155    }156  }157}158159// ───────────────────────────── coalescer ─────────────────────────────160161/** Key for streaming text/thinking blocks: `${requestId}#${blockIndex}`. */162export type BlockKey = string;163164export function blockKey(requestId: string, blockIndex: number): BlockKey {165  return `${requestId}#${blockIndex}`;166}167168/** One render frame worth of coalesced deltas + durable events (EVENT_MODEL.md §7). */169export interface CoalescedFrame {170  textAppends: Map<BlockKey, string>; // concatenated ModelTextDelta / ModelThinkingDelta171  toolInputPreviews: Map<string, string>; // latest accumulated partial JSON per toolUseId172  toolOutputAppends: Map<string, string>; // concatenated ToolOutput per toolUseId173  processOutputAppends: Map<string, string>; // concatenated ProcessOutput per processId174  durables: DurableEvent[]; // durable events in this window, in seq order175}176177export interface CoalescerOptions {178  /** Flush window in milliseconds (contract: ~16 ms). */179  windowMs?: number;180  /** Per-key per-frame cap on buffered append strings (default 16 KiB of UTF-16 units). */181  maxAppendPerKey?: number;182}183184export const COALESCER_TRUNCATION_MARKER = "…[preview truncated]";185186const DEFAULT_WINDOW_MS = 16;187const DEFAULT_MAX_APPEND_PER_KEY = 16 * 1024;188189/**190 * The single sanctioned buffering point between the bus and the TUI.191 * Buffers ephemeral deltas, flushes at most one frame per ~16 ms window,192 * and flushes immediately when a durable event settles a buffered key193 * (deltas delivered in the same frame, before the durable) or on Interrupted.194 */195export class Coalescer {196  private readonly windowMs: number;197  private readonly maxAppendPerKey: number;198  private readonly subscribers = new Set<(f: CoalescedFrame) => void>();199  private readonly unsubscribeBus: Unsubscribe;200201  private timer: ReturnType<typeof setTimeout> | null = null;202  private textAppends = new Map<BlockKey, string>();203  private toolOutputAppends = new Map<string, string>();204  private processOutputAppends = new Map<string, string>();205  private durables: DurableEvent[] = [];206  /** toolUseIds whose input preview changed in the current window. */207  private toolInputTouched = new Set<string>();208  /** Accumulated (capped) tool input previews — persists across frames until the tool settles. */209  private readonly toolInputAccum = new Map<string, string>();210  /** Keys that overflowed the per-frame cap in the current window. */211  private readonly overflowedKeys = new Set<string>();212213  constructor(bus: EventBus, options: CoalescerOptions = {}) {214    this.windowMs = options.windowMs ?? DEFAULT_WINDOW_MS;215    this.maxAppendPerKey = options.maxAppendPerKey ?? DEFAULT_MAX_APPEND_PER_KEY;216    this.unsubscribeBus = bus.onAny((event) => {217      this.ingest(event);218    });219  }220221  subscribe(onFrame: (frame: CoalescedFrame) => void): Unsubscribe {222    this.subscribers.add(onFrame);223    return () => {224      this.subscribers.delete(onFrame);225    };226  }227228  /** Detach from the bus and drop any buffered state without delivering it. */229  dispose(): void {230    this.unsubscribeBus();231    if (this.timer !== null) {232      clearTimeout(this.timer);233      this.timer = null;234    }235    this.resetWindow();236    this.subscribers.clear();237  }238239  private ingest(event: KhaelorEvent): void {240    switch (event.type) {241      case "model.text-delta":242      case "model.thinking-delta": {243        const key = blockKey(event.payload.requestId, event.payload.blockIndex);244        this.appendCapped(this.textAppends, key, event.payload.text);245        this.armTimer();246        return;247      }248      case "model.tool-input-delta": {249        const id = event.payload.toolUseId;250        const current = this.toolInputAccum.get(id) ?? "";251        this.toolInputAccum.set(252          id,253          this.capAppend(current, event.payload.partialJson, `input:${id}`),254        );255        this.toolInputTouched.add(id);256        this.armTimer();257        return;258      }259      case "tool.output": {260        this.appendCapped(this.toolOutputAppends, event.payload.toolUseId, event.payload.chunk);261        this.armTimer();262        return;263      }264      case "process.output": {265        this.appendCapped(this.processOutputAppends, event.payload.processId, event.payload.chunk);266        this.armTimer();267        return;268      }269      case "model.tool-call-started": {270        // UI hint only; nothing to buffer, but it starts a window so the271        // preview state (if any follows) flushes on cadence.272        this.armTimer();273        return;274      }275      default: {276        // Durable event.277        const durable = event as DurableEvent;278        this.durables.push(durable);279        if (this.settlesBufferedKey(durable) || durable.type === "user.interrupted") {280          this.flushNow();281        } else {282          this.armTimer();283        }284        return;285      }286    }287  }288289  private settlesBufferedKey(event: DurableEvent): boolean {290    switch (event.type) {291      case "model.text-block-completed":292      case "model.thinking-block-completed":293        return this.textAppends.has(blockKey(event.payload.requestId, event.payload.blockIndex));294      case "tool.requested": {295        const id = event.payload.toolUseId;296        const buffered = this.toolInputTouched.has(id);297        // Settlement replaces: accumulated preview is dropped permanently.298        this.toolInputAccum.delete(id);299        return buffered;300      }301      case "tool.completed":302      case "tool.failed":303      case "tool.cancelled":304        return this.toolOutputAppends.has(event.payload.toolUseId);305      case "process.exited":306        return this.processOutputAppends.has(event.payload.processId);307      default:308        return false;309    }310  }311312  private appendCapped(map: Map<string, string>, key: string, chunk: string): void {313    map.set(key, this.capAppend(map.get(key) ?? "", chunk, key));314  }315316  private capAppend(current: string, chunk: string, capKey: string): string {317    if (this.overflowedKeys.has(capKey)) return current;318    const combined = current + chunk;319    if (combined.length <= this.maxAppendPerKey) return combined;320    this.overflowedKeys.add(capKey);321    return combined.slice(0, this.maxAppendPerKey) + COALESCER_TRUNCATION_MARKER;322  }323324  private armTimer(): void {325    if (this.timer !== null) return;326    this.timer = setTimeout(() => {327      this.timer = null;328      this.flushNow();329    }, this.windowMs);330  }331332  private flushNow(): void {333    if (this.timer !== null) {334      clearTimeout(this.timer);335      this.timer = null;336    }337    if (338      this.textAppends.size === 0 &&339      this.toolInputTouched.size === 0 &&340      this.toolOutputAppends.size === 0 &&341      this.processOutputAppends.size === 0 &&342      this.durables.length === 0343    ) {344      return;345    }346    const toolInputPreviews = new Map<string, string>();347    for (const id of this.toolInputTouched) {348      const preview = this.toolInputAccum.get(id);349      if (preview !== undefined) toolInputPreviews.set(id, preview);350    }351    const frame: CoalescedFrame = {352      textAppends: this.textAppends,353      toolInputPreviews,354      toolOutputAppends: this.toolOutputAppends,355      processOutputAppends: this.processOutputAppends,356      durables: this.durables,357    };358    this.resetWindow();359    for (const subscriber of [...this.subscribers]) {360      try {361        subscriber(frame);362      } catch {363        // A slow or broken renderer must never break the engine.364      }365    }366  }367368  private resetWindow(): void {369    this.textAppends = new Map();370    this.toolOutputAppends = new Map();371    this.processOutputAppends = new Map();372    this.durables = [];373    this.toolInputTouched.clear();374    this.overflowedKeys.clear();375  }376}377