fix(desktop): keep active-turn badges through transient relay drops (#1120)

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>

desktop/playwright.config.ts

@@ -33,6 +33,7 @@ export default defineConfig({
         "**/channel-controls-screenshots.spec.ts",
         "**/team-management-screenshots.spec.ts",
         "**/active-turn-screenshots.spec.ts",
+        "**/active-turn-resilience-screenshots.spec.ts",
         "**/profile-active-turn-screenshots.spec.ts",
         "**/file-attachment.spec.ts",
         "**/video-attachment.spec.ts",

desktop/src/features/agents/activeAgentTurnsStore.test.mjs

@@ -643,51 +643,133 @@ describe("activeAgentTurnsStore", () => {
       );
     });
 
-    it("prunes a turn that receives no activity past the bound", () => {
+    it("prunes a dead turn at the bound while a live sibling keeps the stream fresh", () => {
+      // The no-regression case for the all-stale pause: a turn dies (no more
+      // liveness) but ANOTHER turn keeps refreshing. The pause gates on the MAX
+      // lastActivityAt, so the live sibling keeps it fresh and the dead turn
+      // still prunes at 25s — the pause only engages when EVERY turn is stale.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({
+          seq: 1,
+          turnId: "dead",
+          channelId: "c1",
+          timestamp: at(0),
+        }),
+        makeEvent({
+          seq: 2,
+          turnId: "live",
+          channelId: "c2",
+          timestamp: at(0),
+        }),
+      ]);
+      assert.equal(getActiveTurnsForAgent(AGENT).length, 2);
+
+      // Keep the live turn fresh across the dead turn's bound: ping every 10s.
+      for (let t = 10_000; t <= 30_000; t += 10_000) {
+        mock.timers.tick(10_000);
+        syncAgentTurnsFromEvents(AGENT, [
+          makeEvent({
+            seq: 2 + t / 10_000,
+            kind: "turn_liveness",
+            turnId: "live",
+            channelId: "c2",
+            timestamp: at(t),
+          }),
+        ]);
+      }
+
+      const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
+      assert.ok(!channels.has("c1"), "the dead turn must prune at the bound");
+      assert.ok(channels.has("c2"), "the live sibling must survive");
+    });
+
+    it("pauses pruning when EVERY tracked turn goes stale at once (relay drop)", () => {
+      // The "all at once" drop signature: all liveness stops simultaneously.
+      // No turn refreshes the max, so the pause engages before the 25s prune
+      // and the badges stay visible through the transient drop.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
+        makeEvent({ seq: 2, turnId: "t2", channelId: "c2", timestamp: at(0) }),
+      ]);
+      assert.equal(getActiveTurnsForAgent(AGENT).length, 2);
+
+      // Silence past the bound — the pause must hold both badges.
+      mock.timers.tick(REMOVE_AFTER_MS + PRUNE_INTERVAL_MS);
+
+      assert.equal(
+        getActiveTurnsForAgent(AGENT).length,
+        2,
+        "all-stale-at-once must pause the prune so badges survive a drop",
+      );
+    });
+
+    it("holds a lone silent turn past the bound until the next frame (residual)", () => {
+      // The accepted residual: a single turn whose host dies (kill -9) under a
+      // HEALTHY relay is indistinguishable from a drop with one live turn, so
+      // its badge lingers past 25s. It clears the instant any frame arrives.
       syncAgentTurnsFromEvents(AGENT, [
         makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
       ]);
       assert.equal(getActiveTurnsForAgent(AGENT).length, 1);
 
-      // No liveness pings — the host died without unwinding. Advance past the
-      // 25s bound; the next prune sweep evicts the silent turn.
       mock.timers.tick(REMOVE_AFTER_MS + PRUNE_INTERVAL_MS);
 
       assert.equal(
         getActiveTurnsForAgent(AGENT).length,
-        0,
-        "a turn with no activity past the bound must be pruned",
+        1,
+        "a lone silent turn lingers (pause engages) — accepted residual",
       );
     });
 
     it("treats a turn_liveness with a null turnId as a no-op", () => {
+      // A null-turnId liveness must refresh NOTHING. With a live sibling
+      // keeping the max fresh, the dead turn still prunes at the bound — so if
+      // the null ping wrongly refreshed the dead turn it would survive here.
       syncAgentTurnsFromEvents(AGENT, [
-        makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
+        makeEvent({
+          seq: 1,
+          turnId: "dead",
+          channelId: "c1",
+          timestamp: at(0),
+        }),
+        makeEvent({
+          seq: 2,
+          turnId: "live",
+          channelId: "c2",
+          timestamp: at(0),
+        }),
       ]);
 
-      // A liveness ping with no turnId must refresh nothing (recordActivity
-      // no-ops on null). If it wrongly refreshed, the turn would survive the
-      // bound below — so the prune is the observable proof of the no-op, and
-      // the missing turnId must not throw.
-      mock.timers.tick(20_000);
+      // A null-turnId liveness for the dead turn must not refresh it. Keep the
+      // live sibling pinging so the pause never engages.
       assert.doesNotThrow(() => {
-        syncAgentTurnsFromEvents(AGENT, [
-          makeEvent({
-            seq: 2,
-            kind: "turn_liveness",
-            turnId: null,
-            channelId: "c1",
-            timestamp: at(20_000),
-          }),
-        ]);
+        for (let t = 10_000; t <= 30_000; t += 10_000) {
+          mock.timers.tick(10_000);
+          syncAgentTurnsFromEvents(AGENT, [
+            makeEvent({
+              seq: 100 + t,
+              kind: "turn_liveness",
+              turnId: null,
+              channelId: "c1",
+              timestamp: at(t),
+            }),
+            makeEvent({
+              seq: 200 + t,
+              kind: "turn_liveness",
+              turnId: "live",
+              channelId: "c2",
+              timestamp: at(t),
+            }),
+          ]);
+        }
       });
-      mock.timers.tick(REMOVE_AFTER_MS + PRUNE_INTERVAL_MS);
 
-      assert.equal(
-        getActiveTurnsForAgent(AGENT).length,
-        0,
-        "a null-turnId liveness must not refresh activity, so the turn still prunes",
+      const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
+      assert.ok(
+        !channels.has("c1"),
+        "a null-turnId liveness must not refresh the dead turn, so it prunes",
       );
+      assert.ok(channels.has("c2"), "the live sibling must survive");
     });
   });
 
@@ -832,6 +914,256 @@ describe("activeAgentTurnsStore", () => {
       );
     });
   });
+
+  describe("resurrection after a prune (A) gated by completion (C)", () => {
+    const EPOCH = Date.parse("2024-01-01T00:00:00Z");
+    const at = (ms) => new Date(EPOCH + ms).toISOString();
+    const REMOVE_AFTER_MS = 25_000;
+    const PRUNE_INTERVAL_MS = 5_000;
+
+    let unsubscribe;
+
+    beforeEach(() => {
+      mock.timers.enable({ apis: ["setInterval", "Date"], now: EPOCH });
+      unsubscribe = subscribeActiveAgentTurns(() => {});
+    });
+
+    afterEach(() => {
+      unsubscribe();
+      mock.timers.reset();
+    });
+
+    it("resurrects a lone turn pruned out from under a still-running host", () => {
+      // The lone-crash residual self-heals: the badge is pruned during silence,
+      // then a recovered liveness frame for the same turn revives it.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
+      ]);
+      mock.timers.tick(REMOVE_AFTER_MS + PRUNE_INTERVAL_MS);
+
+      // A live sibling appears, which lets the prune fire and clear the lone
+      // stale turn; then a recovered liveness for t1 must revive its badge.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({
+          seq: 2,
+          turnId: "t2",
+          channelId: "c2",
+          timestamp: at(40_000),
+        }),
+      ]);
+      mock.timers.tick(PRUNE_INTERVAL_MS);
+      assert.ok(
+        !channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
+        "the stale lone turn must prune once a live sibling unblocks the sweep",
+      );
+
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({
+          seq: 3,
+          kind: "turn_liveness",
+          turnId: "t1",
+          channelId: "c1",
+          timestamp: at(45_000),
+        }),
+      ]);
+      assert.ok(
+        channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
+        "a recovered liveness must resurrect the pruned turn's badge",
+      );
+    });
+
+    it("does NOT resurrect a turn whose liveness is older than its completion", () => {
+      // Bound-proving (stale side): a turn completes, then a liveness frame
+      // arrives carrying a timestamp BEFORE the completion. It must not revive.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
+        makeEvent({
+          seq: 2,
+          kind: "turn_completed",
+          turnId: "t1",
+          channelId: "c1",
+          timestamp: at(10_000),
+        }),
+      ]);
+      assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
+
+      // A liveness stamped at 5s (before the 10s completion) but delivered with
+      // a later seq so it clears the watermark on seq. It is stale relative to
+      // the completion and must NOT resurrect.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({
+          seq: 3,
+          kind: "turn_liveness",
+          turnId: "t1",
+          channelId: "c1",
+          timestamp: at(5_000),
+        }),
+      ]);
+      assert.equal(
+        getActiveTurnsForAgent(AGENT).length,
+        0,
+        "a liveness older than the recorded completion must not resurrect the turn",
+      );
+    });
+
+    it("DOES resurrect a turn whose liveness is strictly newer than its completion", () => {
+      // Bound-proving (live side): the same completed turn, but a liveness frame
+      // strictly NEWER than the completion (a genuine restart of the same id)
+      // must revive — the completion only blocks stale frames, not new work.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
+        makeEvent({
+          seq: 2,
+          kind: "turn_completed",
+          turnId: "t1",
+          channelId: "c1",
+          timestamp: at(10_000),
+        }),
+      ]);
+      assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
+
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({
+          seq: 3,
+          kind: "turn_liveness",
+          turnId: "t1",
+          channelId: "c1",
+          timestamp: at(20_000),
+        }),
+      ]);
+      assert.ok(
+        channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
+        "a liveness strictly newer than the completion must resurrect the turn",
+      );
+    });
+
+    it("does NOT resurrect from a liveness frame with no channelId", () => {
+      // A pruned turn cannot be rebuilt without a channelId to anchor the badge.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
+        makeEvent({
+          seq: 2,
+          turnId: "t2",
+          channelId: "c2",
+          timestamp: at(0),
+        }),
+      ]);
+      // Drop t1 by ending it, then send a channelId-less liveness for it.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({
+          seq: 3,
+          kind: "turn_completed",
+          turnId: "t1",
+          channelId: "c1",
+          timestamp: at(5_000),
+        }),
+        makeEvent({
+          seq: 4,
+          kind: "turn_liveness",
+          turnId: "t1",
+          channelId: null,
+          timestamp: at(10_000),
+        }),
+      ]);
+      assert.ok(
+        !channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
+        "a channelId-less liveness cannot resurrect a badge",
+      );
+    });
+
+    it("clears completion tombstones on reset so a later turn can run", () => {
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
+        makeEvent({
+          seq: 2,
+          kind: "turn_completed",
+          turnId: "t1",
+          channelId: "c1",
+          timestamp: at(10_000),
+        }),
+      ]);
+
+      resetActiveAgentTurnsStore();
+
+      // After reset, an OLD-stamped liveness for the same id must resurrect,
+      // proving the tombstone (which would otherwise block it) was cleared.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({
+          seq: 1,
+          kind: "turn_liveness",
+          turnId: "t1",
+          channelId: "c1",
+          timestamp: at(1_000),
+        }),
+      ]);
+      assert.ok(
+        channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
+        "reset must clear terminal tombstones so they do not leak across reset",
+      );
+    });
+
+    it("evicts the oldest tombstone once past the cap so the map stays bounded", () => {
+      // The tombstone map is capped at MAX_TERMINAL_TOMBSTONES (16). Complete
+      // 18 distinct turns so eviction fires twice, dropping the two oldest by
+      // insertion order (t0, t1). Probe via the ONE behavior a tombstone gates
+      // that a strictly-newer frame cannot mask: an EQUAL-timestamp liveness
+      // (frameAt == terminalAt). All completions share timestamp T with rising
+      // seq, so the probe clears the per-agent watermark on the seq tiebreak
+      // (compareObserverEvents is timestamp-primary, seq-secondary) yet stays
+      // equal to the recorded terminal — reaching resurrectTurn's tombstone
+      // check rather than being shadowed by the watermark.
+      const CAP = 16;
+      const TOTAL = CAP + 2;
+      const T = at(0);
+      const completions = [];
+      for (let i = 0; i < TOTAL; i++) {
+        completions.push(
+          makeEvent({
+            seq: i + 1,
+            kind: "turn_completed",
+            turnId: `t${i}`,
+            channelId: `c${i}`,
+            timestamp: T,
+          }),
+        );
+      }
+      syncAgentTurnsFromEvents(AGENT, completions);
+
+      // A surviving tombstone (t2, third-completed) still blocks an
+      // equal-timestamp liveness — proves the tombstone is present and doing
+      // the work the watermark cannot.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({
+          seq: TOTAL + 1,
+          kind: "turn_liveness",
+          turnId: "t2",
+          channelId: "c2",
+          timestamp: T,
+        }),
+      ]);
+      assert.ok(
+        !channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c2"),
+        "a surviving tombstone must still block an equal-timestamp liveness",
+      );
+
+      // The oldest tombstone (t0) was evicted, so the same equal-timestamp
+      // liveness now resurrects — proving the cap fired AND evicted the
+      // oldest-by-insertion entry, not an arbitrary one.
+      syncAgentTurnsFromEvents(AGENT, [
+        makeEvent({
+          seq: TOTAL + 2,
+          kind: "turn_liveness",
+          turnId: "t0",
+          channelId: "c0",
+          timestamp: T,
+        }),
+      ]);
+      assert.ok(
+        channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c0"),
+        "the oldest tombstone must be evicted once past the cap",
+      );
+    });
+  });
 });
 
 describe("formatElapsed", () => {

desktop/src/features/agents/activeAgentTurnsStore.ts

@@ -16,8 +16,18 @@ const LIVENESS_INTERVAL_MS = 10_000;
  * graceful exits clear via turn_completed and working turns refresh on every
  * stream event. Derived from the interval so it tracks if the interval changes. */
 const REMOVE_AFTER_MS = LIVENESS_INTERVAL_MS * 2.5;
+/** Pause pruning once EVERY tracked turn has gone this long without activity —
+ * the "all at once" signature of a relay drop (flaky VPN), where liveness frames
+ * stop arriving for all agents simultaneously. Set below REMOVE_AFTER_MS so the
+ * pause engages before the 25s prune would wipe the badges. */
+const FRAME_GAP_PAUSE_MS = LIVENESS_INTERVAL_MS * 2;
 /** Maximum concurrent active turns tracked per agent (matches pool size). */
 const MAX_TURNS_PER_AGENT = 4;
+/** Cap on per-agent terminal tombstones (A's resurrection guard). Only the
+ * most recently completed turns can be raced by a late liveness frame; older
+ * ones are already below the watermark, so a small multiple of the live cap is
+ * ample and keeps the map from growing across a long session. */
+const MAX_TERMINAL_TOMBSTONES = MAX_TURNS_PER_AGENT * 4;
 /** Interval for pruning stale/expired turns. */
 const PRUNE_INTERVAL_MS = 5_000;
 
@@ -65,6 +75,14 @@ const cachedTurnSummaries = new Map<string, ActiveTurnSummary[]>();
 // streams (seq resets to 1, timestamp keeps climbing) handled for free.
 const lastProcessed = new Map<string, ObserverEvent>();
 
+// Per-agent record of when each turn terminally ended (turnId →
+// terminal-event timestamp, in agent-host clock ms). endTurn hard-deletes a
+// turn with no surviving record, so without this a late liveness frame for an
+// already-completed turn would resurrect a dead badge. Resurrection (A) checks
+// this: a turn is revived only if the recovered liveness is strictly newer
+// than its recorded terminal timestamp.
+const terminalAtByAgent = new Map<string, Map<string, number>>();
+
 let pruneInterval: ReturnType<typeof setInterval> | null = null;
 
 function invalidateCache(agentKey: string) {
@@ -132,14 +150,60 @@ function startTurn(
   invalidateCache(key);
 }
 
-function recordActivity(agentPubkey: string, turnId: string | null) {
-  if (!turnId) return;
+function recordActivity(agentPubkey: string, turnId: string | null): boolean {
+  if (!turnId) return false;
   const key = normalizePubkey(agentPubkey);
   const agentTurns = activeTurnsByAgent.get(key);
-  if (!agentTurns) return;
+  if (!agentTurns) return false;
   const turn = agentTurns.get(turnId);
   if (turn) {
     turn.lastActivityAt = Date.now();
+    return true;
+  }
+  return false;
+}
+
+/**
+ * A — resurrect a badge that was pruned out from under a still-running turn.
+ * A recovered liveness/acp frame for a turn no longer in the live map recreates
+ * it, UNLESS C's tombstone shows the turn already terminally ended at or after
+ * this frame's time (a stale frame must not revive a completed turn). The frame
+ * carries no record of the original start, so the badge re-anchors to this
+ * frame's timestamp — it resumes counting from recovery, which is the honest
+ * floor for a turn whose true start is unrecoverable. Returns true on revive.
+ */
+function resurrectTurn(agentPubkey: string, event: ObserverEvent): boolean {
+  if (!event.turnId || !event.channelId) return false;
+  const key = normalizePubkey(agentPubkey);
+  const terminalAt = terminalAtByAgent.get(key)?.get(event.turnId);
+  const frameAt = Date.parse(event.timestamp);
+  // Only revive when this frame is strictly newer than the recorded terminal.
+  if (
+    terminalAt !== undefined &&
+    (!Number.isFinite(frameAt) || frameAt <= terminalAt)
+  ) {
+    return false;
+  }
+  startTurn(agentPubkey, event.channelId, event.turnId, event.timestamp);
+  return true;
+}
+
+function recordTerminal(agentKey: string, turnId: string, terminalAt: number) {
+  if (!Number.isFinite(terminalAt)) return;
+  let terminals = terminalAtByAgent.get(agentKey);
+  if (!terminals) {
+    terminals = new Map();
+    terminalAtByAgent.set(agentKey, terminals);
+  }
+  terminals.set(turnId, terminalAt);
+  // Bound the tombstone map: only recently-completed turns can be the target of
+  // a racing late liveness frame (older ones are already below the watermark).
+  // Evict the oldest terminal once past the cap so the map can't grow unbounded
+  // across a long session. Insertion order tracks completion order closely
+  // enough; the first key is the oldest survivor.
+  if (terminals.size > MAX_TERMINAL_TOMBSTONES) {
+    const oldest = terminals.keys().next().value;
+    if (oldest !== undefined) terminals.delete(oldest);
   }
 }
 
@@ -147,18 +211,29 @@ function endTurn(
   agentPubkey: string,
   turnId: string | null,
   channelId: string | null,
+  terminalAt: number,
 ) {
   const key = normalizePubkey(agentPubkey);
+  // Tombstone the terminal time so a late liveness frame can't resurrect a
+  // completed turn (A's guard). With an explicit turnId this is recorded even
+  // when the turn was already pruned and the agent's live map is gone — the
+  // completion is authoritative and must outlive the active record.
+  if (turnId) {
+    recordTerminal(key, turnId, terminalAt);
+  }
+
   const agentTurns = activeTurnsByAgent.get(key);
   if (!agentTurns) return;
 
   if (turnId) {
     agentTurns.delete(turnId);
   } else if (channelId) {
-    // Fallback: remove by channelId if turnId not available
+    // Fallback: remove by channelId if turnId not available. Tombstone the
+    // resolved turn so a later stale liveness for it can't resurrect a badge.
     for (const [tid, turn] of agentTurns) {
       if (turn.channelId === channelId) {
         agentTurns.delete(tid);
+        recordTerminal(key, tid, terminalAt);
         break;
       }
     }
@@ -169,8 +244,32 @@ function endTurn(
   invalidateCache(key);
 }
 
+/** True when every tracked turn across every agent is simultaneously stale —
+ * no turn has had activity within FRAME_GAP_PAUSE_MS. With no tracked turns
+ * there is nothing to prune, so it returns false (never pause). */
+function shouldPausePrune(now: number): boolean {
+  let maxActivity = 0;
+  for (const agentTurns of activeTurnsByAgent.values())
+    for (const turn of agentTurns.values())
+      if (turn.lastActivityAt > maxActivity) maxActivity = turn.lastActivityAt;
+  return maxActivity > 0 && now - maxActivity > FRAME_GAP_PAUSE_MS;
+}
+
 function pruneExpired() {
   const now = Date.now();
+  // Pause pruning when ALL tracked turns are simultaneously stale — the "all
+  // at once" signature of a relay drop, where every agent's liveness stops in
+  // the same instant. Gating on the MAX lastActivityAt (not a global frame
+  // clock) is what keeps this from over-pausing: a single live sibling turn
+  // keeps the max fresh, so a genuinely dead turn still prunes at 25s — no
+  // regression for the multi-agent crash case. Residual: a LONE turn kill -9'd
+  // under a HEALTHY relay (it was the only active turn) keeps its badge until
+  // the next frame instead of clearing at 25s, since local-only sensing cannot
+  // distinguish that from a drop. The badge self-heals the instant any frame
+  // arrives. Accepted tradeoff to keep badges visible through transient drops.
+  if (shouldPausePrune(now)) {
+    return;
+  }
   let changed = false;
   for (const [agentKey, agentTurns] of activeTurnsByAgent) {
     for (const [turnId, turn] of agentTurns) {
@@ -201,8 +300,10 @@ function processEvent(agentPubkey: string, event: ObserverEvent) {
   // no-op. Evictions must be gated too — replaying a stale turn_error/
   // agent_panic (emitted with a null turnId) would otherwise fall back to
   // deleting the first turn in the channel, killing the live turn. Resurrection
-  // is not a concern: it would require reprocessing a stale start, which the
-  // watermark already blocks.
+  // (the turn_liveness/acp case below) is gated here too: it runs only for a
+  // frame that passes the watermark, so replayed stale frames cannot revive a
+  // pruned turn, and the per-turn terminal tombstone blocks reviving a turn
+  // that already completed.
   const last = lastProcessed.get(key);
   if (last && compareObserverEvents(event, last) <= 0) {
     return;
@@ -230,17 +331,29 @@ function processEvent(agentPubkey: string, event: ObserverEvent) {
     case "turn_completed":
     case "turn_error":
     case "agent_panic":
-      endTurn(agentPubkey, event.turnId ?? null, event.channelId ?? null);
+      endTurn(
+        agentPubkey,
+        event.turnId ?? null,
+        event.channelId ?? null,
+        Date.parse(event.timestamp),
+      );
       notifyListeners();
       return;
     case "acp_read":
     case "acp_write":
     // turn_liveness keeps a quiet-but-alive turn from being pruned; same
     // refresh-only path as stream activity — no surfaced summary change on its
-    // own, so it only notifies when the offset above actually moved.
-    case "turn_liveness":
-      recordActivity(agentPubkey, event.turnId ?? null);
+    // own, so it only notifies when the offset above actually moved. If the
+    // turn was pruned out from under a still-running host (a transient drop
+    // raced the pause, or the lone-crash residual self-healed), resurrect it.
+    case "turn_liveness": {
+      const refreshed = recordActivity(agentPubkey, event.turnId ?? null);
+      if (!refreshed && resurrectTurn(agentPubkey, event)) {
+        notifyListeners();
+        return;
+      }
       break;
+    }
   }
 
   if (offsetChanged) {
@@ -372,5 +485,6 @@ export function resetActiveAgentTurnsStore() {
   lastProcessed.clear();
   clockOffsetByAgent.clear();
   cachedTurnSummaries.clear();
+  terminalAtByAgent.clear();
   notifyListeners();
 }

desktop/tests/e2e/active-turn-resilience-screenshots.spec.ts

@@ -0,0 +1,136 @@
+import { expect, test } from "@playwright/test";
+
+import { installMockBridge } from "../helpers/bridge";
+
+const SHOTS = "test-results/active-turn-resilience";
+
+// Mock agent pubkeys (distinct from the relay agents seeded by default).
+const AGENT_PAUL = "aa".repeat(32);
+const AGENT_DUNCAN = "bb".repeat(32);
+
+// Mock channel IDs from the e2e bridge.
+const CHANNEL_GENERAL = "9a1657ac-f7aa-5db0-b632-d8bbeb6dfb50";
+const CHANNEL_ENGINEERING = "1c7e1c02-87bb-5e88-b2da-5a7a9432d0c9";
+
+// A fixed epoch so the mocked clock is deterministic across runs.
+const T0 = new Date("2026-06-18T12:00:00.000Z");
+
+// Past both thresholds: FRAME_GAP_PAUSE_MS (20s) and REMOVE_AFTER_MS (25s).
+// Several 5s prune ticks fire across this span, so shouldPausePrune is what
+// keeps the badges alive — not the absence of a prune tick.
+const FRAME_GAP_MS = 30_000;
+
+type SeedInput = {
+  agentPubkey: string;
+  channelId: string;
+  turnId: string;
+};
+
+async function waitForBridge(page: import("@playwright/test").Page) {
+  await page.waitForFunction(
+    () =>
+      typeof (window as Window & { __BUZZ_E2E_SEED_ACTIVE_TURNS__?: unknown })
+        .__BUZZ_E2E_SEED_ACTIVE_TURNS__ === "function",
+    null,
+    { timeout: 10_000 },
+  );
+}
+
+async function openAgentsView(page: import("@playwright/test").Page) {
+  await page.goto("/", { waitUntil: "domcontentloaded" });
+  await waitForBridge(page);
+  await page.getByTestId("open-agents-view").click();
+  await expect(page.getByTestId("unified-agents-groups")).toBeVisible({
+    timeout: 10_000,
+  });
+}
+
+async function seedTurns(
+  page: import("@playwright/test").Page,
+  turns: SeedInput[],
+) {
+  await page.evaluate((seeds) => {
+    const win = window as Window & {
+      __BUZZ_E2E_SEED_ACTIVE_TURNS__?: (input: {
+        agentPubkey: string;
+        channelId: string;
+        turnId: string;
+      }) => void;
+    };
+    for (const seed of seeds) win.__BUZZ_E2E_SEED_ACTIVE_TURNS__?.(seed);
+  }, turns);
+}
+
+test.describe("active turn badge resilience screenshots", () => {
+  test.use({ viewport: { width: 1280, height: 720 } });
+
+  test("badges persist through an all-at-once liveness gap", async ({
+    page,
+  }) => {
+    // Install the mocked clock BEFORE navigation so the store's Date.now() /
+    // setInterval and the badge's useNow(1000) all run on the mocked clock from
+    // module init. Seeded turns then stamp lastActivityAt at T0.
+    await installMockBridge(page, {
+      managedAgents: [
+        {
+          pubkey: AGENT_PAUL,
+          name: "Paul",
+          status: "running",
+          channelNames: ["general", "engineering"],
+        },
+        {
+          pubkey: AGENT_DUNCAN,
+          name: "Duncan",
+          status: "running",
+          channelNames: ["general"],
+        },
+      ],
+    });
+    await page.clock.install({ time: T0 });
+
+    await openAgentsView(page);
+
+    // Both agents working across channels — the healthy multi-agent state.
+    await seedTurns(page, [
+      {
+        agentPubkey: AGENT_PAUL,
+        channelId: CHANNEL_GENERAL,
+        turnId: "t-paul-g",
+      },
+      {
+        agentPubkey: AGENT_PAUL,
+        channelId: CHANNEL_ENGINEERING,
+        turnId: "t-paul-e",
+      },
+      {
+        agentPubkey: AGENT_DUNCAN,
+        channelId: CHANNEL_GENERAL,
+        turnId: "t-duncan-g",
+      },
+    ]);
+
+    const paulRow = page.getByTestId(`managed-agent-${AGENT_PAUL}`);
+    const duncanRow = page.getByTestId(`managed-agent-${AGENT_DUNCAN}`);
+    await expect(paulRow).toContainText("Working", { timeout: 5_000 });
+    await expect(duncanRow).toContainText("Working", { timeout: 5_000 });
+
+    const agentsSection = page.getByTestId("unified-agents-groups");
+    await agentsSection.screenshot({
+      path: `${SHOTS}/01-badges-before-gap.png`,
+    });
+
+    // Simulate the all-at-once relay drop: no further frames, advance the clock
+    // past both thresholds. This fires several real prune ticks; shouldPausePrune
+    // sees every turn's lastActivityAt stuck at T0 (gap > 20s) and pauses the
+    // prune, so the badges survive. Under the pre-fix code every badge would be
+    // gone after the first tick past 25s.
+    await page.clock.fastForward(FRAME_GAP_MS);
+
+    await expect(paulRow).toContainText("Working");
+    await expect(duncanRow).toContainText("Working");
+
+    await agentsSection.screenshot({
+      path: `${SHOTS}/02-badges-survive-gap.png`,
+    });
+  });
+});