web: coalesce live timeline refetches per burst instead of per SignalR signal

Agent uploads arrive as bursts of batches ~0.3 s apart with seconds of silence in
between, and every batch raises a SignalR signal. The 300 ms trailing debounce on the
session page either refetched once per batch or hid a tight burst until it paused;
session 46096bcd produced 155 full first-page reads for 261 batches in 13 minutes.

lib/burstScheduler: leading edge + trailing debounce + max wait. The first signal after
a quiet period refetches at once, later signals of a burst refetch once after the
burst's last signal, and a burst that never pauses still refetches every max-wait
(spool replay after reconnect/reboot). Fetch semantics are unchanged: every run reads
the full first page, the server stays the single source of truth, no incremental path.

Session page: trailing 300 ms, max wait 1 s. Progress portal (same pattern, two
requests per refresh): trailing 500 ms, max wait 1.5 s. scheduleFetchEvents loses its
delayMs parameter; the scheduler is created lazily on the first signal so refs stay
out of render (react-hooks/refs). Seven fake-timer tests pin the timing rules.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

src/Web/autopilot-monitor-web/app/progress/hooks/useProgressEvents.ts

@@ -3,9 +3,15 @@
 import { useCallback, useEffect, useRef, useState } from "react";
 import { api } from "@/lib/api";
 import { authenticatedFetch, TokenExpiredError } from "@/lib/authenticatedFetch";
+import { createBurstScheduler, type BurstScheduler } from "@/lib/burstScheduler";
 import { EnrollmentEvent, Session } from "@/types";
 import type { NotificationType } from "@/contexts/NotificationContext";
 
+// Live refetch coalescing on SignalR signals (see lib/burstScheduler). Each refresh is two
+// requests (summary lookup + events), hence a slightly wider window than the session page.
+const REFRESH_TRAILING_MS = 500;
+const REFRESH_MAX_WAIT_MS = 1_500;
+
 type AddNotification = (
   type: NotificationType,
   title: string,
@@ -26,14 +32,15 @@ export interface UseProgressEventsReturn {
   events: EnrollmentEvent[];
   setEvents: React.Dispatch<React.SetStateAction<EnrollmentEvent[]>>;
   sessionRef: React.RefObject<Session | null>;
-  scheduleFetchEvents: (delayMs?: number) => void;
+  /** Coalesced live refetch — call on every SignalR signal, the scheduler decides when to fetch. */
+  scheduleFetchEvents: () => void;
 }
 
 /**
  * Owns the progress page's event list lifecycle:
- *  - keeps a sessionRef in sync (used by SignalR hook + debounced refetch)
+ *  - keeps a sessionRef in sync (used by SignalR hook + coalesced refetch)
  *  - initial event fetch once per session (StrictMode-safe guard)
- *  - debounced scheduleFetchEvents that refreshes session summary + events
+ *  - burst-coalesced scheduleFetchEvents that refreshes session summary + events
  *  - merge-by-sequence dedup so repeat signals don't duplicate rows
  */
 export function useProgressEvents({
@@ -46,7 +53,10 @@ export function useProgressEvents({
   const [events, setEvents] = useState<EnrollmentEvent[]>([]);
   const sessionRef = useRef<Session | null>(null);
   const lastFetchedSessionId = useRef<string | null>(null);
-  const refetchTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
+  // The scheduler is created lazily on the first signal (refs stay out of render) and
+  // runs through refreshRef so it always sees the latest closure.
+  const refreshRef = useRef<() => Promise<void>>(async () => {});
+  const refreshScheduler = useRef<BurstScheduler | null>(null);
 
   useEffect(() => {
     sessionRef.current = session;
@@ -114,57 +124,65 @@ export function useProgressEvents({
     void run();
   }, [session, tenantId, getAccessToken, addNotification]);
 
-  const scheduleFetchEvents = useCallback(
-    (delayMs: number = 500) => {
-      if (refetchTimerRef.current) clearTimeout(refetchTimerRef.current);
-      refetchTimerRef.current = setTimeout(async () => {
-        const currentSession = sessionRef.current;
-        if (!currentSession) return;
-        try {
-          // Refresh the session summary via the serial lookup (the canonical serial from the
-          // session row always matches exactly). Guard on sessionId: the lookup returns the
-          // NEWEST session for the serial, which after a re-enrollment is a different session —
-          // the page must keep showing the one the user selected.
-          const lookupResponse = await authenticatedFetch(
-            api.progress.lookup(tenantId, currentSession.serialNumber),
-            getAccessToken,
-          );
-          if (lookupResponse.ok) {
-            const lookupData = await lookupResponse.json();
-            const updated: Session | null = lookupData.found ? lookupData.session : null;
-            if (updated && updated.sessionId === currentSession.sessionId) {
-              setSession(updated);
-            }
+  const refresh = useCallback(
+    async () => {
+      const currentSession = sessionRef.current;
+      if (!currentSession) return;
+      try {
+        // Refresh the session summary via the serial lookup (the canonical serial from the
+        // session row always matches exactly). Guard on sessionId: the lookup returns the
+        // NEWEST session for the serial, which after a re-enrollment is a different session —
+        // the page must keep showing the one the user selected.
+        const lookupResponse = await authenticatedFetch(
+          api.progress.lookup(tenantId, currentSession.serialNumber),
+          getAccessToken,
+        );
+        if (lookupResponse.ok) {
+          const lookupData = await lookupResponse.json();
+          const updated: Session | null = lookupData.found ? lookupData.session : null;
+          if (updated && updated.sessionId === currentSession.sessionId) {
+            setSession(updated);
           }
-
-          const eventsResponse = await authenticatedFetch(
-            api.progress.sessionEvents(currentSession.sessionId, tenantId, currentSession.serialNumber),
-            getAccessToken,
-          );
-          if (eventsResponse.ok) {
-            const eventsData = await eventsResponse.json();
-            const fetched: EnrollmentEvent[] = eventsData.events || [];
-            setEvents((prev) => {
-              const existingIds = new Set(prev.map((e) => e.eventId));
-              const newEvents = fetched.filter((e) => !existingIds.has(e.eventId));
-              if (newEvents.length === 0) return prev;
-              return [...prev, ...newEvents].sort(
-                (a, b) => a.sequence - b.sequence,
-              );
-            });
-          }
-        } catch (error) {
-          console.error("[Progress] Refetch failed:", error);
         }
-      }, delayMs);
+
+        const eventsResponse = await authenticatedFetch(
+          api.progress.sessionEvents(currentSession.sessionId, tenantId, currentSession.serialNumber),
+          getAccessToken,
+        );
+        if (eventsResponse.ok) {
+          const eventsData = await eventsResponse.json();
+          const fetched: EnrollmentEvent[] = eventsData.events || [];
+          setEvents((prev) => {
+            const existingIds = new Set(prev.map((e) => e.eventId));
+            const newEvents = fetched.filter((e) => !existingIds.has(e.eventId));
+            if (newEvents.length === 0) return prev;
+            return [...prev, ...newEvents].sort(
+              (a, b) => a.sequence - b.sequence,
+            );
+          });
+        }
+      } catch (error) {
+        console.error("[Progress] Refetch failed:", error);
+      }
     },
     [tenantId, getAccessToken, setSession],
   );
 
   useEffect(() => {
-    return () => {
-      if (refetchTimerRef.current) clearTimeout(refetchTimerRef.current);
-    };
+    refreshRef.current = refresh;
+  }, [refresh]);
+
+  const scheduleFetchEvents = useCallback(() => {
+    refreshScheduler.current ??= createBurstScheduler(
+      () => { void refreshRef.current(); },
+      { trailingMs: REFRESH_TRAILING_MS, maxWaitMs: REFRESH_MAX_WAIT_MS },
+    );
+    refreshScheduler.current.trigger();
+  }, []);
+
+  // Drop a pending trailing refetch on unmount
+  useEffect(() => {
+    return () => refreshScheduler.current?.cancel();
   }, []);
 
   return {

src/Web/autopilot-monitor-web/app/progress/hooks/useProgressSignalR.ts

@@ -28,7 +28,7 @@ interface UseProgressSignalRParams {
   session: Session | null;
   sessionRef: React.RefObject<Session | null>;
   signalR: SignalRApi;
-  scheduleFetchEvents: (delayMs?: number) => void;
+  scheduleFetchEvents: () => void;
   addNotification: AddNotification;
 }
 
@@ -94,7 +94,7 @@ export function useProgressSignalR({
     const scheduleRefetch = (source: string, sessionId: string) => {
       if (!sessionRef.current || sessionId !== sessionRef.current.sessionId) return;
       console.log(`[Progress] ${source} signal for current session, scheduling refetch`);
-      scheduleFetchEvents(500);
+      scheduleFetchEvents();
     };
 
     const handleNewEvents = (data: { sessionId: string }) => {

src/Web/autopilot-monitor-web/app/sessions/hooks/useSessionEvents.ts

@@ -3,6 +3,7 @@
 import { useCallback, useEffect, useRef, useState } from "react";
 import { api } from "@/lib/api";
 import { authenticatedFetch, TokenExpiredError } from "@/lib/authenticatedFetch";
+import { createBurstScheduler, type BurstScheduler } from "@/lib/burstScheduler";
 import { extractContinuation, MAX_EAGER_PAGES } from "@/lib/paginationLink";
 import { isGuid } from "@/utils/inputValidation";
 import { isTerminalStatus } from "@/utils/sessionStatus";
@@ -11,6 +12,14 @@ import type { NotificationType } from "@/contexts/NotificationContext";
 
 const TIMELINE_PAGE_SIZE = 200;
 
+// Live refetch coalescing on SignalR signals (see lib/burstScheduler): the first signal
+// of a burst refetches at once, later ones refetch 300 ms after the burst's last signal,
+// and a burst that never pauses still refetches every second. Agent uploads arrive as
+// bursts of batches ~0.3 s apart with seconds of silence in between, so this keeps the
+// real-time feel while a burst costs one or two page reads instead of one per batch.
+const EVENT_REFRESH_TRAILING_MS = 300;
+const EVENT_REFRESH_MAX_WAIT_MS = 1_000;
+
 // Single-shot refetch delay after the session transitions to a terminal status.
 // EnrollmentTerminationHandler emits trailing events (enrollment_summary_shown,
 // app_tracking_summary, diagnostics_collecting, diagnostics_uploaded) within ~5-10s
@@ -75,7 +84,8 @@ export interface UseSessionEventsReturn {
   events: EnrollmentEvent[];
   setEvents: React.Dispatch<React.SetStateAction<EnrollmentEvent[]>>;
   fetchEvents: () => Promise<void>;
-  scheduleFetchEvents: (delayMs?: number) => void;
+  /** Coalesced live refetch — call on every SignalR signal, the scheduler decides when to fetch. */
+  scheduleFetchEvents: () => void;
   /**
    * True while a Pattern-A eager-fetch is still streaming pages after the first
    * batch has rendered. Surfaces a "loading more events…" indicator on the
@@ -88,7 +98,7 @@ export interface UseSessionEventsReturn {
  * Owns the session detail page's event list lifecycle:
  *  - fetch events from Table Storage (canonical truth)
  *  - in-flight dedup (SignalR + 30s timer + group-join can overlap)
- *  - debounced scheduleFetchEvents to absorb bursts
+ *  - burst-coalesced scheduleFetchEvents (leading + trailing + max wait)
  *  - empty-refresh guard: ignores transient empty lists, keeps last known-good
  *  - terminal-event detection: triggers session refetch if SignalR status delta was lost
  *  - triggers initial fetch once sessionTenantId is known
@@ -109,14 +119,15 @@ export function useSessionEvents({
   const [events, setEvents] = useState<EnrollmentEvent[]>([]);
   const [isStreamingMore, setIsStreamingMore] = useState(false);
 
-  // Debounce real-time event refreshes to avoid burst reads in Table Storage.
-  const eventRefreshTimeoutRef = useRef<NodeJS.Timeout | null>(null);
   // Deduplication: track in-flight fetchEvents to avoid concurrent calls
   const fetchEventsInFlight = useRef(false);
   const fetchEventsQueued = useRef(false);
   // The queued follow-up re-invokes fetchEvents through this ref: a memoized callback
   // that calls itself directly cannot be memoized by the React compiler.
   const fetchEventsRef = useRef<() => Promise<void>>(async () => {});
+  // Created lazily on the first signal (refs stay out of render); runs through
+  // fetchEventsRef so it always calls the latest fetchEvents.
+  const eventRefreshScheduler = useRef<BurstScheduler | null>(null);
 
   const fetchEvents = useCallback(async () => {
     // Deduplication: if a fetch is already in flight, queue one follow-up instead of
@@ -244,14 +255,13 @@ export function useSessionEvents({
     fetchEventsRef.current = fetchEvents;
   }, [fetchEvents]);
 
-  const scheduleFetchEvents = useCallback((delayMs = 300) => {
-    if (eventRefreshTimeoutRef.current) {
-      clearTimeout(eventRefreshTimeoutRef.current);
-    }
-    eventRefreshTimeoutRef.current = setTimeout(() => {
-      fetchEvents();
-    }, delayMs);
-  }, [fetchEvents]);
+  const scheduleFetchEvents = useCallback(() => {
+    eventRefreshScheduler.current ??= createBurstScheduler(
+      () => { void fetchEventsRef.current(); },
+      { trailingMs: EVENT_REFRESH_TRAILING_MS, maxWaitMs: EVENT_REFRESH_MAX_WAIT_MS },
+    );
+    eventRefreshScheduler.current.trigger();
+  }, []);
 
   // Fetch events when we have the session's tenant ID
   useEffect(() => {
@@ -264,13 +274,9 @@ export function useSessionEvents({
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [sessionTenantId, sessionId]);
 
-  // Clear debounce timer on unmount
+  // Drop a pending trailing refetch on unmount
   useEffect(() => {
-    return () => {
-      if (eventRefreshTimeoutRef.current) {
-        clearTimeout(eventRefreshTimeoutRef.current);
-      }
-    };
+    return () => eventRefreshScheduler.current?.cancel();
   }, []);
 
   // One-shot trailing-events refetch on terminal transition.

src/Web/autopilot-monitor-web/app/sessions/hooks/useSessionSignalR.ts

@@ -25,7 +25,7 @@ interface UseSessionSignalRParams {
   sessionRef: React.RefObject<Session | null>;
   resolveEffectiveTenantId: () => string | null;
   signalR: SignalRApi;
-  scheduleFetchEvents: (delayMs?: number) => void;
+  scheduleFetchEvents: () => void;
   setSession: React.Dispatch<React.SetStateAction<Session | null>>;
   setSessionTenantId: React.Dispatch<React.SetStateAction<string | null>>;
   fetchAnalysisResults: (reanalyze?: boolean) => Promise<void>;
@@ -76,7 +76,7 @@ export function useSessionSignalR({
         // Re-fetch events after group join to catch any SignalR messages
         // that were sent before the client joined the session group.
         // The frontend deduplicates by eventId, so no duplicates.
-        scheduleFetchEvents(0);
+        scheduleFetchEvents();
       };
       joinAndCatchUp();
     }
@@ -109,7 +109,7 @@ export function useSessionSignalR({
       // Rule-result + tenant-id side effects still run.
       const status = sessionRef.current?.status;
       if (!isTerminalStatus(status)) {
-        // Fetch full events from storage (single source of truth), but debounce bursts.
+        // Fetch full events from storage (single source of truth); the scheduler coalesces bursts.
         // Session updates arrive via the "newevents" message (tenant group) — no session
         // object in this signal to keep payloads minimal.
         scheduleFetchEvents();

src/Web/autopilot-monitor-web/lib/__tests__/burstScheduler.test.ts

@@ -0,0 +1,98 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { createBurstScheduler } from "../burstScheduler";
+
+/**
+ * The scheduler decides WHEN the session timeline refetches on SignalR signals.
+ * Shape under test (trailing 300 ms, max wait 1000 ms — the session page's values):
+ *  - an isolated signal refetches at once (real-time feel),
+ *  - a short burst refetches at its start and once after it ends,
+ *  - a stream that never pauses refetches exactly once per second.
+ */
+
+const TRAILING = 300;
+const MAX_WAIT = 1000;
+
+function harness() {
+  const runs: number[] = [];
+  const t0 = Date.now();
+  const scheduler = createBurstScheduler(() => runs.push(Date.now() - t0), {
+    trailingMs: TRAILING,
+    maxWaitMs: MAX_WAIT,
+  });
+  return { runs, scheduler };
+}
+
+describe("createBurstScheduler", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("runs an isolated trigger immediately (leading edge)", () => {
+    const { runs, scheduler } = harness();
+    scheduler.trigger();
+    expect(runs).toEqual([0]);
+    vi.advanceTimersByTime(5_000);
+    expect(runs).toEqual([0]);
+  });
+
+  it("runs a short burst at its start and once after it ends", () => {
+    const { runs, scheduler } = harness();
+    scheduler.trigger(); // 0 ms
+    vi.advanceTimersByTime(200);
+    scheduler.trigger(); // 200 ms
+    vi.advanceTimersByTime(200);
+    scheduler.trigger(); // 400 ms
+    vi.advanceTimersByTime(5_000);
+    // 0 = leading, 700 = trailing (400 + 300); the 200-ms trigger was absorbed.
+    expect(runs).toEqual([0, 700]);
+  });
+
+  it("caps a stream that never pauses at one run per max wait", () => {
+    const { runs, scheduler } = harness();
+    for (let t = 0; t <= 3_000; t += 100) {
+      scheduler.trigger();
+      vi.advanceTimersByTime(100);
+    }
+    vi.advanceTimersByTime(5_000);
+    // Leading at 0, then max-wait bound at 1000/2000/3000, then the trailing run
+    // 300 ms after the last trigger (3000 + 300).
+    expect(runs).toEqual([0, 1000, 2000, 3000, 3300]);
+  });
+
+  it("does not run twice inside one trailing window right after a run", () => {
+    const { runs, scheduler } = harness();
+    scheduler.trigger(); // leading at 0
+    vi.advanceTimersByTime(100);
+    scheduler.trigger(); // 100 ms: cool-down, becomes a trailing run at 400
+    vi.advanceTimersByTime(5_000);
+    expect(runs).toEqual([0, 400]);
+  });
+
+  it("is leading again once the trailing window has been quiet", () => {
+    const { runs, scheduler } = harness();
+    scheduler.trigger(); // 0
+    vi.advanceTimersByTime(TRAILING);
+    scheduler.trigger(); // 300: quiet for a full trailing window since the last run
+    expect(runs).toEqual([0, 300]);
+  });
+
+  it("cancel drops the pending trailing run only", () => {
+    const { runs, scheduler } = harness();
+    scheduler.trigger(); // 0
+    vi.advanceTimersByTime(100);
+    scheduler.trigger(); // pending trailing at 400
+    scheduler.cancel();
+    vi.advanceTimersByTime(5_000);
+    expect(runs).toEqual([0]);
+    // Cancel is not a reset: a later trigger still follows the normal rules.
+    scheduler.trigger();
+    expect(runs).toEqual([0, 5_100]);
+  });
+
+  it("rejects a max wait shorter than the trailing window", () => {
+    expect(() => createBurstScheduler(() => {}, { trailingMs: 500, maxWaitMs: 100 })).toThrow();
+  });
+});

src/Web/autopilot-monitor-web/lib/burstScheduler.ts

@@ -0,0 +1,79 @@
+/**
+ * Burst scheduler for real-time refetches: leading edge + trailing debounce + max wait.
+ *
+ * Telemetry arrives in bursts (several agent batches within a second, then seconds of
+ * silence) and every batch raises a SignalR signal. A plain trailing debounce either
+ * fires once per batch (window shorter than the batch spacing) or hides the whole burst
+ * until it pauses (window longer than the spacing, timer keeps resetting). This scheduler
+ * keeps the live feel and still coalesces:
+ *
+ *  - leading: the first trigger after a quiet period runs immediately (0 ms delay);
+ *  - trailing: further triggers inside a burst run once, `trailingMs` after the last one,
+ *    so the refetch follows the END of the burst instead of a fixed tick;
+ *  - max wait: a burst that never pauses still runs at least every `maxWaitMs`
+ *    (spool replay after a reconnect or reboot).
+ *
+ * A steady stream of triggers spaced closer than `trailingMs` therefore runs exactly once
+ * per `maxWaitMs`; isolated triggers run at once. Pure timers, no React — the hooks own
+ * the "what to run" via a ref so the callback always sees the latest closure.
+ */
+export interface BurstSchedulerOptions {
+  /** Quiet time after the last trigger before the trailing run (ms). */
+  trailingMs: number;
+  /** Upper bound between two runs while triggers keep arriving (ms); >= trailingMs. */
+  maxWaitMs: number;
+}
+
+export interface BurstScheduler {
+  /** Request a run; coalesced per the rules above. */
+  trigger: () => void;
+  /** Drop a pending trailing run (unmount). Does not reset the leading-edge cool-down. */
+  cancel: () => void;
+}
+
+export function createBurstScheduler(run: () => void, options: BurstSchedulerOptions): BurstScheduler {
+  const { trailingMs, maxWaitMs } = options;
+  if (!(trailingMs >= 0) || !(maxWaitMs >= trailingMs)) {
+    throw new Error(`createBurstScheduler: need 0 <= trailingMs <= maxWaitMs, got ${trailingMs}/${maxWaitMs}`);
+  }
+
+  let timer: ReturnType<typeof setTimeout> | null = null;
+  // Time of the last run; the max-wait window for the next trailing run starts here.
+  let lastRunAt = Number.NEGATIVE_INFINITY;
+
+  const fire = () => {
+    timer = null;
+    lastRunAt = Date.now();
+    run();
+  };
+
+  const scheduleTrailing = (now: number) => {
+    if (timer) clearTimeout(timer);
+    const fireAt = Math.min(now + trailingMs, lastRunAt + maxWaitMs);
+    timer = setTimeout(fire, Math.max(0, fireAt - now));
+  };
+
+  return {
+    trigger: () => {
+      const now = Date.now();
+      if (timer) {
+        // Inside a burst: push the trailing run out, but never past the max-wait bound.
+        scheduleTrailing(now);
+        return;
+      }
+      if (now - lastRunAt >= trailingMs) {
+        // Quiet long enough: leading edge, run right away.
+        fire();
+        return;
+      }
+      // Just ran: cool down into a trailing run so a burst cannot run twice per trailingMs.
+      scheduleTrailing(now);
+    },
+    cancel: () => {
+      if (timer) {
+        clearTimeout(timer);
+        timer = null;
+      }
+    },
+  };
+}