spb/worthdoing Public

1/**2 * WorthDoing.ai3 * Author: Simon-Pierre Boucher4 * Contact: contact@spboucher.ai5 * File: src/lib/agent/events.ts6 * Description: Agent event bus — persists AgentEvents to Postgres and publishes them in-process for SSE.7 */8import { EventEmitter } from "events";9import { and, asc, eq, gt } from "drizzle-orm";10import { db } from "@/lib/db/client";11import { agentEvents } from "@/lib/db/schema";1213export type AgentEventType =14  | "investigation.started"15  | "agent.plan"16  | "agent.error"17  | "phase.changed"18  | "budget.updated"19  | "search.started"20  | "search.completed"21  | "scrape.started"22  | "scrape.completed"23  | "scrape.failed"24  | "crawl.started"25  | "crawl.completed"26  | "crawl.failed"27  | "extract.started"28  | "extract.completed"29  | "extract.failed"30  | "hypothesis.created"31  | "hypothesis.updated"32  | "hypothesis.rejected"33  | "evidence.saved"34  | "claim.created"35  | "opportunity.created"36  | "opportunity.updated"37  | "report.started"38  | "report.delta"39  | "report.completed"40  | "investigation.completed"41  | "investigation.failed";4243export type AgentEvent = {44  seq: number;45  investigationId: string;46  type: AgentEventType;47  payload: Record<string, unknown>;48  createdAt: string;49};5051const globalForBus = globalThis as unknown as {52  __wdBus?: EventEmitter;53  __wdSeq?: Map<string, number>;54};55const bus = (globalForBus.__wdBus ??= new EventEmitter());56bus.setMaxListeners(500);57const seqCounters = (globalForBus.__wdSeq ??= new Map<string, number>());5859async function nextSeq(investigationId: string): Promise<number> {60  const current = seqCounters.get(investigationId);61  if (current !== undefined) {62    seqCounters.set(investigationId, current + 1);63    return current + 1;64  }65  const rows = await db66    .select({ seq: agentEvents.seq })67    .from(agentEvents)68    .where(eq(agentEvents.investigationId, investigationId))69    .orderBy(asc(agentEvents.seq));70  const max = rows.length ? rows[rows.length - 1].seq : 0;71  seqCounters.set(investigationId, max + 1);72  return max + 1;73}7475/**76 * Persist and publish a real agent event. Every UI event flows through here —77 * nothing user-visible is ever fabricated.78 */79export async function emitEvent(80  investigationId: string,81  type: AgentEventType,82  payload: Record<string, unknown>,83  options: { transient?: boolean } = {},84): Promise<AgentEvent> {85  const seq = await nextSeq(investigationId);86  const event: AgentEvent = {87    seq,88    investigationId,89    type,90    payload,91    createdAt: new Date().toISOString(),92  };93  // report.delta events are high-frequency; publish live but skip per-delta persistence94  // (the full report is persisted on report.completed).95  if (!options.transient) {96    await db.insert(agentEvents).values({ investigationId, seq, type, payload });97  }98  bus.emit(`inv:${investigationId}`, event);99  return event;100}101102export function subscribe(investigationId: string, listener: (e: AgentEvent) => void): () => void {103  const channel = `inv:${investigationId}`;104  bus.on(channel, listener);105  return () => bus.off(channel, listener);106}107108/** Replay persisted events after a given seq (for SSE Last-Event-ID reconnects). */109export async function eventsAfter(investigationId: string, afterSeq: number): Promise<AgentEvent[]> {110  const rows = await db111    .select()112    .from(agentEvents)113    .where(and(eq(agentEvents.investigationId, investigationId), gt(agentEvents.seq, afterSeq)))114    .orderBy(asc(agentEvents.seq));115  return rows.map((r) => ({116    seq: r.seq,117    investigationId: r.investigationId,118    type: r.type as AgentEventType,119    payload: r.payload as Record<string, unknown>,120    createdAt: r.createdAt.toISOString(),121  }));122}123