From 0456f78c86c436d477878bd486884de6b38fd71f Mon Sep 17 00:00:00 2001 From: iaj6 Date: Sun, 12 Jul 2026 11:32:00 -0400 Subject: [PATCH] fix(events): dedupe SSE emission, scope live streams, fix reconnect leak Fixes the July 2026 audit's SSE/event-streaming cluster: - Add an EventPollCursor to @agentops/db pairing an inclusive (gte) since boundary with the event IDs already emitted at that timestamp, so polling emits each event exactly once without dropping a second event that shares the boundary timestamp. Used by both the web SSE route and CLI 'events tail', which previously re-emitted the newest event on every poll. - useEvents: hold events + total in one state object updated through a pure applyLiveEvent reducer, so the 'N total' counter only increments when an event is genuinely new instead of on every SSE delivery. - useEventSource: store the backoff reconnect timer handle and clear it on cleanup and on (re)connect, so unmount/filter changes can no longer leak a pending reconnect that opens an EventSource nobody closes. - Forward view/userId scope params on the SSE connection from useRuns and useEvents (plus a client-side run.userId guard in useRuns), so an admin viewing ?userId=X no longer sees other users' live runs/events. - SSE route: re-resolve the owned sourceId set on each poll so runs and sessions created after connect appear in the owner's live stream without a reconnect. Co-Authored-By: Claude Fable 5 --- packages/cli/src/commands/events.ts | 24 +++- .../db/src/__tests__/event-cursor.test.ts | 134 ++++++++++++++++++ packages/db/src/__tests__/events.test.ts | 11 ++ packages/db/src/event-cursor.ts | 67 +++++++++ packages/db/src/index.ts | 1 + packages/web/src/app/api/events/route.ts | 49 +++++-- packages/web/src/hooks/useEventSource.ts | 33 ++++- packages/web/src/hooks/useEvents.ts | 28 ++-- packages/web/src/hooks/useRuns.ts | 19 ++- .../web/src/lib/__tests__/live-events.test.ts | 44 ++++++ packages/web/src/lib/live-events.ts | 28 ++++ 11 files changed, 402 insertions(+), 36 deletions(-) create mode 100644 packages/db/src/__tests__/event-cursor.test.ts create mode 100644 packages/db/src/event-cursor.ts create mode 100644 packages/web/src/lib/__tests__/live-events.test.ts create mode 100644 packages/web/src/lib/live-events.ts diff --git a/packages/cli/src/commands/events.ts b/packages/cli/src/commands/events.ts index 1c9363b..29c4b6c 100644 --- a/packages/cli/src/commands/events.ts +++ b/packages/cli/src/commands/events.ts @@ -1,7 +1,14 @@ import { Command } from "commander"; import { EventCategory } from "@agentops/core"; import type { AgentEvent } from "@agentops/core"; -import { getDb, listEvents, getEventsBySource, getRecentEvents } from "@agentops/db"; +import { + getDb, + listEvents, + getEventsBySource, + getRecentEvents, + createEventPollCursor, + advanceEventPollCursor, +} from "@agentops/db"; import { table } from "../format.js"; const dim = (s: string) => `\x1b[2m${s}\x1b[0m`; @@ -97,21 +104,26 @@ export function registerEventsCommands(program: Command): void { console.log(dim("Press Ctrl+C to stop")); console.log(); - let lastTimestamp = new Date().toISOString(); + // listEvents' `since` is inclusive (gte) so equal-timestamp events + // are never dropped; the cursor tracks IDs already printed at the + // boundary timestamp so nothing is printed twice. + let cursor = createEventPollCursor(new Date().toISOString()); const interval = setInterval(() => { - const results = listEvents(db, { + const batch = listEvents(db, { type: opts.type, - since: lastTimestamp, + since: cursor.since, limit: 100, }); - for (const event of results.reverse()) { + const { fresh, next } = advanceEventPollCursor(cursor, batch); + cursor = next; + + for (const event of fresh) { const now = new Date().toLocaleTimeString(); console.log( `${dim(now)} ${colorCategory(event.category)} ${cyan(event.type)} ${dim(event.sourceId)}`, ); - lastTimestamp = event.timestamp; } }, 1000); diff --git a/packages/db/src/__tests__/event-cursor.test.ts b/packages/db/src/__tests__/event-cursor.test.ts new file mode 100644 index 0000000..3e09e42 --- /dev/null +++ b/packages/db/src/__tests__/event-cursor.test.ts @@ -0,0 +1,134 @@ +import { describe, it, expect, beforeEach } from "vitest"; +import { getDb } from "../connection.js"; +import type { AgentOpsDb } from "../connection.js"; +import { insertEvent, listEvents } from "../events.js"; +import { + createEventPollCursor, + advanceEventPollCursor, + type EventPollCursor, +} from "../event-cursor.js"; +import type { AgentEvent } from "@agentops/core"; +import { createEventId, EventCategory } from "@agentops/core"; + +function makeEvent(id: string, timestamp: string, overrides: Partial = {}): AgentEvent { + return { + id: createEventId(id), + category: EventCategory.Job, + type: "job.queued", + payload: { jobId: "job_1" }, + sourceId: "job_1", + timestamp, + ...overrides, + }; +} + +describe("event poll cursor", () => { + describe("advanceEventPollCursor (pure)", () => { + it("returns fresh events oldest-first from a newest-first batch", () => { + const cursor = createEventPollCursor("2025-01-01T00:00:00.000Z"); + const batch = [ + makeEvent("evt_2", "2025-01-01T00:00:02.000Z"), + makeEvent("evt_1", "2025-01-01T00:00:01.000Z"), + ]; + + const { fresh } = advanceEventPollCursor(cursor, batch); + expect(fresh.map((e) => e.id)).toEqual(["evt_1", "evt_2"]); + }); + + it("keeps the cursor unchanged on an empty batch", () => { + const cursor = createEventPollCursor("2025-01-01T00:00:00.000Z"); + const { fresh, next } = advanceEventPollCursor(cursor, []); + expect(fresh).toEqual([]); + expect(next).toBe(cursor); + }); + + it("does not re-emit the boundary event on the next poll", () => { + let cursor: EventPollCursor = createEventPollCursor("2025-01-01T00:00:00.000Z"); + const evt = makeEvent("evt_1", "2025-01-01T00:00:01.000Z"); + + const first = advanceEventPollCursor(cursor, [evt]); + expect(first.fresh).toHaveLength(1); + cursor = first.next; + + // A gte query re-matches evt_1 on every subsequent poll. + const second = advanceEventPollCursor(cursor, [evt]); + expect(second.fresh).toHaveLength(0); + expect(second.next).toBe(cursor); + }); + + it("emits a later event that shares the boundary timestamp exactly once", () => { + const ts = "2025-01-01T00:00:01.000Z"; + let cursor: EventPollCursor = createEventPollCursor("2025-01-01T00:00:00.000Z"); + const evtA = makeEvent("evt_a", ts); + const evtB = makeEvent("evt_b", ts); + + // Poll 1: only evt_a exists yet. + const first = advanceEventPollCursor(cursor, [evtA]); + expect(first.fresh.map((e) => e.id)).toEqual(["evt_a"]); + cursor = first.next; + + // Poll 2: evt_b arrived with the SAME timestamp. A bare gt cursor + // would drop it; gte + seen-ID tracking emits it once. + const second = advanceEventPollCursor(cursor, [evtB, evtA]); + expect(second.fresh.map((e) => e.id)).toEqual(["evt_b"]); + cursor = second.next; + + // Poll 3: nothing new — neither re-emits. + const third = advanceEventPollCursor(cursor, [evtB, evtA]); + expect(third.fresh).toHaveLength(0); + }); + + it("clears boundary IDs once the boundary advances past them", () => { + let cursor: EventPollCursor = createEventPollCursor("2025-01-01T00:00:00.000Z"); + const evtA = makeEvent("evt_a", "2025-01-01T00:00:01.000Z"); + cursor = advanceEventPollCursor(cursor, [evtA]).next; + + const evtB = makeEvent("evt_b", "2025-01-01T00:00:02.000Z"); + cursor = advanceEventPollCursor(cursor, [evtB, evtA]).next; + + expect(cursor.since).toBe("2025-01-01T00:00:02.000Z"); + expect(cursor.seenIdsAtBoundary.has("evt_b")).toBe(true); + expect(cursor.seenIdsAtBoundary.has("evt_a")).toBe(false); + }); + }); + + describe("integration with listEvents polling", () => { + let db: AgentOpsDb; + + beforeEach(() => { + db = getDb(":memory:"); + }); + + function poll(cursor: EventPollCursor) { + const batch = listEvents(db, { since: cursor.since, limit: 100 }); + return advanceEventPollCursor(cursor, batch); + } + + it("emits each event exactly once across repeated polls", () => { + let cursor = createEventPollCursor("2025-01-01T00:00:00.000Z"); + const emitted: string[] = []; + + insertEvent(db, makeEvent("evt_1", "2025-01-01T00:00:01.000Z")); + + // Several idle polls after the first event — the old timestamp-only + // cursor re-emitted the newest event on every one of these. + for (let i = 0; i < 3; i++) { + const { fresh, next } = poll(cursor); + emitted.push(...fresh.map((e) => e.id as string)); + cursor = next; + } + expect(emitted).toEqual(["evt_1"]); + + // Two more events, one sharing the boundary timestamp. + insertEvent(db, makeEvent("evt_2", "2025-01-01T00:00:01.000Z")); + insertEvent(db, makeEvent("evt_3", "2025-01-01T00:00:05.000Z")); + + for (let i = 0; i < 3; i++) { + const { fresh, next } = poll(cursor); + emitted.push(...fresh.map((e) => e.id as string)); + cursor = next; + } + expect(emitted).toEqual(["evt_1", "evt_2", "evt_3"]); + }); + }); +}); diff --git a/packages/db/src/__tests__/events.test.ts b/packages/db/src/__tests__/events.test.ts index f8c9fcf..f93d320 100644 --- a/packages/db/src/__tests__/events.test.ts +++ b/packages/db/src/__tests__/events.test.ts @@ -96,6 +96,17 @@ describe("Events repository", () => { expect(results).toHaveLength(2); }); + it("since is inclusive of the boundary timestamp", () => { + // Poll cursors (SSE route, CLI tail) rely on gte semantics so a + // second event sharing the boundary timestamp is never dropped. + insertEvent(db, makeEvent("evt_1", { timestamp: "2025-01-05T00:00:00.000Z" })); + insertEvent(db, makeEvent("evt_2", { timestamp: "2025-01-06T00:00:00.000Z" })); + + const results = listEvents(db, { since: "2025-01-05T00:00:00.000Z" }); + expect(results).toHaveLength(2); + expect(results.map((e) => e.id)).toContain("evt_1"); + }); + it("filters by until timestamp", () => { insertEvent(db, makeEvent("evt_1", { timestamp: "2025-01-01T00:00:00.000Z" })); insertEvent(db, makeEvent("evt_2", { timestamp: "2025-01-05T00:00:00.000Z" })); diff --git a/packages/db/src/event-cursor.ts b/packages/db/src/event-cursor.ts new file mode 100644 index 0000000..9f3302c --- /dev/null +++ b/packages/db/src/event-cursor.ts @@ -0,0 +1,67 @@ +import type { AgentEvent } from "@agentops/core"; + +/** + * Cursor for polling the events table without duplicates or gaps. + * + * `listEvents({ since })` is deliberately inclusive (gte): two events can + * share a timestamp, and an exclusive (gt) cursor would silently drop the + * second event if the first had already been emitted when the boundary + * advanced. The price of gte is that already-emitted events sitting exactly + * on the boundary timestamp re-match on every subsequent poll — this cursor + * tracks their IDs so each event is emitted exactly once. + */ +export interface EventPollCursor { + /** Boundary timestamp to pass as `since` (inclusive). */ + readonly since: string; + /** IDs of events already emitted whose timestamp equals `since`. */ + readonly seenIdsAtBoundary: ReadonlySet; +} + +export function createEventPollCursor(since: string): EventPollCursor { + return { since, seenIdsAtBoundary: new Set() }; +} + +/** + * Advance a cursor over one batch returned by + * `listEvents({ since: cursor.since, ... })`. + * + * `batch` is expected newest-first (listEvents' order). `fresh` contains + * only events not yet emitted, oldest-first — ready for chronological + * emission. `next` is the cursor to use for the following poll. + */ +export function advanceEventPollCursor( + cursor: EventPollCursor, + batch: ReadonlyArray, +): { fresh: AgentEvent[]; next: EventPollCursor } { + const fresh = [...batch] + .reverse() + .filter( + (e) => + !( + e.timestamp === cursor.since && + cursor.seenIdsAtBoundary.has(e.id as string) + ), + ); + + if (fresh.length === 0) { + return { fresh, next: cursor }; + } + + // New boundary = newest timestamp emitted. All batch events satisfy + // timestamp >= cursor.since and only boundary-timestamp events were + // filtered out, so the newest fresh event carries the batch maximum. + const nextSince = fresh[fresh.length - 1]!.timestamp; + + // Seed with the previous boundary's seen IDs when the boundary did not + // move — those events would still re-match a gte query at nextSince. + const seenIdsAtBoundary = new Set( + nextSince === cursor.since ? cursor.seenIdsAtBoundary : [], + ); + for (const e of fresh) { + if (e.timestamp === nextSince) { + seenIdsAtBoundary.add(e.id as string); + } + } + + return { fresh, next: { since: nextSince, seenIdsAtBoundary } }; +} diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 762babd..78f546e 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -77,6 +77,7 @@ export { insertSession, getSession, listSessions, updateSession, getActiveSessio // Event repository (WS3) export { insertEvent, getEvent, listEvents, countEvents, getEventsBySource, getRecentEvents } from "./events.js"; +export { createEventPollCursor, advanceEventPollCursor, type EventPollCursor } from "./event-cursor.js"; // Lock repository (WS4) export { insertLock, getLock, listLocks, updateLock, getActiveLocks, getActiveLocksForHolder, releaseLocksForHolder, releaseExpiredLocks } from "./locks.js"; diff --git a/packages/web/src/app/api/events/route.ts b/packages/web/src/app/api/events/route.ts index 5c423a3..366273f 100644 --- a/packages/web/src/app/api/events/route.ts +++ b/packages/web/src/app/api/events/route.ts @@ -1,6 +1,12 @@ import { NextRequest } from "next/server"; import { NextResponse } from "next/server"; -import { listRuns, getRun, listEvents } from "@agentops/db"; +import { + listRuns, + getRun, + listEvents, + createEventPollCursor, + advanceEventPollCursor, +} from "@agentops/db"; import { createRunId } from "@agentops/core"; import type { Run } from "@agentops/core"; import { db } from "@/lib/db"; @@ -39,13 +45,16 @@ export async function GET(request: NextRequest) { const categoryParam = request.nextUrl.searchParams.get("category"); const typeParam = request.nextUrl.searchParams.get("type"); - // Resolve view scope once at stream start. The user's owned - // sourceIds are captured here so the poll loop can filter every - // batch through the same set — without this an admin filtered to - // one user would still stream the team's events back. + // Resolve view scope at stream start. The user's owned sourceIds + // are captured here so the poll loop can filter every batch through + // the scoped set — without this an admin filtered to one user would + // still stream the team's events back. The ownership set is + // re-resolved on every poll (cheap indexed lookups) so runs and + // sessions created after connect appear in the owner's live stream + // without a reconnect. const scope = resolveViewScope(user, request.nextUrl.searchParams); - const scopedSourceIds = resolveOwnedSourceIds(scope.userId); - const scopedSourceIdSet = scopedSourceIds + let scopedSourceIds = resolveOwnedSourceIds(scope.userId); + let scopedSourceIdSet = scopedSourceIds ? new Set(scopedSourceIds) : null; @@ -56,7 +65,10 @@ export async function GET(request: NextRequest) { async start(controller) { // Snapshot: track known state const knownRuns = new Map(); - let lastEventTimestamp = new Date().toISOString(); + // The cursor pairs an inclusive (gte) `since` boundary with the + // IDs already emitted at that exact timestamp, so events sharing + // a timestamp are neither dropped nor re-emitted across polls. + let eventCursor = createEventPollCursor(new Date().toISOString()); function loadSnapshot() { if (runIdParam) { @@ -105,6 +117,15 @@ export async function GET(request: NextRequest) { } try { + // Refresh the ownership set so entities created after the + // stream opened are visible to their owner immediately. + if (scope.userId) { + scopedSourceIds = resolveOwnedSourceIds(scope.userId); + scopedSourceIdSet = scopedSourceIds + ? new Set(scopedSourceIds) + : null; + } + // Poll persisted events from events table const eventFilters: { since?: string; @@ -113,19 +134,23 @@ export async function GET(request: NextRequest) { sourceIds?: ReadonlyArray; limit?: number; } = { - since: lastEventTimestamp, + since: eventCursor.since, limit: 100, }; if (categoryParam) eventFilters.category = categoryParam; if (typeParam) eventFilters.type = typeParam; if (scopedSourceIds) eventFilters.sourceIds = scopedSourceIds; - const newEvents = listEvents(db(), eventFilters); - for (const evt of newEvents.reverse()) { + const batch = listEvents(db(), eventFilters); + const { fresh, next } = advanceEventPollCursor( + eventCursor, + batch, + ); + eventCursor = next; + for (const evt of fresh) { controller.enqueue( encoder.encode(sseMessage(evt.type, evt)), ); - lastEventTimestamp = evt.timestamp; } // Backward compat: still detect run changes via diffing diff --git a/packages/web/src/hooks/useEventSource.ts b/packages/web/src/hooks/useEventSource.ts index 7b41d2e..48fd913 100644 --- a/packages/web/src/hooks/useEventSource.ts +++ b/packages/web/src/hooks/useEventSource.ts @@ -35,6 +35,14 @@ interface UseEventSourceOptions { category?: string; /** Optional type filter for persisted events */ eventType?: string; + /** + * View scope forwarded to the SSE endpoint (e.g. "mine"). Without it + * the server resolves the connection's default scope, which for an + * admin is the whole team — not what a scoped page is showing. + */ + view?: string; + /** User scope forwarded to the SSE endpoint (admin drill-down). */ + userId?: string; /** Callback for each event received */ onEvent?: (event: SSEEvent) => void; } @@ -70,11 +78,14 @@ const ALL_EVENT_TYPES: SSEEventType[] = [ export function useEventSource( options: UseEventSourceOptions = {}, ): UseEventSourceReturn { - const { runId, category, eventType, onEvent } = options; + const { runId, category, eventType, view, userId, onEvent } = options; const [connected, setConnected] = useState(false); const [lastEvent, setLastEvent] = useState(null); const retriesRef = useRef(0); const eventSourceRef = useRef(null); + const reconnectTimerRef = useRef | null>( + null, + ); const onEventRef = useRef(onEvent); // Keep the latest onEvent in a ref without writing it during render // (react-hooks/refs). It's only read inside async SSE handlers that fire @@ -84,6 +95,13 @@ export function useEventSource( }, [onEvent]); const connect = useCallback(() => { + // Cancel any pending reconnect so a filter-change reconnect can't + // race a backoff timer into opening a second stream. + if (reconnectTimerRef.current !== null) { + clearTimeout(reconnectTimerRef.current); + reconnectTimerRef.current = null; + } + // Close any existing connection if (eventSourceRef.current) { eventSourceRef.current.close(); @@ -94,6 +112,8 @@ export function useEventSource( if (runId) params.set("runId", runId); if (category) params.set("category", category); if (eventType) params.set("type", eventType); + if (view) params.set("view", view); + if (userId) params.set("userId", userId); const qs = params.toString(); const url = qs ? `/api/events?${qs}` : "/api/events"; @@ -137,16 +157,23 @@ export function useEventSource( // Self-reference is runtime-safe: this async error handler only fires // after `connect` is fully defined, so the reconnect closes over the // assigned const. (react-hooks/immutability flags the textual order.) + // The handle is stored so unmount/filter-change cleanup can cancel + // the pending reconnect — otherwise it would open an EventSource + // nobody closes. // eslint-disable-next-line react-hooks/immutability - setTimeout(connect, delay); + reconnectTimerRef.current = setTimeout(connect, delay); } }; - }, [runId, category, eventType]); + }, [runId, category, eventType, view, userId]); useEffect(() => { connect(); return () => { + if (reconnectTimerRef.current !== null) { + clearTimeout(reconnectTimerRef.current); + reconnectTimerRef.current = null; + } if (eventSourceRef.current) { eventSourceRef.current.close(); eventSourceRef.current = null; diff --git a/packages/web/src/hooks/useEvents.ts b/packages/web/src/hooks/useEvents.ts index 9789a51..50fef06 100644 --- a/packages/web/src/hooks/useEvents.ts +++ b/packages/web/src/hooks/useEvents.ts @@ -2,6 +2,7 @@ import { useState, useCallback, useEffect } from "react"; import type { AgentEvent } from "@agentops/core"; +import { applyLiveEvent, type LiveEventState } from "@/lib/live-events"; import { useEventSource, type SSEEvent } from "./useEventSource"; interface UseEventsOptions { @@ -21,9 +22,11 @@ interface UseEventsReturn { export function useEvents(options: UseEventsOptions = {}): UseEventsReturn { const { category, type, userId, since } = options; - const [events, setEvents] = useState([]); + // Events list + total live in one state object so the SSE handler can + // update both through the pure applyLiveEvent reducer — the total must + // only grow when an event is genuinely new, not on every delivery. + const [live, setLive] = useState({ events: [], total: 0 }); const [loading, setLoading] = useState(false); - const [total, setTotal] = useState(0); useEffect(() => { let cancelled = false; @@ -40,8 +43,7 @@ export function useEvents(options: UseEventsOptions = {}): UseEventsReturn { const res = await fetch(`/api/events/list?${params.toString()}`); if (res.ok && !cancelled) { const data = await res.json(); - setEvents(data.events); - setTotal(data.total); + setLive({ events: data.events, total: data.total }); } } finally { if (!cancelled) setLoading(false); @@ -62,23 +64,23 @@ export function useEvents(options: UseEventsOptions = {}): UseEventsReturn { if (category && agentEvent.category !== category) return; if (type && agentEvent.type !== type) return; - setEvents((prev) => { - // Prepend new event if not already present - if (prev.some((e) => (e.id as string) === (agentEvent.id as string))) { - return prev; - } - return [agentEvent, ...prev]; - }); - setTotal((prev) => prev + 1); + // Prepend new event if not already present; the total only moves + // when the event was genuinely inserted (applyLiveEvent returns + // the previous state unchanged for duplicates). + setLive((prev) => applyLiveEvent(prev, agentEvent)); }, [category, type], ); + // Forward the user scope to the SSE endpoint — without it the server + // resolves an admin connection to the team-wide stream and the live + // feed mixes other users' events into a user-filtered page. const { connected } = useEventSource({ category, eventType: type, + userId, onEvent, }); - return { events, loading, connected, total }; + return { events: live.events, loading, connected, total: live.total }; } diff --git a/packages/web/src/hooks/useRuns.ts b/packages/web/src/hooks/useRuns.ts index 279281b..96d1205 100644 --- a/packages/web/src/hooks/useRuns.ts +++ b/packages/web/src/hooks/useRuns.ts @@ -83,6 +83,9 @@ export function useRuns( timeoutsRef.current.set(runId, timeout); }, []); + const scopeView = scope?.view ?? undefined; + const scopeUserId = scope?.userId ?? undefined; + const onEvent = useCallback( (event: SSEEvent) => { // Only handle run events (not AgentEvents from the event bus) @@ -90,6 +93,11 @@ export function useRuns( const run = event.data as Run; const runId = run.id as string; + // Belt-and-suspenders: the SSE URL already carries the view scope, + // but drop runs outside an explicit user filter in case a stale + // connection delivers unscoped data. + if (scopeUserId && run.userId !== scopeUserId) return; + setRunsWithSummaries((prev) => { if (event.type === "run_created") { // Prepend new run if not already present @@ -112,10 +120,17 @@ export function useRuns( markUpdated(runId); }, - [markUpdated], + [markUpdated, scopeUserId], ); - const { connected } = useEventSource({ onEvent }); + // Forward the view scope to the SSE connection. Without these params + // the server resolves an admin's stream to the team-wide view, mixing + // other users' live runs into a ?userId=X filtered page. + const { connected } = useEventSource({ + view: scopeView, + userId: scopeUserId, + onEvent, + }); // Cleanup timeouts on unmount useEffect(() => { diff --git a/packages/web/src/lib/__tests__/live-events.test.ts b/packages/web/src/lib/__tests__/live-events.test.ts new file mode 100644 index 0000000..3239a50 --- /dev/null +++ b/packages/web/src/lib/__tests__/live-events.test.ts @@ -0,0 +1,44 @@ +import { describe, it, expect } from "vitest"; +import { createEventId, EventCategory, type AgentEvent } from "@agentops/core"; +import { applyLiveEvent, type LiveEventState } from "@/lib/live-events"; + +function makeEvent(id: string): AgentEvent { + return { + id: createEventId(id), + category: EventCategory.Run, + type: "run.started", + payload: {}, + sourceId: "run_1", + timestamp: "2025-01-01T00:00:00.000Z", + }; +} + +describe("applyLiveEvent", () => { + it("prepends a new event and increments the total", () => { + const state: LiveEventState = { events: [makeEvent("evt_1")], total: 10 }; + const next = applyLiveEvent(state, makeEvent("evt_2")); + + expect(next.events.map((e) => e.id as string)).toEqual(["evt_2", "evt_1"]); + expect(next.total).toBe(11); + }); + + it("does not inflate the total when the event is a duplicate", () => { + const state: LiveEventState = { events: [makeEvent("evt_1")], total: 10 }; + + // The SSE server used to redeliver the newest event every 2s poll; + // the counter must not creep even if a duplicate slips through again. + let next = applyLiveEvent(state, makeEvent("evt_1")); + next = applyLiveEvent(next, makeEvent("evt_1")); + + expect(next).toBe(state); + expect(next.total).toBe(10); + expect(next.events).toHaveLength(1); + }); + + it("starts counting from an empty state", () => { + const empty: LiveEventState = { events: [], total: 0 }; + const next = applyLiveEvent(empty, makeEvent("evt_1")); + expect(next.events).toHaveLength(1); + expect(next.total).toBe(1); + }); +}); diff --git a/packages/web/src/lib/live-events.ts b/packages/web/src/lib/live-events.ts new file mode 100644 index 0000000..17a573d --- /dev/null +++ b/packages/web/src/lib/live-events.ts @@ -0,0 +1,28 @@ +import type { AgentEvent } from "@agentops/core"; + +/** + * State shared by the Events page list and its "N total" counter. + * Kept in one object so a live SSE event updates both atomically via a + * single pure reducer — the previous split-state version bumped the + * total on every SSE delivery even when the event was a duplicate, + * inflating the counter forever while connected. + */ +export interface LiveEventState { + readonly events: AgentEvent[]; + readonly total: number; +} + +/** + * Pure reducer: prepend a live event unless it is already present. + * The total only increments when the event was actually inserted. + */ +export function applyLiveEvent( + state: LiveEventState, + event: AgentEvent, +): LiveEventState { + const id = event.id as string; + if (state.events.some((e) => (e.id as string) === id)) { + return state; + } + return { events: [event, ...state.events], total: state.total + 1 }; +}