101 lines
3.6 KiB
TypeScript
101 lines
3.6 KiB
TypeScript
import { EventEmitter } from 'node:events';
|
|
|
|
export type AgentHubEventType = 'task' | 'handoff' | 'decision' | 'memory' | 'message' | '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);
|
|
}
|