refactor(events): make grouped-event-buffer self-notify

Move bufferVersion ownership into the grouped-event-buffer and bump it
from invalidateCache() — the single mutation choke point — coalescing
many synchronous ingests into one notification via a microtask.

- events.ts re-exports bufferVersion instead of owning it
- workflow-run-layout drops all manual bufferVersion bumps; live-event
  reactivity now flows through appendLiveEvent -> invalidateCache
- clearWorkflowData resets the buffer (clearing stale derived state)
  instead of poking the version counter
- remove the now-dead onNewEvents hook from live-poll and its tests

Author: tegan-temporal

src/lib/components/lines-and-dots/workflow-details.svelte

@@ -81,7 +81,7 @@
       : '',
   );
   let totalActions = $derived.by(() => {
-    $bufferVersion;
+    void $bufferVersion;
     return getEventArray()
       .reduce((acc, e) => (e?.billableActions ?? 0) + acc, 0)
       .toString();

src/lib/layouts/workflow-run-layout.svelte

@@ -60,7 +60,7 @@
   let runId = $derived(page.params.run);
   let showJson = $derived(page.url.searchParams.has('json'));
   let fullJson = $derived.by(() => {
-    $bufferVersion;
+    void $bufferVersion;
     return { ...$workflowRun, eventHistory: getEventArray() };
   });
 
@@ -160,7 +160,6 @@
           latestEventId = Math.max(latestEventId, parseInt(ev.eventId));
         return isNew;
       },
-      onNewEvents: () => bufferVersion.update((v) => v + 1),
     }).then((lastToken) => {
       _lastPollToken = lastToken;
     });
@@ -243,7 +242,6 @@
           const id = parseInt(event.eventId);
           if (id > latestEventId) latestEventId = id;
         }
-        if (events.length) bufferVersion.update((v) => v + 1);
       },
     })
       .then(() => {
@@ -252,7 +250,6 @@
           $workflowRun.workflow?.pendingNexusOperations ?? [],
         );
         fetchComplete = true;
-        bufferVersion.update((v) => v + 1);
       })
       .catch((e: unknown) => {
         if (e instanceof Error && e.name !== 'AbortError') {
@@ -290,7 +287,7 @@
   };
 
   $effect(() => {
-    $bufferVersion;
+    void $bufferVersion;
     untrack(() => {
       const events = getEventArray();
       $fullEventHistory = events;
@@ -301,7 +298,6 @@
     $timelineEvents = null;
     $workflowRun = initialWorkflowRun;
     $fullEventHistory = [];
-    $bufferVersion = 0;
     workflowError = null;
     fetchComplete = false;
     latestEventId = 0;
@@ -312,13 +308,14 @@
     _lastPollToken = '';
     _pollPaused = false;
     abortAll();
+    resetBuffer();
     resetLastDataEncoderSuccess();
     if (refreshInterval) clearInterval(refreshInterval);
     refreshInterval = null;
   };
 
   $effect(() => {
-    runId;
+    void runId;
     untrack(() => {
       clearWorkflowData();
     });

src/lib/services/grouped-event-buffer.ts

@@ -1,3 +1,5 @@
+import { writable } from 'svelte/store';
+
 import {
   addEventToGroup,
   createEventGroup,
@@ -56,6 +58,27 @@ const processedWorkflowTaskIds = new Set<string>();
 let cachedGroups: EventGroup[] | null = null;
 let cachedGroupsNoWFT: EventGroup[] | null = null;
 
+/**
+ * Reactivity signal for the buffer. invalidateCache() — the single choke point
+ * hit by every mutation (ingest, enrichGroups, reset) — schedules a coalesced
+ * bump, so many synchronous ingests (a whole fetch page or live-poll response)
+ * notify subscribers once. Svelte consumers read this in a derived/effect and
+ * then call getEventArray() / getGroupArray().
+ */
+export const bufferVersion = writable(0);
+
+let bumpScheduled = false;
+function scheduleBufferChange(): void {
+  if (bumpScheduled) {
+    return;
+  }
+  bumpScheduled = true;
+  queueMicrotask(() => {
+    bumpScheduled = false;
+    bufferVersion.update((version) => version + 1);
+  });
+}
+
 function toWorkflowEvent(
   raw: HistoryEvent,
   isAscending: boolean,
@@ -79,6 +102,7 @@ function insertEventById(list: WorkflowEvent[], event: WorkflowEvent): void {
 function invalidateCache(): void {
   cachedGroups = null;
   cachedGroupsNoWFT = null;
+  scheduleBufferChange();
 }
 
 function isWorkflowTaskGroup(group: EventGroup): boolean {

src/lib/services/live-poll.test.ts

@@ -66,7 +66,6 @@ function startPoll(
 ) {
   const ctrl = new AbortController();
   const onEvent = vi.fn().mockReturnValue(onEventReturn);
-  const onNewEvents = vi.fn();
 
   let callIdx = 0;
   requestFromAPI.mockImplementation(() => {
@@ -81,11 +80,10 @@ function startPoll(
     startToken: 'start-tok',
     signal: ctrl.signal,
     onEvent,
-    onNewEvents,
     ...FAST,
   });
 
-  return { ctrl, onEvent, onNewEvents, done };
+  return { ctrl, onEvent, done };
 }
 
 // ---------------------------------------------------------------------------
@@ -106,7 +104,6 @@ describe('runLivePoll — waitNewEvent long-poll', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn().mockReturnValue(false),
-      onNewEvents: vi.fn(),
       ...FAST,
     });
 
@@ -131,7 +128,6 @@ describe('runLivePoll — waitNewEvent long-poll', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn().mockReturnValue(false),
-      onNewEvents: vi.fn(),
       ...FAST,
     });
 
@@ -156,7 +152,6 @@ describe('runLivePoll — waitNewEvent long-poll', () => {
       startToken: 'initial-cursor',
       signal: ctrl.signal,
       onEvent: vi.fn().mockReturnValue(false),
-      onNewEvents: vi.fn(),
       ...FAST,
     });
 
@@ -179,7 +174,6 @@ describe('runLivePoll — waitNewEvent long-poll', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn().mockReturnValue(false),
-      onNewEvents: vi.fn(),
       ...FAST,
     });
 
@@ -204,28 +198,6 @@ describe('runLivePoll — event delivery', () => {
     expect(onEvent).toHaveBeenNthCalledWith(2, makeRawEvent(2));
     expect(onEvent).toHaveBeenNthCalledWith(3, makeRawEvent(3));
   });
-
-  it('calls onNewEvents once when at least one event is new', async () => {
-    const { onNewEvents, done } = startPoll([pollResponse([1, 2])], {
-      onEventReturn: true,
-    });
-    await done;
-    expect(onNewEvents).toHaveBeenCalledTimes(1);
-  });
-
-  it('does NOT call onNewEvents when all events are duplicates', async () => {
-    const { onNewEvents, done } = startPoll([pollResponse([1, 2])], {
-      onEventReturn: false,
-    });
-    await done;
-    expect(onNewEvents).not.toHaveBeenCalled();
-  });
-
-  it('does NOT call onNewEvents when response has no events', async () => {
-    const { onNewEvents, done } = startPoll([pollResponse([])]);
-    await done;
-    expect(onNewEvents).not.toHaveBeenCalled();
-  });
 });
 
 // ---------------------------------------------------------------------------
@@ -235,7 +207,6 @@ describe('runLivePoll — event delivery', () => {
 describe('runLivePoll — nextPageToken threading', () => {
   it('follows nextPageToken immediately without delay', async () => {
     const ctrl = new AbortController();
-    const onNewEvents = vi.fn();
     let callIdx = 0;
     const responses = [
       pollResponse([1, 2], 'page-2-tok'),
@@ -253,7 +224,6 @@ describe('runLivePoll — nextPageToken threading', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn().mockReturnValue(true),
-      onNewEvents,
       ...FAST,
     });
 
@@ -285,7 +255,6 @@ describe('runLivePoll — nextPageToken threading', () => {
       startToken: 'initial',
       signal: ctrl.signal,
       onEvent: vi.fn().mockReturnValue(true),
-      onNewEvents: vi.fn(),
       ...FAST,
     });
 
@@ -297,31 +266,6 @@ describe('runLivePoll — nextPageToken threading', () => {
       expect.objectContaining({ token: undefined }),
     );
   });
-
-  it('calls onNewEvents per iteration, not once per event batch', async () => {
-    // Two iterations, each delivering new events.
-    const ctrl = new AbortController();
-    let callIdx = 0;
-    const responses = [pollResponse([1]), pollResponse([2])];
-    requestFromAPI.mockImplementation(() => {
-      const resp = responses[callIdx++];
-      if (callIdx >= responses.length) ctrl.abort();
-      return Promise.resolve(resp);
-    });
-    const onNewEvents = vi.fn();
-
-    await runLivePoll({
-      route: '/api/events',
-      runId: 'run-1',
-      startToken: '',
-      signal: ctrl.signal,
-      onEvent: vi.fn().mockReturnValue(true),
-      onNewEvents,
-      ...FAST,
-    });
-
-    expect(onNewEvents).toHaveBeenCalledTimes(2);
-  });
 });
 
 // ---------------------------------------------------------------------------
@@ -348,7 +292,6 @@ describe('runLivePoll — back-off', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn().mockReturnValue(false),
-      onNewEvents: vi.fn(),
       backoffMs: 0,
       errorBackoffMs: 0,
     });
@@ -357,7 +300,7 @@ describe('runLivePoll — back-off', () => {
   });
 
   it('does NOT delay when response delivers new events without nextPageToken', async () => {
-    // Abort inside onNewEvents so the loop exits right after processing events
+    // Abort while processing the event so the loop exits right after the batch
     // but before any setTimeout could fire. If backoffMs=999 appears in delays
     // it means the code incorrectly backed off even though events were delivered.
     const delays: number[] = [];
@@ -372,16 +315,18 @@ describe('runLivePoll — back-off', () => {
     const ctrl = new AbortController();
     requestFromAPI.mockResolvedValueOnce(pollResponse([1])); // events delivered, no token
 
-    // Abort during onNewEvents — happens before the setTimeout check in the loop.
-    const onNewEvents = vi.fn().mockImplementation(() => ctrl.abort());
+    // Abort during event processing — happens before the setTimeout check in the loop.
+    const onEvent = vi.fn().mockImplementation(() => {
+      ctrl.abort();
+      return true;
+    });
 
     await runLivePoll({
       route: '/api/events',
       runId: 'run-1',
       startToken: '',
       signal: ctrl.signal,
-      onEvent: vi.fn().mockReturnValue(true),
-      onNewEvents,
+      onEvent,
       backoffMs: 999, // must NOT appear in delays
       errorBackoffMs: 0,
     });
@@ -414,7 +359,6 @@ describe('runLivePoll — back-off', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn().mockReturnValue(false),
-      onNewEvents: vi.fn(),
       backoffMs: 1234, // distinctive value
       errorBackoffMs: 0,
     });
@@ -447,7 +391,6 @@ describe('runLivePoll — back-off', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn().mockReturnValue(false),
-      onNewEvents: vi.fn(),
       backoffMs: 0,
       errorBackoffMs: 5678, // distinctive value
     });
@@ -484,7 +427,6 @@ describe('runLivePoll — back-off', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn().mockReturnValue(false),
-      onNewEvents: vi.fn(),
       // No backoffMs / errorBackoffMs → use defaults
     });
 
@@ -511,7 +453,6 @@ describe('runLivePoll — abort', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn(),
-      onNewEvents: vi.fn(),
       ...FAST,
     });
 
@@ -531,7 +472,6 @@ describe('runLivePoll — abort', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn(),
-      onNewEvents: vi.fn(),
       errorBackoffMs: 0,
       backoffMs: 0,
     });
@@ -561,7 +501,6 @@ describe('runLivePoll — pause/resume cursor', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn(),
-      onNewEvents: vi.fn(),
       ...FAST,
     });
 
@@ -581,7 +520,6 @@ describe('runLivePoll — pause/resume cursor', () => {
       startToken: 'resume-tok',
       signal: ctrl.signal,
       onEvent: vi.fn(),
-      onNewEvents: vi.fn(),
       ...FAST,
     });
 
@@ -607,7 +545,6 @@ describe('runLivePoll — pause/resume cursor', () => {
       startToken: '',
       signal: ctrl.signal,
       onEvent: vi.fn(),
-      onNewEvents: vi.fn(),
       ...FAST,
     });
 
@@ -632,7 +569,6 @@ describe('runLivePoll — pause/resume cursor', () => {
       startToken: '',
       signal: ctrl1.signal,
       onEvent: vi.fn().mockReturnValue(true),
-      onNewEvents: vi.fn(),
       ...FAST,
     });
 
@@ -650,7 +586,6 @@ describe('runLivePoll — pause/resume cursor', () => {
       startToken: resumeToken,
       signal: ctrl2.signal,
       onEvent: onEvent2,
-      onNewEvents: vi.fn(),
       ...FAST,
     });