perf(desktop): update active turns incrementally (#5897)

## Problem

Every observer-store publication made the active-turn bridge scan every
running/deployed agent and replay each agent's retained observer
journal. Watermarks kept the replay idempotent, but did not remove the
repeated work. Under an active fleet, one changed agent therefore caused
work proportional to the whole fleet and its retained history.

## Change

- observer publications now identify the changed agent and only the
newly admitted, retained events
- the active-turn bridge still performs one full hydration when its
agent list mounts or changes
- steady-state publications process only that changed active agent's
delta
- other observer-store subscribers keep their existing notification
behavior
- duplicate-only envelopes still do not publish

## Correctness

Regression coverage pins:

- retained/duplicate history is omitted from deltas
- stopped-agent updates do not enter active-turn state
- an incremental terminal clears a turn hydrated from retained history
- batching still publishes once and preserves transcript/terminal
outcomes
- existing watermark, tombstone, pruning, community restore, clear, and
eviction suites remain green

## Validation

Exact pushed head: `a480ffd2531023ea32b2a5518b5d9d41f04577c8`

- focused active-turn + observer-retention suites: 90 passed
- full desktop suite: 4,891 passed
- `pnpm --dir desktop typecheck`: passed
- `pnpm --dir desktop check`: passed (pre-existing repository warnings
only)
- mandatory pre-push hook at the exact pushed head: passed
`branch-skew`, desktop check/typecheck/test, mobile tests, Rust tests,
and Desktop Tauri checks

Packaged same-fleet CPU/RSS validation is follow-up evidence; this PR
proves the algorithmic amplification is removed without claiming an
installed-app percentage from unit tests.

Signed-off-by: Wes <wesbillman@users.noreply.github.com>
Co-authored-by: Carl <c7ebe626f000404285d3686e1dc74cc07cc60a9754a150041ba132e14bd3e2ec@buzz.block.builderlab.xyz>

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

@@ -12,6 +12,7 @@ import {
   restoreActiveAgentTurnsForCommunity,
   clearSavedCommunitySnapshot,
   clearActiveTurnsForAgent,
+  createActiveAgentTurnsObserverListener,
 } from "./activeAgentTurnsStore.ts";
 import {
   injectObserverEventsForE2E,
@@ -1625,6 +1626,82 @@ describe("observer → active-turns bridge sync", () => {
     );
   });
 
+  it("publishes only newly admitted events for the changed agent", () => {
+    const updates = [];
+    const unsubscribeObserver = subscribeAgentObserverStore((update) => {
+      updates.push(update);
+    });
+    const retained = makeEvent({ seq: 1, kind: "turn_started" });
+    const admitted = makeEvent({
+      seq: 2,
+      kind: "acp_write",
+      timestamp: "2024-01-01T00:00:01Z",
+    });
+
+    injectObserverEventsForE2E(AGENT, [retained]);
+    injectObserverEventsForE2E(AGENT, [retained, admitted]);
+    unsubscribeObserver();
+
+    assert.equal(updates.length, 2);
+    assert.equal(updates[1].agentPubkey, AGENT);
+    assert.deepEqual(
+      updates[1].events.map((event) => event.seq),
+      [2],
+      "the publication must omit retained and duplicate history",
+    );
+  });
+
+  it("steady-state listener processes the changed active agent only", () => {
+    const listener = createActiveAgentTurnsObserverListener([
+      { pubkey: AGENT, status: "deployed" },
+      { pubkey: AGENT_2, status: "stopped" },
+    ]);
+
+    listener({
+      agentPubkey: AGENT,
+      events: [makeEvent({ seq: 1, turnId: "active-turn" })],
+    });
+    listener({
+      agentPubkey: AGENT_2,
+      events: [
+        makeEvent({
+          seq: 1,
+          turnId: "stopped-turn",
+          channelId: "stopped-channel",
+        }),
+      ],
+    });
+
+    assert.equal(getActiveTurnsForAgent(AGENT).length, 1);
+    assert.equal(
+      getActiveTurnsForAgent(AGENT_2).length,
+      0,
+      "an unrelated stopped agent update must not enter turn state",
+    );
+  });
+
+  it("incremental terminal update clears a hydrated turn without replay", () => {
+    injectObserverEventsForE2E(AGENT, [
+      makeEvent({ seq: 1, kind: "turn_started" }),
+    ]);
+    syncActiveAgentTurnsFromObserver(bridgeAgents);
+    assert.equal(getActiveTurnsForAgent(AGENT).length, 1);
+
+    const listener = createActiveAgentTurnsObserverListener(bridgeAgents);
+    listener({
+      agentPubkey: AGENT,
+      events: [
+        makeEvent({
+          seq: 2,
+          kind: "turn_completed",
+          timestamp: "2024-01-01T00:00:05Z",
+        }),
+      ],
+    });
+
+    assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
+  });
+
   it("publishes one observer update for a batch while preserving outcomes", () => {
     let observerNotifications = 0;
     const unsubscribeObserver = subscribeAgentObserverStore(() => {

desktop/src/features/agents/activeAgentTurnsStore.ts

@@ -4,6 +4,7 @@ import {
   subscribeAgentObserverStore,
   getAgentObserverSnapshot,
   compareObserverEvents,
+  type AgentObserverStoreUpdate,
 } from "@/features/agents/observerRelayStore";
 import { normalizePubkey } from "@/shared/lib/pubkey";
 import {
@@ -631,19 +632,40 @@ export function syncActiveAgentTurnsFromObserver(
 }
 
 /**
- * Bridge hook: processes observer events into the active-turns store.
- * Should be called by a parent component that has access to the observer events.
+ * Build the steady-state observer listener once per agent-list revision. Observer
+ * publications carry only newly admitted events for one agent, so this callback
+ * does not revisit unrelated agents or their retained journals.
  */
+export function createActiveAgentTurnsObserverListener(
+  agents: readonly { pubkey: string; status: string }[],
+): (update?: AgentObserverStoreUpdate) => void {
+  const activeAgentPubkeys = new Set(
+    agents
+      .filter(
+        (agent) => agent.status === "running" || agent.status === "deployed",
+      )
+      .map((agent) => normalizePubkey(agent.pubkey)),
+  );
+
+  return (update?: AgentObserverStoreUpdate) => {
+    if (
+      !update ||
+      !activeAgentPubkeys.has(normalizePubkey(update.agentPubkey))
+    ) {
+      return;
+    }
+    syncAgentTurnsFromEvents(update.agentPubkey, [...update.events]);
+  };
+}
+
 export function useActiveAgentTurnsBridge(
   agents: readonly { pubkey: string; status: string }[],
 ) {
   React.useEffect(() => {
-    function syncAll() {
-      syncActiveAgentTurnsFromObserver(agents);
-    }
-
-    syncAll();
-    return subscribeAgentObserverStore(syncAll);
+    syncActiveAgentTurnsFromObserver(agents);
+    return subscribeAgentObserverStore(
+      createActiveAgentTurnsObserverListener(agents),
+    );
   }, [agents]);
 }
 

desktop/src/features/agents/observerRelayStore.ts

@@ -53,7 +53,14 @@ const IDLE_SNAPSHOT: ObserverSnapshot = {
 const EMPTY_EVENTS: ObserverEvent[] = [];
 const EMPTY_TRANSCRIPT: TranscriptItem[] = [];
 
-const listeners = new Set<() => void>();
+export type AgentObserverStoreUpdate = {
+  agentPubkey: string;
+  events: readonly ObserverEvent[];
+};
+
+type AgentObserverStoreListener = (update?: AgentObserverStoreUpdate) => void;
+
+const listeners = new Set<AgentObserverStoreListener>();
 const eventsByAgent = new Map<string, ObserverEvent[]>();
 const transcriptByAgent = new Map<string, TranscriptState>();
 const snapshotByAgent = new Map<string, ObserverSnapshot>();
@@ -192,9 +199,9 @@ let startPromise: Promise<void> | null = null;
 let eventProcessingQueue: Promise<void> = Promise.resolve();
 let generation = 0;
 
-function notifyListeners() {
+function notifyListeners(update?: AgentObserverStoreUpdate) {
   for (const listener of listeners) {
-    listener();
+    listener(update);
   }
 }
 
@@ -219,8 +226,8 @@ function observerTag(event: RelayEvent, tagName: string) {
 function appendAgentEvents(
   agentPubkey: string,
   events: readonly ObserverEvent[],
-): boolean {
-  if (events.length === 0) return false;
+): ObserverEvent[] | null {
+  if (events.length === 0) return null;
 
   const key = normalizePubkey(agentPubkey);
   const current = eventsByAgent.get(key) ?? [];
@@ -234,7 +241,7 @@ function appendAgentEvents(
   const admissible = floor
     ? events.filter((event) => isObserverEventAfter(event, floor))
     : events;
-  if (admissible.length === 0) return false;
+  if (admissible.length === 0) return null;
 
   const seen = new Set(
     current.map(
@@ -248,7 +255,7 @@ function appendAgentEvents(
     seen.add(eventKey);
     added.push(event);
   }
-  if (added.length === 0) return false;
+  if (added.length === 0) return null;
 
   const sortedAdded = added.sort(compareObserverEvents);
   const sorted = [...current, ...sortedAdded].sort(compareObserverEvents);
@@ -289,12 +296,24 @@ function appendAgentEvents(
   }
 
   invalidateSnapshot(key);
-  return true;
+  if (!trimmed) return sortedAdded;
+
+  const retainedKeys = new Set(
+    final.map(
+      (event) => `${event.timestamp.length}:${event.timestamp}:${event.seq}`,
+    ),
+  );
+  return sortedAdded.filter((event) =>
+    retainedKeys.has(
+      `${event.timestamp.length}:${event.timestamp}:${event.seq}`,
+    ),
+  );
 }
 
 function appendAgentEvent(agentPubkey: string, event: ObserverEvent) {
-  if (appendAgentEvents(agentPubkey, [event])) {
-    notifyListeners();
+  const added = appendAgentEvents(agentPubkey, [event]);
+  if (added) {
+    notifyListeners({ agentPubkey, events: added });
   }
 }
 
@@ -435,7 +454,7 @@ function processLiveObserverEvents(
   // callbacks. Those callbacks historically observed their triggering frame
   // in the raw/transcript stores; batching must preserve that visibility while
   // deferring only the global external-store publication.
-  const observerChanged = appendAgentEvents(agentPubkey, events);
+  const addedEvents = appendAgentEvents(agentPubkey, events);
 
   for (const parsed of events) {
     // Track the latest-live-session-id per (agent, channel) on the live path.
@@ -479,8 +498,8 @@ function processLiveObserverEvents(
 
   // Preserve the harness's envelope backpressure: retained state was committed
   // before specialized callbacks, but external-store subscribers publish once.
-  if (observerChanged) {
-    notifyListeners();
+  if (addedEvents) {
+    notifyListeners({ agentPubkey, events: addedEvents });
   }
 }
 
@@ -588,7 +607,9 @@ export function ensureRelayObserverSubscription() {
   return startPromise;
 }
 
-export function subscribeAgentObserverStore(listener: () => void) {
+export function subscribeAgentObserverStore(
+  listener: AgentObserverStoreListener,
+) {
   listeners.add(listener);
   return () => {
     listeners.delete(listener);
@@ -827,8 +848,9 @@ export function injectObserverEventsForE2E(
   agentPubkey: string,
   events: ObserverEvent[],
 ) {
-  if (appendAgentEvents(agentPubkey, events)) {
-    notifyListeners();
+  const added = appendAgentEvents(agentPubkey, events);
+  if (added) {
+    notifyListeners({ agentPubkey, events: added });
   }
 }
 
@@ -840,8 +862,9 @@ export function syncAgentObserverEvents(
   agentPubkey: string,
   events: ObserverEvent[],
 ) {
-  if (appendAgentEvents(agentPubkey, events)) {
-    notifyListeners();
+  const added = appendAgentEvents(agentPubkey, events);
+  if (added) {
+    notifyListeners({ agentPubkey, events: added });
   }
 }