fix(events): dedupe SSE emission, scope live streams, fix reconnect leak

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);
 

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> = {}): 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"]);
+    });
+  });
+});

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" }));

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<string>;
+}
+
+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<AgentEvent>,
+): { 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<string>(
+    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 } };
+}

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";

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<string, Run>();
-        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<string>;
               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