fix(desktop): amortize observer journal eviction with a low-water mark (#5808)

Refs #5718.

## What happens

`appendAgentEvents` evicts the per-agent live observer journal back to
*exactly* `MAX_OBSERVER_EVENTS`:

```ts
const trimmed = sorted.length > MAX_OBSERVER_EVENTS;
const final = trimmed ? sorted.slice(sorted.length - MAX_OBSERVER_EVENTS) : sorted;
```

Once an agent's journal reaches 3000, `current.length` is 3000 forever,
every later append makes `sorted.length >= 3001`, and `trimmed` is
`true` on every call. That permanently disables the incremental-fold
gate:

```ts
if (allAtEnd && !trimmed) { /* incremental fold */ }
else { transcriptByAgent.set(key, buildTranscriptState(final)); }
```

So every steady-state append then replays the whole retained window
through `buildTranscriptState`, which is itself O(streamed-text) because
streaming chunks fold as uncapped string concat. Nothing shrinks
`eventsByAgent` except a store reset, so the state is permanent for the
life of the renderer process, per agent. At ~90 frames/min an agent
crosses the cap in ~33 minutes; from then on live CPU escalates (issue
receipts: 188x on a headless ingest, renderer CPU climbing to 119% of a
core after five minutes idle).

This is not an off-by-one — a cap of 3000 does want `>`. The defect is
that trimming *to* the cap re-arms eviction on the very next append, and
eviction is what forces the replay.

## Fix

Evict to a low-water mark below the cap:

```ts
const OBSERVER_EVENTS_LOW_WATER = Math.floor(MAX_OBSERVER_EVENTS * 0.9);
```

The journal still never exceeds `MAX_OBSERVER_EVENTS`; it now has to be
refilled by ~300 ordinary appends before the next eviction, so one
replay is amortized across the appends that refill it. Retention
semantics (newest-N at trim time) and the derived transcript are
unchanged. The mark is a **fraction of the cap** rather than a fixed
count so the math stays correct if the cap is ever made per-agent — a
fixed headroom could exceed a smaller cap and drive the slice length
negative.

### Eviction floor

Low-water eviction leaves headroom below the cap, and the dedup set is
built only from the *retained* array — so once eviction discards the
oldest frames, the journal no longer remembers them. A relay reconnect
replaying a pre-eviction frame (normal relay behavior, and the reason
the dedup set exists) would be re-admitted into the headroom, and a
later refill to the cap would then trim away up to 300 legitimate
retained events with **no new activity** — a bounded display-window loss
plus rebuild churn that partially defeats the amortization.

To close that, each agent carries an **eviction floor**: the ordering
key of the newest event eviction has ever discarded
(`evictionFloorByAgent`, recorded at trim time as the entry just below
the retained window). `appendAgentEvents` rejects any arrival at or
before the floor (`isObserverEventAfter`, so an equal key is rejected —
the floor event itself was evicted); a stale-only batch returns `false`
with no rebuild and no notify. Out-of-order frames *newer* than the
floor are still admitted via the rebuild fallback, so the fold-gate
semantics are unchanged. The floor is cleared in
`resetAgentObserverStore` alongside the other per-agent maps.

## Evidence

`observerTranscriptRetention.test.mjs` asserts the retention window's
**shape** — the observable signal for which ingest path runs, since
transcript *content* is identical on both paths by design — plus
boundary cases and the invariant that the derived transcript still
equals a full replay of the retained window.

Against the pre-fix trim-to-cap shape, three tests fail on the mechanism
itself (`test_append_crossing_cap_trims_to_exactly_low_water`,
`test_headroom_refills_before_next_eviction`,
`test_single_batch_larger_than_cap_trims_to_low_water` — each expects
headroom the old shape never leaves), and the cost shows up directly in
runtime:

| | `observerTranscriptRetention.test.mjs` (single-event appends past
the cap) |
|---|---|
| trim-to-cap (pre-fix) | **429,105 ms** |
| this branch | **16,221 ms** |

~26x on this workload, consistent with the 188x the issue measured on a
heavier one (their events accumulate streaming text; these do not, so
this understates it).

Three further tests pin the **eviction floor** against reconnect replay:
a replay of already-evicted frames leaves the retained window
byte-identical and notifies no listener; a pre-floor frame arriving
after a refill to the cap drops no retained events; and an out-of-order
frame *newer* than the floor is still admitted. Deleting the floor check
turns exactly the first two red while the out-of-order case stays green
— confirming the tests pin the floor's rejection without
over-constraining legitimate out-of-order delivery.

## Merge-order note

This PR collides with #5596 (bounded renderer accumulators) on
`observerRelayStore.ts` by design — #5596 refactors this exact eviction
into `mergeObserverEventBatch` in a new `observerEventOrdering.ts` and
adds a second, unpinned-agent tier (`truncateUnpinnedAgentWindow`,
`UNPINNED_AGENT_EVENT_TAIL`). This PR merges first; #5596 rebases over
it, porting the low-water cap-math **and the per-agent eviction floor**
into `mergeObserverEventBatch`, and applying the same headroom to the
unpinned-tier truncate (which must also record a floor when it trims).
The fraction-of-cap form makes the low-water port mechanical — it feeds
either the 3000 pinned cap or the 100 unpinned tail without a
fixed-count underflow.

## Credits

Supersedes #5767 (Chessing234's low-water-mark approach and the runtime
measurements).

Closes #5718. Issue receipts from the reporter, GeneralJah215 (188x
headless, 119%/core after 5min idle).

---------

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: Duncan <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>

desktop/src/features/agents/observerRelayStore.ts

@@ -27,6 +27,15 @@ import {
 } from "./ui/agentSessionTranscript";
 
 const MAX_OBSERVER_EVENTS = 3000;
+// Length the per-agent journal is evicted down to when it overflows
+// MAX_OBSERVER_EVENTS. Eviction rebuilds the transcript from the retained
+// window (see appendAgentEvents), so trimming back to exactly the cap re-arms
+// eviction on the very next append — every steady-state append then replays the
+// whole history. Leaving 10% headroom amortizes one rebuild across the ~300
+// appends that refill it, while keeping the window within the cap. Expressed as
+// a fraction (not a fixed count) so the same math stays correct if the cap is
+// ever made per-agent, where a fixed headroom could exceed a smaller cap.
+const OBSERVER_EVENTS_LOW_WATER = Math.floor(MAX_OBSERVER_EVENTS * 0.9);
 const MAX_PENDING_UNKNOWN_AGENT_FRAMES = 100;
 
 export type ObserverSnapshot = {
@@ -49,6 +58,20 @@ const eventsByAgent = new Map<string, ObserverEvent[]>();
 const transcriptByAgent = new Map<string, TranscriptState>();
 const snapshotByAgent = new Map<string, ObserverSnapshot>();
 
+// Per-agent eviction floor: the ordering key of the newest event that eviction
+// has ever discarded for this agent. Once the journal is trimmed to the
+// low-water mark, the dedup set (built only from the retained array) no longer
+// remembers the discarded frames, so a delayed/replayed relay frame at or below
+// that boundary would be re-admitted into the headroom — and a later refill to
+// the cap would then trim away 300 legitimate retained events with no new
+// activity. The floor rejects any arrival at or before it (equal included: the
+// floor event itself was evicted), so already-evicted history can never
+// re-enter. Cleared with the observer store; only advances forward.
+const evictionFloorByAgent = new Map<
+  string,
+  { timestamp: string; seq: number }
+>();
+
 // Channel-scoped archive event journal — holds paged history loaded from the local
 // SQLite archive without the MAX_OBSERVER_EVENTS live-relay cap. Keyed by
 // `${normalizedAgentPubkey}:${channelId}`. The live relay path writes to
@@ -201,13 +224,25 @@ function appendAgentEvents(
 
   const key = normalizePubkey(agentPubkey);
   const current = eventsByAgent.get(key) ?? [];
+
+  // Reject any arrival at or before the eviction floor: those frames were
+  // already discarded, so re-admitting them (they fit within the headroom
+  // below the cap) would let a later refill trim away legitimate retained
+  // events. Admit only frames strictly after the floor — the floor event
+  // itself was evicted, so an equal ordering key is rejected too.
+  const floor = evictionFloorByAgent.get(key);
+  const admissible = floor
+    ? events.filter((event) => isObserverEventAfter(event, floor))
+    : events;
+  if (admissible.length === 0) return false;
+
   const seen = new Set(
     current.map(
       (event) => `${event.timestamp.length}:${event.timestamp}:${event.seq}`,
     ),
   );
   const added: ObserverEvent[] = [];
-  for (const event of events) {
+  for (const event of admissible) {
     const eventKey = `${event.timestamp.length}:${event.timestamp}:${event.seq}`;
     if (seen.has(eventKey)) continue;
     seen.add(eventKey);
@@ -219,10 +254,21 @@ function appendAgentEvents(
   const sorted = [...current, ...sortedAdded].sort(compareObserverEvents);
   const trimmed = sorted.length > MAX_OBSERVER_EVENTS;
   const final = trimmed
-    ? sorted.slice(sorted.length - MAX_OBSERVER_EVENTS)
+    ? sorted.slice(sorted.length - OBSERVER_EVENTS_LOW_WATER)
     : sorted;
   eventsByAgent.set(key, final);
 
+  // Record the newest event this trim discarded as the agent's eviction floor.
+  // It is the entry just below the retained window; the floor only advances,
+  // since the retained window is always the newest tail.
+  if (trimmed) {
+    const boundary = sorted[sorted.length - OBSERVER_EVENTS_LOW_WATER - 1];
+    evictionFloorByAgent.set(key, {
+      timestamp: boundary.timestamp,
+      seq: boundary.seq,
+    });
+  }
+
   // The common live path appends a sorted batch after the retained window. Fold
   // that batch through the transcript state once without rebuilding history.
   // Out-of-order arrivals and cap eviction rebuild from the final window so
@@ -807,6 +853,7 @@ export function resetAgentObserverStore() {
   eventProcessingQueue = Promise.resolve();
   eventsByAgent.clear();
   transcriptByAgent.clear();
+  evictionFloorByAgent.clear();
   snapshotByAgent.clear();
   archiveEventsByChannel.clear();
   knownAgentPubkeys.clear();

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

@@ -0,0 +1,305 @@
+/**
+ * Retention behavior of the per-agent live observer journal.
+ *
+ * `appendAgentEvents` derives the transcript incrementally when a batch lands
+ * after the retained window, and falls back to a full `buildTranscriptState`
+ * replay when the window is evicted. Evicting back to *exactly* the cap made
+ * that fallback permanent: an agent parked at the cap evicts one event on every
+ * append, so `trimmed` is true forever and every steady-state append replays the
+ * whole history through `buildTranscriptState` (issue #5718: 188x headless,
+ * live CPU escalating to 119%/core after 5min idle).
+ *
+ * The fix evicts to a low-water mark below the cap, so the window must refill
+ * through ordinary appends before the next eviction — one replay amortized
+ * across the refill. Transcript content is identical on the fold and rebuild
+ * paths (that is the point of the fallback), so these tests assert the
+ * observable that distinguishes them: the retained window's SHAPE after
+ * eviction. They also pin the invariant that the derived transcript still equals
+ * a full replay of the retained window regardless of which path ran.
+ */
+
+import assert from "node:assert/strict";
+import { beforeEach, describe, it } from "node:test";
+
+import {
+  getAgentObserverSnapshot,
+  getAgentTranscript,
+  resetAgentObserverStore,
+  subscribeAgentObserverStore,
+  syncAgentObserverEvents,
+} from "@/features/agents/observerRelayStore.ts";
+import { buildTranscript } from "@/features/agents/ui/agentSessionTranscript.ts";
+
+// Mirrors the private constants in observerRelayStore.ts. LOW_WATER is
+// Math.floor(MAX * 0.9); the tests assert exact shapes against these values so
+// a regression in the eviction math (e.g. reverting to trim-to-cap) fails here.
+const MAX_OBSERVER_EVENTS = 3000;
+const OBSERVER_EVENTS_LOW_WATER = Math.floor(MAX_OBSERVER_EVENTS * 0.9);
+
+const AGENT_PUBKEY = "a".repeat(64);
+
+/** One live observer event; monotonic timestamp keyed to seq so the store's
+ *  timestamp-then-seq sort matches insertion order. */
+function makeEvent(seq) {
+  return {
+    seq,
+    timestamp: new Date(1_760_000_000_000 + seq * 1000).toISOString(),
+    kind: "turn_started",
+    agentIndex: 0,
+    channelId: "chan-1",
+    sessionId: "sess-1",
+    turnId: `turn-${seq}`,
+    payload: {},
+  };
+}
+
+function windowLength() {
+  return getAgentObserverSnapshot(AGENT_PUBKEY).events.length;
+}
+
+/** Append events seq 1..count one at a time, mirroring the live relay path
+ *  where each frame appends and notifies individually. */
+function fillSequential(count) {
+  for (let seq = 1; seq <= count; seq += 1) {
+    syncAgentObserverEvents(AGENT_PUBKEY, [makeEvent(seq)]);
+  }
+}
+
+describe("live observer journal retention — amortized eviction", () => {
+  beforeEach(() => {
+    resetAgentObserverStore();
+  });
+
+  it("test_window_never_exceeds_cap", () => {
+    for (let seq = 1; seq <= MAX_OBSERVER_EVENTS + 750; seq += 1) {
+      syncAgentObserverEvents(AGENT_PUBKEY, [makeEvent(seq)]);
+      assert.ok(
+        windowLength() <= MAX_OBSERVER_EVENTS,
+        `window grew to ${windowLength()} at seq ${seq}`,
+      );
+    }
+  });
+
+  it("test_append_at_cap_does_not_trim_prematurely", () => {
+    // Filling to exactly the cap must NOT evict — `trimmed` is `length > cap`,
+    // and length === cap is not over. A premature trim here would mean the
+    // fraction math or the comparison regressed.
+    fillSequential(MAX_OBSERVER_EVENTS);
+    assert.equal(
+      windowLength(),
+      MAX_OBSERVER_EVENTS,
+      "reaching exactly the cap retains the full window, no eviction",
+    );
+  });
+
+  it("test_append_crossing_cap_trims_to_exactly_low_water", () => {
+    // The append that pushes past the cap must leave the window at exactly the
+    // low-water mark — not back at the cap (which would re-arm eviction on the
+    // very next append and keep the transcript rebuilding forever).
+    fillSequential(MAX_OBSERVER_EVENTS);
+    syncAgentObserverEvents(AGENT_PUBKEY, [makeEvent(MAX_OBSERVER_EVENTS + 1)]);
+
+    assert.equal(
+      windowLength(),
+      OBSERVER_EVENTS_LOW_WATER,
+      "crossing the cap trims to exactly the low-water mark, leaving headroom",
+    );
+    assert.ok(
+      windowLength() < MAX_OBSERVER_EVENTS,
+      "headroom exists below the cap after eviction",
+    );
+  });
+
+  it("test_headroom_refills_before_next_eviction", () => {
+    // After the first eviction leaves headroom, subsequent appends must GROW
+    // the window (no eviction) until it refills to the cap — proving eviction
+    // is amortized across the refill, not per-append.
+    fillSequential(MAX_OBSERVER_EVENTS + 1);
+    assert.equal(windowLength(), OBSERVER_EVENTS_LOW_WATER);
+
+    const headroom = MAX_OBSERVER_EVENTS - OBSERVER_EVENTS_LOW_WATER;
+    for (let i = 1; i <= headroom; i += 1) {
+      syncAgentObserverEvents(AGENT_PUBKEY, [
+        makeEvent(MAX_OBSERVER_EVENTS + 1 + i),
+      ]);
+      assert.equal(
+        windowLength(),
+        OBSERVER_EVENTS_LOW_WATER + i,
+        `append ${i} into the headroom must grow the window, not evict`,
+      );
+    }
+    // The window is now back at the cap; the next append evicts again.
+    syncAgentObserverEvents(AGENT_PUBKEY, [
+      makeEvent(MAX_OBSERVER_EVENTS + 2 + headroom),
+    ]);
+    assert.equal(
+      windowLength(),
+      OBSERVER_EVENTS_LOW_WATER,
+      "the window only evicts again after the headroom is refilled",
+    );
+  });
+
+  it("test_eviction_keeps_newest_events_drops_oldest", () => {
+    const total = MAX_OBSERVER_EVENTS + 400;
+    fillSequential(total);
+    const events = getAgentObserverSnapshot(AGENT_PUBKEY).events;
+    assert.equal(events.at(-1).seq, total, "newest event is retained");
+    assert.equal(
+      events.at(0).seq,
+      total - events.length + 1,
+      "retention is the newest-N contiguous tail",
+    );
+  });
+
+  it("test_derived_transcript_equals_full_replay_after_eviction", () => {
+    // The rebuild fallback and the incremental fold must agree: after crossing
+    // the cap (rebuild path) the stored transcript equals a fresh replay of the
+    // retained window.
+    fillSequential(MAX_OBSERVER_EVENTS + 600);
+    const retained = getAgentObserverSnapshot(AGENT_PUBKEY).events;
+    assert.deepEqual(
+      getAgentTranscript(AGENT_PUBKEY),
+      buildTranscript(retained),
+      "the derived transcript matches a full replay of the retained window",
+    );
+  });
+
+  it("test_single_batch_larger_than_cap_trims_to_low_water", () => {
+    const batch = [];
+    for (let seq = 1; seq <= MAX_OBSERVER_EVENTS + 900; seq += 1) {
+      batch.push(makeEvent(seq));
+    }
+    syncAgentObserverEvents(AGENT_PUBKEY, batch);
+    assert.equal(
+      windowLength(),
+      OBSERVER_EVENTS_LOW_WATER,
+      "a single over-cap batch also trims to the low-water mark",
+    );
+    assert.equal(
+      getAgentObserverSnapshot(AGENT_PUBKEY).events.at(-1).seq,
+      MAX_OBSERVER_EVENTS + 900,
+      "the newest event of an over-cap batch is retained",
+    );
+  });
+});
+
+describe("live observer journal retention — eviction floor (reconnect replay)", () => {
+  // The dedup set is built only from the retained array, so once eviction
+  // discards the oldest frames the journal no longer remembers them. Relay
+  // reconnect replays old frames as a normal behavior; without a floor a
+  // replayed pre-eviction frame is re-admitted into the headroom and a later
+  // refill to the cap trims away legitimate retained events with no new
+  // activity. These pin the floor that rejects already-evicted history.
+  beforeEach(() => {
+    resetAgentObserverStore();
+  });
+
+  it("test_replayed_pre_floor_frames_do_not_change_retained_window", () => {
+    // Overflow to the low-water mark, then replay the discarded oldest frames.
+    // They are at or below the eviction floor, so none is re-admitted: the
+    // retained window is byte-identical and no listener is notified.
+    fillSequential(MAX_OBSERVER_EVENTS + 1);
+    const before = getAgentObserverSnapshot(AGENT_PUBKEY).events;
+    assert.equal(before.length, OBSERVER_EVENTS_LOW_WATER);
+    const beforeFirstSeq = before.at(0).seq;
+
+    let notifications = 0;
+    const unsubscribe = subscribeAgentObserverStore(() => {
+      notifications += 1;
+    });
+    try {
+      // seq 1..(beforeFirstSeq - 1) were discarded; replay a spread of them
+      // plus the boundary event itself (beforeFirstSeq - 1 is the floor).
+      for (let seq = 1; seq < beforeFirstSeq; seq += 1) {
+        syncAgentObserverEvents(AGENT_PUBKEY, [makeEvent(seq)]);
+      }
+    } finally {
+      unsubscribe();
+    }
+
+    const after = getAgentObserverSnapshot(AGENT_PUBKEY).events;
+    assert.deepEqual(
+      after,
+      before,
+      "replaying already-evicted frames must not change the retained window",
+    );
+    assert.equal(
+      notifications,
+      0,
+      "a stale-only replay does no work and notifies no listener",
+    );
+  });
+
+  it("test_pre_floor_frame_after_refill_drops_no_retained_events", () => {
+    // Overflow to low-water, refill to the cap through ordinary appends, then
+    // inject a single pre-floor (already-evicted) frame. Pre-fix this pushed
+    // length to cap+1 and trimmed away 300 legitimate retained events; the
+    // floor now rejects it, so the retained window is untouched.
+    fillSequential(MAX_OBSERVER_EVENTS + 1);
+    const floorBoundarySeq =
+      getAgentObserverSnapshot(AGENT_PUBKEY).events.at(0).seq;
+    const headroom = MAX_OBSERVER_EVENTS - OBSERVER_EVENTS_LOW_WATER;
+    for (let i = 1; i <= headroom; i += 1) {
+      syncAgentObserverEvents(AGENT_PUBKEY, [
+        makeEvent(MAX_OBSERVER_EVENTS + 1 + i),
+      ]);
+    }
+    const atCap = getAgentObserverSnapshot(AGENT_PUBKEY).events;
+    assert.equal(atCap.length, MAX_OBSERVER_EVENTS, "refilled back to the cap");
+
+    // A pre-floor frame (seq strictly below the boundary that was evicted).
+    syncAgentObserverEvents(AGENT_PUBKEY, [makeEvent(floorBoundarySeq - 1)]);
+
+    const after = getAgentObserverSnapshot(AGENT_PUBKEY).events;
+    assert.equal(
+      after.length,
+      MAX_OBSERVER_EVENTS,
+      "a rejected pre-floor frame must not trigger a trim below the cap",
+    );
+    assert.deepEqual(
+      after,
+      atCap,
+      "no legitimate retained event is dropped by a pre-floor arrival",
+    );
+  });
+
+  it("test_out_of_order_frame_newer_than_floor_is_still_admitted", () => {
+    // The floor must reject only already-evicted history, never a legitimate
+    // out-of-order frame that sorts after the floor. Such a frame lands in the
+    // retained window (via the rebuild fallback) and advances the length.
+    fillSequential(MAX_OBSERVER_EVENTS + 1);
+    const retained = getAgentObserverSnapshot(AGENT_PUBKEY).events;
+    const oldestRetainedSeq = retained.at(0).seq;
+    const lengthBefore = retained.length;
+
+    // The eviction floor is the boundary event just below the retained window
+    // (seq oldestRetainedSeq - 1). Construct a never-seen frame whose timestamp
+    // sits strictly between the floor and the oldest retained event — out of
+    // order versus the tail, but newer than the floor, so it must be admitted.
+    const floorSeq = oldestRetainedSeq - 1;
+    const oooEvent = {
+      seq: 1_000_000_000,
+      timestamp: new Date(
+        1_760_000_000_000 + floorSeq * 1000 + 500,
+      ).toISOString(),
+      kind: "turn_started",
+      agentIndex: 0,
+      channelId: "chan-1",
+      sessionId: "sess-1",
+      turnId: "turn-ooo",
+      payload: {},
+    };
+    syncAgentObserverEvents(AGENT_PUBKEY, [oooEvent]);
+
+    const after = getAgentObserverSnapshot(AGENT_PUBKEY).events;
+    assert.equal(
+      after.length,
+      lengthBefore + 1,
+      "an out-of-order frame newer than the floor is admitted, not rejected",
+    );
+    assert.ok(
+      after.some((event) => event.seq === oooEvent.seq),
+      "the admitted frame is present in the retained window",
+    );
+  });
+});