agenthub/src/server/events.ts

import { EventEmitter } from 'node:events';

export type AgentHubEventType = 'task' | 'handoff' | 'decision' | 'memory' | 'message' | 'ask' | 'agent';
export type AgentHubEventAction = 'created' | 'updated' | 'deleted' | 'joined' | 'left';

export interface AgentHubEvent {
  type: AgentHubEventType;
  action: AgentHubEventAction;
  /** Entity id, or — for `agent` presence events — the agent name. */
  id: string;
  title?: string;
  status?: string;
  role?: string;
  assignedTo?: string;
  claimedBy?: string;
  reviewer?: string;
}

/**
 * In-process event bus. Routes emit here on every successful mutating REST
 * operation; the SSE /events handler fans them out to connected subscribers.
 *
 * Singleton per Node.js process — in server mode that is always exactly one
 * process, which is the intended topology.
 */
/** A live progress line an agent streams while working a task. */
export interface TaskLogPayload {
  taskId: string;
  ts: string;
  agent?: string;
  level?: string;
  text: string;
}

class AgentHubEventBus extends EventEmitter {
  /** Publish a change event to all current SSE subscribers. */
  publish(event: AgentHubEvent): void {
    this.emit('change', event);
  }

  /** Publish a task-log line — delivered as a NAMED `task-log` SSE event so the
   *  board's generic onmessage handler ignores it and only the task-detail
   *  live console picks it up. */
  publishLog(payload: TaskLogPayload): void {
    this.emit('log', payload);
  }
}

export const eventBus = new AgentHubEventBus();
// Allow an arbitrary number of SSE clients without triggering the
// default-listener-count warning.
eventBus.setMaxListeners(0);

// ─── De-duplication between the REST and filesystem-watch emit paths ────────
// Both mutating REST routes AND the filesystem watcher (fsWatch.ts) can observe
// the same change: a REST POST/PATCH writes the entity file, then the watcher
// sees that very write. Without coordination the subscriber would receive the
// event twice. We solve it with a short-lived signature cache keyed by
// `${type}:${id}:${updatedAt|createdAt}`. Whichever path emits first records the
// signature; the other path sees it via `seenRecently()` and stays silent.
const DEDUP_TTL_MS = 15_000;
const recentlyEmitted = new Map<string, number>();

/** Stable key for a single logical mutation of one entity revision. */
export function signatureOf(type: string, id: string, stamp: string | undefined): string {
  return `${type}:${id}:${stamp ?? ''}`;
}

function markEmitted(signature: string): void {
  const now = Date.now();
  recentlyEmitted.set(signature, now);
  // Opportunistic GC so the map can't grow unbounded under heavy churn.
  if (recentlyEmitted.size > 500) {
    for (const [key, ts] of recentlyEmitted) {
      if (now - ts > DEDUP_TTL_MS) recentlyEmitted.delete(key);
    }
  }
}

/** True if a change with this signature was emitted within the dedup window. */
export function seenRecently(signature: string): boolean {
  const ts = recentlyEmitted.get(signature);
  if (ts === undefined) return false;
  if (Date.now() - ts > DEDUP_TTL_MS) {
    recentlyEmitted.delete(signature);
    return false;
  }
  return true;
}

/**
 * Publish a change and record its signature for cross-path de-duplication.
 * Used by both the REST routes and the filesystem watcher. `stamp` is the
 * entity's `updatedAt` (or `createdAt` for handoffs) and must match what the
 * other path derives from the same file revision.
 */
export function emitChange(event: AgentHubEvent, stamp: string | undefined): void {
  markEmitted(signatureOf(event.type, event.id, stamp));
  eventBus.publish(event);
}