fix(explorer): dedupe temporal snapshot requests and apply latest-wins (#1241)

fix(explorer): dedupe temporal snapshot requests and apply latest-wins

The temporal snapshot effect fetched /api/temporal/snapshot with no
idempotency or ordering guards. Upstream churn (timeline recreation
while bounds settle, play ticks resetting the playhead, drag events)
could re-request the same `at` repeatedly, and with variable network
latency an older position's response could land after a newer one's,
overwriting the active-node count, so the chip visibly lagged the
scrubber.

Add a small stateful guard module (temporalSnapshotGuards.ts) built
around a per-position cache, keyed by the debounced timestamp's
primitive millisecond value rather than the Date object, so upstream
object-identity churn cannot defeat the dedup on its own:

- at most one in-flight request per scrubber position, so identical
  `at` values arriving while a request is pending are dropped instead
  of firing a fresh fetch, breaking the idle/play polling loop;
- successful snapshots are cached per position and re-applied when the
  scrubber returns to it (play wrap-around, back-scrubbing) without a
  network round trip;
- a response is applied only while the scrubber is still on the
  position it was requested for, so an out-of-order response can never
  clobber a newer position's count;
- failed, cancelled, or superseded requests release their position so
  it can be fetched again the next time it's visited, rather than
  stalling it permanently;
- reset() drops all cached and in-flight state when the underlying
  graph summary changes (reload/retry), since snapshots cached against
  the previous graph no longer describe anything real. Keyed on the
  summary query's data identity, which react-query keeps stable
  (staleTime: Infinity plus structural sharing) unless the graph data
  itself was replaced, so reset fires exactly on a real reload and not
  on cosmetic re-renders.

The snapshot effect is wired through the guards end to end: begin()
returns either a fresh sequence number to fetch under or a cached
snapshot to reapply directly; the same shouldApply()/apply() gate
handles both the network and cached-reapply paths so they can't drift
apart; finish() runs from both the fetch's failure branch and its
cleanup function, so a cancelled or failed request is always retryable
on the next visit instead of leaving its position stuck in-flight.

16 unit tests cover dedup, independent positions, revisit re-apply,
play wrap-around, failure retry, stale-sequence protection (a late
response or a late release from a superseded request cannot act on a
newer request's position), reset-on-reload, and cache-bound eviction.

Closes #1128

explorer/src/workspaces/GraphWorkspace/GraphWorkspace.tsx

@@ -41,6 +41,7 @@ import {
 } from "./plugins";
 import { explorationEffectsShouldLoad, neighborhoodPanelShouldLoad, temporalOverlayShouldLoad } from "./pluginRegistryPredicates";
 import { shouldFetchTemporalBounds, shouldFetchTemporalSnapshot } from "./temporalLifecyclePredicates";
+import { createTemporalSnapshotGuards, type TemporalSnapshotResponse } from "./temporalSnapshotGuards";
 import type { LinkPrediction, PathResponse } from "./GraphInspectorPanel";
 import type { GraphSceneHandle, GraphSceneRuntime } from "./scene";
 import type {
@@ -1479,6 +1480,23 @@ export function GraphWorkspace({ externalFocusNodeId, externalFocusToken }: Grap
     summary?.edgeCount,
   ]);
 
+  // Guards the snapshot lifecycle: at most one in-flight request per scrubber
+  // position (identical-`at` polls are deduplicated, breaking the idle/play
+  // polling loop), applied snapshots are cached and re-applied on revisit, and
+  // a response applies only while the scrubber is still on its position
+  // (out-of-order responses cannot clobber the active-node count).
+  const temporalSnapshotGuardsRef = useRef<ReturnType<typeof createTemporalSnapshotGuards> | null>(null);
+  if (temporalSnapshotGuardsRef.current === null) {
+    temporalSnapshotGuardsRef.current = createTemporalSnapshotGuards();
+  }
+  const temporalSnapshotGuards = temporalSnapshotGuardsRef.current;
+
+  // A new graph summary means the graph data was replaced (reload/retry);
+  // snapshots cached against the previous graph are stale, so reset all state.
+  useEffect(() => {
+    temporalSnapshotGuards.reset();
+  }, [summary]);
+
   useEffect(() => {
     if (!canFetchTemporalSnapshot) {
       return;
@@ -1488,20 +1506,25 @@ export function GraphWorkspace({ externalFocusNodeId, externalFocusToken }: Grap
       return;
     }
 
+    const atMs = debouncedTime.getTime();
+    const { seq, cached } = temporalSnapshotGuards.begin(atMs);
+    if (seq === null) {
+      // An identical request is already in flight: one request per position.
+      return;
+    }
+
     let cancelled = false;
 
-    const applySnapshot = async () => {
-      try {
-        const at = debouncedTime.toISOString();
-        const response = await fetch(`/api/temporal/snapshot?at=${encodeURIComponent(at)}`);
-        if (!response.ok || cancelled) return;
-
-        const data: { active_node_ids: string[]; active_node_count: number } = await response.json();
-        if (cancelled) return;
-
+    const applyData = (data: TemporalSnapshotResponse) => {
       const nextActiveIds = new Set(data.active_node_ids);
       requestAnimationFrame(() => {
         if (cancelled) return;
+        if (!temporalSnapshotGuards.shouldApply(atMs, seq)) {
+          // The scrubber moved on (or this request was superseded): release the
+          // position so a return to it refetches instead of stalling.
+          temporalSnapshotGuards.finish(atMs, seq);
+          return;
+        }
         const previous = prevActiveIdsRef.current;
         previous.forEach((id) => {
           if (!nextActiveIds.has(id) && graph.hasNode(id)) {
@@ -1517,8 +1540,33 @@ export function GraphWorkspace({ externalFocusNodeId, externalFocusToken }: Grap
         setActiveNodeCount(data.active_node_count);
         setGraphVersion((current) => current + 1);
         sceneRef.current?.getRuntime()?.requestRender();
+        temporalSnapshotGuards.apply(atMs, seq, data);
       });
+    };
+
+    if (cached) {
+      // Returning to a position whose snapshot was already applied: re-apply
+      // the cached result without a network request.
+      applyData(cached);
+      return;
+    }
+
+    const applySnapshot = async () => {
+      try {
+        const at = debouncedTime.toISOString();
+        const response = await fetch(`/api/temporal/snapshot?at=${encodeURIComponent(at)}`);
+        if (!response.ok) {
+          // A failed request must be retryable if the scrubber returns.
+          if (!cancelled) temporalSnapshotGuards.finish(atMs, seq);
+          return;
+        }
+        if (cancelled) return;
+
+        const data: TemporalSnapshotResponse = await response.json();
+        if (cancelled) return;
+        applyData(data);
       } catch (fetchError) {
+        temporalSnapshotGuards.finish(atMs, seq);
         if (!cancelled) {
           console.error("[Temporal] Snapshot fetch failed", fetchError);
         }
@@ -1528,6 +1576,8 @@ export function GraphWorkspace({ externalFocusNodeId, externalFocusToken }: Grap
     applySnapshot();
     return () => {
       cancelled = true;
+      // A cancelled request must be retryable when its position is revisited.
+      temporalSnapshotGuards.finish(atMs, seq);
     };
   }, [
     canFetchTemporalSnapshot,

explorer/src/workspaces/GraphWorkspace/temporalSnapshotGuards.ts

@@ -0,0 +1,113 @@
+/**
+ * Guards for the temporal snapshot fetch/apply lifecycle.
+ *
+ * The snapshot effect previously fetched /api/temporal/snapshot with no
+ * idempotency or ordering protection. Upstream churn (timeline recreation
+ * while bounds settle, play ticks resetting the playhead, drag events) could
+ * re-request the same `at` repeatedly, and responses could arrive after the
+ * scrubber had moved on.
+ *
+ * The guards enforce:
+ * - at most one in-flight request per scrubber position (identical `at`
+ *   values are deduplicated while a request is pending, breaking the
+ *   idle/play polling loop);
+ * - successful snapshots are cached per position and re-applied when the
+ *   scrubber returns (play wrap-around, back-scrubbing) without a refetch;
+ * - a response is applied only while the scrubber is still on its position,
+ *   so out-of-order responses cannot clobber a newer position's count;
+ * - failed, cancelled, or superseded requests release their position so it
+ *   can be fetched again on the next visit;
+ * - `reset()` drops all state when the underlying graph data is replaced
+ *   (reload/retry), because cached snapshots describe the previous graph.
+ *
+ * `createTemporalSnapshotGuards()` is stateful by design.
+ */
+
+export interface TemporalSnapshotResponse {
+  active_node_ids: string[];
+  active_node_count: number;
+}
+
+export interface TemporalSnapshotRequest {
+  /** null when the request was deduplicated because one is already in flight. */
+  seq: number | null;
+  /** The snapshot previously applied for this position, when revisiting it. */
+  cached: TemporalSnapshotResponse | null;
+}
+
+export interface TemporalSnapshotGuards {
+  /** Begin (or dedupe) a request for `atMs`; marks it as the current position. */
+  begin(atMs: number): TemporalSnapshotRequest;
+  /** True when the response for `atMs`/`seq` may be applied (scrubber still on `atMs`). */
+  shouldApply(atMs: number, seq: number): boolean;
+  /** Record a successful application and cache its snapshot for revisits. */
+  apply(atMs: number, seq: number, data: TemporalSnapshotResponse): void;
+  /** Release a position whose request failed, was cancelled, or was superseded. */
+  finish(atMs: number, seq: number): void;
+  /** Drop all state; call when the underlying graph data is replaced (reload). */
+  reset(): void;
+}
+
+interface SnapshotEntry {
+  seq: number;
+  /** null while the request is in flight (or before the first success). */
+  data: TemporalSnapshotResponse | null;
+}
+
+/** Upper bound on cached positions so long scrubbing sessions stay bounded. */
+const MAX_CACHED_POSITIONS = 256;
+
+export function createTemporalSnapshotGuards(): TemporalSnapshotGuards {
+  const entries = new Map<number, SnapshotEntry>();
+  let latestRequestSeq = 0;
+  let currentAtMs: number | null = null;
+
+  const evictOldest = () => {
+    while (entries.size > MAX_CACHED_POSITIONS) {
+      const oldestAtMs = entries.keys().next().value;
+      if (oldestAtMs === undefined) return;
+      entries.delete(oldestAtMs);
+    }
+  };
+
+  return {
+    begin(atMs) {
+      const existing = entries.get(atMs);
+      if (existing && existing.data === null) {
+        // Identical request already in flight: dedupe, but the scrubber is here now.
+        currentAtMs = atMs;
+        return { seq: null, cached: null };
+      }
+      latestRequestSeq += 1;
+      const seq = latestRequestSeq;
+      entries.set(atMs, { seq, data: existing?.data ?? null });
+      currentAtMs = atMs;
+      evictOldest();
+      return { seq, cached: existing?.data ?? null };
+    },
+
+    shouldApply(atMs, seq) {
+      return atMs === currentAtMs && entries.get(atMs)?.seq === seq;
+    },
+
+    apply(atMs, seq, data) {
+      const entry = entries.get(atMs);
+      if (entry && entry.seq === seq) {
+        entry.data = data;
+      }
+    },
+
+    finish(atMs, seq) {
+      const entry = entries.get(atMs);
+      if (entry && entry.seq === seq && entry.data === null) {
+        entries.delete(atMs);
+      }
+    },
+
+    reset() {
+      entries.clear();
+      latestRequestSeq = 0;
+      currentAtMs = null;
+    },
+  };
+}

explorer/tests/temporalSnapshotGuards.test.ts

@@ -0,0 +1,150 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+
+import { createTemporalSnapshotGuards } from "../src/workspaces/GraphWorkspace/temporalSnapshotGuards.ts";
+
+const POSITION_1 = new Date("2023-07-02T00:00:00Z").getTime();
+const POSITION_2 = new Date("2024-01-02T00:00:00Z").getTime();
+const POSITION_3 = new Date("2024-07-02T00:00:00Z").getTime();
+
+const SNAPSHOT = { active_node_ids: ["n1", "n2"], active_node_count: 2 };
+
+// ── begin: one request per scrubber position ─────────────────────────────────
+
+test("begin: a new position returns a fresh request sequence", () => {
+  const guards = createTemporalSnapshotGuards();
+  assert.deepEqual(guards.begin(POSITION_1), { seq: 1, cached: null });
+});
+
+test("begin: an identical in-flight request is deduplicated (no duplicate fetch)", () => {
+  const guards = createTemporalSnapshotGuards();
+  guards.begin(POSITION_1);
+  assert.deepEqual(guards.begin(POSITION_1), { seq: null, cached: null });
+});
+
+test("begin: distinct positions request independently", () => {
+  const guards = createTemporalSnapshotGuards();
+  assert.equal(guards.begin(POSITION_1).seq, 1);
+  assert.equal(guards.begin(POSITION_2).seq, 2);
+});
+
+test("begin: revisiting an applied position returns its cached snapshot", () => {
+  const guards = createTemporalSnapshotGuards();
+  const { seq } = guards.begin(POSITION_1);
+  guards.apply(POSITION_1, seq, SNAPSHOT);
+  const revisit = guards.begin(POSITION_1);
+  assert.equal(revisit.seq, 2);
+  assert.deepEqual(revisit.cached, SNAPSHOT);
+});
+
+test("begin: a failed position (finished) can be requested again", () => {
+  const guards = createTemporalSnapshotGuards();
+  const { seq } = guards.begin(POSITION_1);
+  guards.finish(POSITION_1, seq);
+  const retry = guards.begin(POSITION_1);
+  assert.equal(retry.seq, 2);
+  assert.equal(retry.cached, null);
+});
+
+test("finish: does not clear a position whose snapshot was already applied", () => {
+  const guards = createTemporalSnapshotGuards();
+  const { seq } = guards.begin(POSITION_1);
+  guards.apply(POSITION_1, seq, SNAPSHOT);
+  guards.finish(POSITION_1, seq);
+  assert.deepEqual(guards.begin(POSITION_1).cached, SNAPSHOT);
+});
+
+test("finish: a stale sequence cannot release a newer request's position", () => {
+  const guards = createTemporalSnapshotGuards();
+  const first = guards.begin(POSITION_1);
+  guards.finish(POSITION_1, first.seq);
+  guards.begin(POSITION_1); // seq 2, in flight again
+  guards.finish(POSITION_1, first.seq); // stale seq: must not release seq 2
+  assert.deepEqual(guards.begin(POSITION_1), { seq: null, cached: null });
+});
+
+// ── shouldApply: applied only while the scrubber is on that position ─────────
+
+test("shouldApply: the current position's response is applied", () => {
+  const guards = createTemporalSnapshotGuards();
+  const { seq } = guards.begin(POSITION_1);
+  assert.equal(guards.shouldApply(POSITION_1, seq), true);
+});
+
+test("shouldApply: a response for a position the scrubber left is discarded", () => {
+  const guards = createTemporalSnapshotGuards();
+  const { seq: seq1 } = guards.begin(POSITION_1);
+  guards.begin(POSITION_2);
+  assert.equal(guards.shouldApply(POSITION_1, seq1), false);
+  assert.equal(guards.shouldApply(POSITION_2, 2), true);
+});
+
+test("shouldApply: a late response for the position the scrubber returned to is applied", () => {
+  const guards = createTemporalSnapshotGuards();
+  const { seq: seq1 } = guards.begin(POSITION_1);
+  const { seq: seq2 } = guards.begin(POSITION_2);
+  guards.begin(POSITION_1); // back to 1: deduplicated, no new request
+  assert.equal(guards.shouldApply(POSITION_1, seq1), true);
+  assert.equal(guards.shouldApply(POSITION_2, seq2), false);
+});
+
+test("shouldApply: an unknown sequence is discarded", () => {
+  const guards = createTemporalSnapshotGuards();
+  guards.begin(POSITION_1);
+  assert.equal(guards.shouldApply(POSITION_1, 99), false);
+});
+
+test("shouldApply: after a reset no pre-reset response applies", () => {
+  const guards = createTemporalSnapshotGuards();
+  const { seq } = guards.begin(POSITION_1);
+  guards.reset();
+  assert.equal(guards.shouldApply(POSITION_1, seq), false);
+});
+
+// ── apply: caching for revisits ─────────────────────────────────────────────
+
+test("apply: stores the snapshot so a revisit re-applies it without a request", () => {
+  const guards = createTemporalSnapshotGuards();
+  const { seq } = guards.begin(POSITION_1);
+  guards.apply(POSITION_1, seq, SNAPSHOT);
+  guards.begin(POSITION_2);
+  assert.deepEqual(guards.begin(POSITION_1).cached, SNAPSHOT);
+});
+
+test("apply: play wrap-around re-applies the wrapped-to position's snapshot", () => {
+  const guards = createTemporalSnapshotGuards();
+  const { seq } = guards.begin(POSITION_1);
+  guards.apply(POSITION_1, seq, SNAPSHOT);
+  guards.begin(POSITION_2);
+  guards.begin(POSITION_3);
+  const wrap = guards.begin(POSITION_1);
+  assert.deepEqual(wrap.cached, SNAPSHOT);
+  assert.equal(guards.shouldApply(POSITION_1, wrap.seq), true);
+});
+
+// ── reset: graph reload ─────────────────────────────────────────────────────
+
+test("reset: clears requested and cached state so positions refetch", () => {
+  const guards = createTemporalSnapshotGuards();
+  const { seq } = guards.begin(POSITION_1);
+  guards.apply(POSITION_1, seq, SNAPSHOT);
+  guards.reset();
+  const fresh = guards.begin(POSITION_1);
+  assert.equal(fresh.seq, 1);
+  assert.equal(fresh.cached, null);
+});
+
+// ── cache bound ─────────────────────────────────────────────────────────────
+
+test("cache: oldest positions are evicted when the cache is full", () => {
+  const guards = createTemporalSnapshotGuards();
+  const count = 300;
+  for (let i = 0; i < count; i++) {
+    const { seq } = guards.begin(POSITION_1 + i * 1000);
+    guards.apply(POSITION_1 + i * 1000, seq, SNAPSHOT);
+  }
+  const oldest = guards.begin(POSITION_1);
+  assert.equal(oldest.cached, null); // evicted: must refetch on revisit
+  const newest = guards.begin(POSITION_1 + (count - 1) * 1000);
+  assert.deepEqual(newest.cached, SNAPSHOT); // still cached
+});