From 5b10fe02d99eb75f382855183ef89d45943432ae Mon Sep 17 00:00:00 2001 From: Duncan Date: Tue, 11 Aug 2026 14:52:42 -0400 Subject: [PATCH] perf(desktop): tier the observer live-event window by viewer pin MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-agent live-event stores (`eventsByAgent`/`transcriptByAgent`/the memoized snapshot) each kept the full MAX_OBSERVER_EVENTS window, and a long-lived session observes many agents over time — so the accumulator grew with the number of distinct agents seen and never shrank, one of the renderer leaks behind the v0.5.9 lag. Tier the window by whether a session viewer is mounted. A viewer (the two session panels and the activity bar, via useObserverEvents/useAgentTranscript) pins its agent with a refcount, keeping the full window so the deep scroll-back it displays is never truncated under it. An unpinned agent — observed only by background consumers that never display the transcript — is bounded to UNPINNED_AGENT_EVENT_TAIL. Unpinning the last viewer immediately re-bounds the window it accumulated, rebuilding events, transcript, and snapshot atomically before notifying so no consumer sees a transcript referencing dropped events. The tail is safe for every non-viewer consumer: preventSleep reads only the newest event (always retained); the active-turns bridge processes each event once as it arrives, with turn state living in its own store and gated by a per-channel watermark, so a turn_started aging out of the tail cannot lose the turn; and the profile activity feed derives a channel set that degrades to the tail (its durable history is channel-scoped in SQLite). Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../src/features/agents/observerRelayStore.ts | 91 ++++++- .../agents/observerTieredWindow.test.mjs | 255 ++++++++++++++++++ .../agents/ui/useObserverAgentPin.test.mjs | 148 ++++++++++ .../features/agents/ui/useObserverEvents.ts | 21 ++ 4 files changed, 511 insertions(+), 4 deletions(-) create mode 100644 desktop/src/features/agents/observerTieredWindow.test.mjs create mode 100644 desktop/src/features/agents/ui/useObserverAgentPin.test.mjs diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index f4ba1509df..c119a9e27d 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -30,6 +30,17 @@ import { selectArchiveEvictionKeys } from "./lib/observerArchiveEviction"; const MAX_OBSERVER_EVENTS = 3000; const MAX_PENDING_UNKNOWN_AGENT_FRAMES = 100; +// Live-event window retained in RAM for an agent that has NO mounted session +// viewer. The full MAX_OBSERVER_EVENTS window is kept only while a viewer is +// pinned (see pinObserverAgent); an unpinned agent keeps just the newest tail. +// A long-lived session observes many agents over time and each unpinned agent +// would otherwise hold its full window forever — the accumulator this bounds. +// The tail must stay large enough that the non-viewer consumers keep working: +// preventSleep reads only the newest event, the active-turns bridge processes +// each event once as it arrives (turn state then lives in its own store), and +// the profile activity feed derives a channel set that degrades to the tail. +const UNPINNED_AGENT_EVENT_TAIL = 100; + // Maximum number of distinct channels whose archive scroll-back is retained in // RAM. The archive window grows only by explicit paged loads from SQLite, so // without a bound a session that visits many channels accumulates their full @@ -114,6 +125,78 @@ export function unpinArchiveChannel( } } +// Agents whose full live-event window must be retained because a session viewer +// is mounted on them (the two session panels and the activity bar). Refcounted +// so co-mounted viewers of the same agent compose; the agent keeps its full +// window until the last viewer unmounts. Keyed by normalized pubkey. An agent +// with a positive count is pinned; an unpinned agent's window is bounded to +// UNPINNED_AGENT_EVENT_TAIL. Consumers that read the window WITHOUT displaying +// the transcript (preventSleep, the active-turns bridge, the profile activity +// feed) intentionally do NOT pin — they tolerate the tail by design. +const viewerPinnedAgentCounts = new Map(); + +/** + * Pin an agent's full live-event window against tail-bounding while a session + * viewer is mounted on it. Refcounted so multiple mounted viewers of the same + * agent compose. No-op for a null/empty pubkey (viewers mount before an agent + * resolves). Pinning does not itself grow the window — it lifts the cap so + * subsequent events accumulate up to MAX_OBSERVER_EVENTS. + */ +export function pinObserverAgent(agentPubkey: string | null | undefined): void { + if (!agentPubkey) return; + const key = normalizePubkey(agentPubkey); + viewerPinnedAgentCounts.set(key, (viewerPinnedAgentCounts.get(key) ?? 0) + 1); +} + +/** + * Release one pin acquired via `pinObserverAgent`. When the refcount reaches + * zero the agent's window is immediately bounded back to + * UNPINNED_AGENT_EVENT_TAIL (see `truncateUnpinnedAgentWindow`) so closing a + * viewer reclaims the deep scroll-back it accumulated. No-op for a null/empty + * pubkey or an agent with no outstanding pins. + */ +export function unpinObserverAgent( + agentPubkey: string | null | undefined, +): void { + if (!agentPubkey) return; + const key = normalizePubkey(agentPubkey); + const current = viewerPinnedAgentCounts.get(key); + if (current === undefined) return; + if (current <= 1) { + viewerPinnedAgentCounts.delete(key); + truncateUnpinnedAgentWindow(key); + } else { + viewerPinnedAgentCounts.set(key, current - 1); + } +} + +/** Retention cap for one agent: full window while a viewer is pinned, else the newest-N tail. */ +function agentEventCap(key: string): number { + return viewerPinnedAgentCounts.has(key) + ? MAX_OBSERVER_EVENTS + : UNPINNED_AGENT_EVENT_TAIL; +} + +/** + * Bound a now-unpinned agent's event window to UNPINNED_AGENT_EVENT_TAIL, + * dropping the deep scroll-back its viewer accumulated. Atomic with the derived + * state: the event array, its transcript, and the memoized snapshot are all + * rebuilt from the same truncated window before listeners are notified, so no + * consumer can observe a transcript that references dropped events. No-op when + * the window already fits, so unpinning an agent that never grew is free. + */ +function truncateUnpinnedAgentWindow(key: string): void { + const current = eventsByAgent.get(key); + if (!current || current.length <= UNPINNED_AGENT_EVENT_TAIL) { + return; + } + const trimmed = current.slice(current.length - UNPINNED_AGENT_EVENT_TAIL); + eventsByAgent.set(key, trimmed); + transcriptByAgent.set(key, buildTranscriptState(trimmed)); + invalidateSnapshot(key); + notifyListeners(); +} + // Per-agent, per-channel latest-live-session-id. // Key: `${normalizePubkey(agentPubkey)}:${channelId}`. // Set when a live relay observer event with a sessionId arrives. @@ -273,10 +356,9 @@ function appendAgentEvents( const sortedAdded = added.sort(compareObserverEvents); const sorted = [...current, ...sortedAdded].sort(compareObserverEvents); - const trimmed = sorted.length > MAX_OBSERVER_EVENTS; - const final = trimmed - ? sorted.slice(sorted.length - MAX_OBSERVER_EVENTS) - : sorted; + const cap = agentEventCap(key); + const trimmed = sorted.length > cap; + const final = trimmed ? sorted.slice(sorted.length - cap) : sorted; eventsByAgent.set(key, final); // The common live path appends a sorted batch after the retained window. Fold @@ -909,6 +991,7 @@ export function resetAgentObserverStore() { archiveAccessSeq.clear(); archiveAccessCounter = 0; pinnedArchiveChannelCounts.clear(); + viewerPinnedAgentCounts.clear(); knownAgentPubkeys.clear(); knownAgentsBySubscription.clear(); pendingUnknownAgentFrames.length = 0; diff --git a/desktop/src/features/agents/observerTieredWindow.test.mjs b/desktop/src/features/agents/observerTieredWindow.test.mjs new file mode 100644 index 0000000000..d618cd217a --- /dev/null +++ b/desktop/src/features/agents/observerTieredWindow.test.mjs @@ -0,0 +1,255 @@ +/** + * Store-level tests for the tiered observer live-event window. + * + * The observer store keeps the full MAX_OBSERVER_EVENTS window only while a + * session viewer is pinned to an agent (pinObserverAgent). An UNPINNED agent — + * one observed only by background consumers, never displayed — is bounded to + * UNPINNED_AGENT_EVENT_TAIL. This is the accumulator Option B bounds: a + * long-lived session observes many agents over time, and without the tail each + * would hold its full window forever. + * + * These exercise the REAL ingestion path (injectObserverEventsForE2E → + * appendAgentEvent → the same trim/transcript rebuild the live relay uses) plus + * the public pin API, asserting: + * + * - an unpinned agent's window truncates to the tail, newest events kept; + * - a pinned agent retains its full window (the cap is lifted, not applied); + * - unpinning the last viewer immediately bounds the window it accumulated; + * - the pin is refcounted — two viewers hold until the last release; + * - the newest event always survives (preventSleep reads only events[last]); + * - active turns survive event-array eviction (the CRITICAL constraint: the + * active-turns bridge processes each event once as it arrives, so a + * turn_started aging out of the tail cannot lose the turn); + * - resetAgentObserverStore clears pins so they don't leak across sessions. + * + * No React/Tauri runtime. The tail (UNPINNED_AGENT_EVENT_TAIL = 100) and the + * full cap (MAX_OBSERVER_EVENTS = 3000) are private constants; tests drive + * counts relative to the tail and stay well under the full cap so a pinned + * agent's retained count proves the cap was lifted rather than merely not hit. + */ + +import assert from "node:assert/strict"; +import { beforeEach, describe, it } from "node:test"; + +import { + getAgentObserverSnapshot, + injectObserverEventsForE2E, + pinObserverAgent, + resetAgentObserverStore, + subscribeAgentObserverStore, + unpinObserverAgent, +} from "@/features/agents/observerRelayStore.ts"; +import { + getActiveTurnsForAgent, + resetActiveAgentTurnsStore, + syncActiveAgentTurnsFromObserver, +} from "@/features/agents/activeAgentTurnsStore.ts"; + +const TAIL = 100; // UNPINNED_AGENT_EVENT_TAIL (private constant, mirrored here) +const OVER_TAIL = TAIL + 50; // enough to force truncation, far under the 3000 cap + +/** 64-hex agent pubkey from a short label. */ +function agentPubkey(label) { + return label.padEnd(64, "0"); +} + +const AGENT = agentPubkey("a"); + +/** + * One live observer event. Monotonic timestamp keyed to seq so the store's + * timestamp-then-seq sort matches insertion order — the newest event is always + * the highest seq. + */ +function makeEvent(seq, overrides = {}) { + return { + seq, + timestamp: new Date(1_000_000 + seq * 1000).toISOString(), + kind: "acp_write", + agentIndex: 0, + channelId: "chan-1", + sessionId: "sess-1", + turnId: "turn-1", + payload: {}, + ...overrides, + }; +} + +/** Inject `count` events (seq 1..count) one at a time, mirroring the live path + * where each relay frame appends and notifies individually. */ +function injectSequential(agent, count, makeAt = makeEvent) { + for (let seq = 1; seq <= count; seq++) { + injectObserverEventsForE2E(agent, [makeAt(seq)]); + } +} + +function snapshotEvents(agent) { + return getAgentObserverSnapshot(agent, true).events; +} + +describe("observer tiered live-event window — store level", () => { + beforeEach(() => { + resetAgentObserverStore(); + resetActiveAgentTurnsStore(); + }); + + it("test_unpinned_agent_window_truncates_to_tail", () => { + injectSequential(AGENT, OVER_TAIL); + + const events = snapshotEvents(AGENT); + assert.equal( + events.length, + TAIL, + "an unpinned agent's window must be bounded to the tail", + ); + assert.equal( + events[events.length - 1].seq, + OVER_TAIL, + "the newest event must be retained", + ); + assert.equal( + events[0].seq, + OVER_TAIL - TAIL + 1, + "the oldest events beyond the tail must be dropped", + ); + }); + + it("test_pinned_agent_retains_full_window", () => { + // Pinning lifts the cap to MAX_OBSERVER_EVENTS. OVER_TAIL is far under it, + // so a pinned agent keeps every event — proving the pin lifted the cap + // rather than the count merely not reaching it. + pinObserverAgent(AGENT); + injectSequential(AGENT, OVER_TAIL); + + const events = snapshotEvents(AGENT); + assert.equal( + events.length, + OVER_TAIL, + "a pinned agent retains its full window, well past the unpinned tail", + ); + assert.equal(events[events.length - 1].seq, OVER_TAIL); + assert.equal(events[0].seq, 1, "the oldest event is still present"); + }); + + it("test_unpin_last_viewer_truncates_existing_window", () => { + // A viewer accumulates a deep window; closing it must immediately reclaim + // the scroll-back down to the tail (not wait for the next append to trim). + pinObserverAgent(AGENT); + injectSequential(AGENT, OVER_TAIL); + assert.equal(snapshotEvents(AGENT).length, OVER_TAIL, "full while pinned"); + + unpinObserverAgent(AGENT); + + const events = snapshotEvents(AGENT); + assert.equal( + events.length, + TAIL, + "unpinning the last viewer bounds the window it accumulated", + ); + assert.equal( + events[events.length - 1].seq, + OVER_TAIL, + "truncation keeps the newest events, drops the oldest", + ); + }); + + it("test_refcount_two_viewers_stays_pinned_until_last_unpin", () => { + // Two mounted viewers of the same agent compose. The window stays full + // until the SECOND unpin — a refcount, not a boolean flag. + pinObserverAgent(AGENT); + pinObserverAgent(AGENT); + injectSequential(AGENT, OVER_TAIL); + + unpinObserverAgent(AGENT); + assert.equal( + snapshotEvents(AGENT).length, + OVER_TAIL, + "still pinned while one viewer remains", + ); + + unpinObserverAgent(AGENT); + assert.equal( + snapshotEvents(AGENT).length, + TAIL, + "window bounded only after the last viewer unpins", + ); + }); + + it("test_newest_event_survives_tail_for_prevent_sleep", () => { + // preventSleepActivity reads ONLY events[events.length - 1]. The tail must + // never drop the newest event, or wake-lock refresh would stall. + injectSequential(AGENT, OVER_TAIL); + + const events = snapshotEvents(AGENT); + const newest = events[events.length - 1]; + assert.equal(newest.seq, OVER_TAIL, "newest event survives the tail"); + assert.equal( + newest.timestamp, + makeEvent(OVER_TAIL).timestamp, + "the retained newest event is the last one appended", + ); + }); + + it("test_active_turns_survive_event_eviction", () => { + // The CRITICAL constraint. The active-turns bridge subscribes to the store + // and drains the snapshot on every notify — so each event is processed once + // as it arrives (as the newest event), advancing the per-(agent,channel) + // watermark. A turn_started can then age out of the tail without losing the + // turn: turn state lives in activeAgentTurnsStore, independent of the event + // array, and the watermark blocks any re-processing that never comes. + const RUNNING = [{ pubkey: AGENT, status: "running" }]; + const unsubscribe = subscribeAgentObserverStore(() => { + syncActiveAgentTurnsFromObserver(RUNNING); + }); + + try { + // A turn starts. The bridge drains this notify and marks the turn active. + injectObserverEventsForE2E(AGENT, [ + makeEvent(1, { kind: "turn_started", channelId: "c1", turnId: "t1" }), + ]); + assert.ok( + getActiveTurnsForAgent(AGENT).some((t) => t.channelId === "c1"), + "turn is active immediately after turn_started", + ); + + // Flood the window past the tail so turn_started ages out. Each event + // notifies, draining the bridge — but the watermark is already past + // turn_started, so it is never reprocessed and the turn is not disturbed. + for (let seq = 2; seq <= OVER_TAIL + 1; seq++) { + injectObserverEventsForE2E(AGENT, [ + makeEvent(seq, { kind: "acp_write", channelId: "c1", turnId: "t1" }), + ]); + } + + // The event array has evicted turn_started... + const events = snapshotEvents(AGENT); + assert.equal(events.length, TAIL, "window is bounded to the tail"); + assert.ok( + !events.some((e) => e.kind === "turn_started"), + "turn_started has aged out of the event array", + ); + + // ...but the turn is still active. Truncation cannot lose turns. + assert.ok( + getActiveTurnsForAgent(AGENT).some((t) => t.channelId === "c1"), + "the active turn survives event-array eviction", + ); + } finally { + unsubscribe(); + } + }); + + it("test_reset_clears_pins", () => { + // A pin outstanding at reset must not survive it — otherwise a phantom pin + // would keep the next session's same-key agent unbounded. + pinObserverAgent(AGENT); + resetAgentObserverStore(); + + // If the pin leaked, the window below would stay full at OVER_TAIL. + injectSequential(AGENT, OVER_TAIL); + assert.equal( + snapshotEvents(AGENT).length, + TAIL, + "pin state must not survive reset — the agent is unpinned again", + ); + }); +}); diff --git a/desktop/src/features/agents/ui/useObserverAgentPin.test.mjs b/desktop/src/features/agents/ui/useObserverAgentPin.test.mjs new file mode 100644 index 0000000000..12408cca0b --- /dev/null +++ b/desktop/src/features/agents/ui/useObserverAgentPin.test.mjs @@ -0,0 +1,148 @@ +/** + * Mounted-hook lifecycle test for the observer-agent viewer pin. + * + * The tiered live-event window (observerRelayStore) keeps an agent's full + * MAX_OBSERVER_EVENTS window only while a session viewer is pinned; an unpinned + * agent is bounded to UNPINNED_AGENT_EVENT_TAIL. The pin/unpin lifecycle is + * owned by the viewer hooks: useObserverEvents (session panel) and + * useAgentTranscript (activity bar). This mounts the REAL hooks in a React tree + * and asserts the wiring end to end: + * + * - mounting a viewer pins the agent, so its window accumulates the full + * window (well past the unpinned tail); + * - unmounting the viewer unpins and immediately bounds the window back to + * the tail — reclaiming the deep scroll-back the panel accumulated. + * + * This is a mutation test for the useObserverAgentPin wiring: remove the pin + * from the hook and the mounted-window assertion fails (window would be the + * tail); remove the unpin/truncate and the post-unmount assertion fails + * (window would stay full). + * + * The hooks are mounted with `enabled = false`: the pin is keyed on pubkey and + * is independent of `enabled` (which gates only the live relay subscription), + * so the pin still fires while the relay/Tauri subscription side effect stays + * dormant — no IPC mock needed. + * + * DOM shim: react-dom/client needs a minimal DOM; installDOMShim (from the + * shared observed-unread harness) provides it, matching the convention used by + * the other mounted-hook suites in this package. + */ + +import assert from "node:assert/strict"; +import { beforeEach, describe, it } from "node:test"; + +import { installDOMShim } from "@/features/channels/observedUnreadTestHarness.mjs"; + +installDOMShim(); + +import React from "react"; +import { createRoot } from "react-dom/client"; +import { act } from "react"; + +import { + getAgentObserverSnapshot, + injectObserverEventsForE2E, + resetAgentObserverStore, +} from "@/features/agents/observerRelayStore.ts"; +import { + useAgentTranscript, + useObserverEvents, +} from "@/features/agents/ui/useObserverEvents.ts"; + +const TAIL = 100; // UNPINNED_AGENT_EVENT_TAIL (private constant, mirrored here) +const OVER_TAIL = TAIL + 50; // forces truncation when unpinned; far under 3000 + +const AGENT = "a".repeat(64); + +function makeEvent(seq) { + return { + seq, + timestamp: new Date(1_000_000 + seq * 1000).toISOString(), + kind: "acp_write", + agentIndex: 0, + channelId: "chan-1", + sessionId: "sess-1", + turnId: "turn-1", + payload: {}, + }; +} + +// Injection notifies store listeners, which re-renders the mounted viewer's +// useSyncExternalStore. Wrap in act() so those commits are flushed inside the +// React act scope rather than warning about unbatched updates. +async function injectSequential(agent, count) { + await act(async () => { + for (let seq = 1; seq <= count; seq++) { + injectObserverEventsForE2E(agent, [makeEvent(seq)]); + } + }); +} + +function windowLength(agent) { + return getAgentObserverSnapshot(agent, true).events.length; +} + +/** + * Mount a viewer hook (enabled = false so only the pin fires) and return an + * unmount fn. `useHook` is the viewer hook under test. + */ +async function mountViewer(useHook) { + function Harness() { + useHook(false, AGENT); + return null; + } + const root = createRoot(document.createElement("div")); + await act(async () => { + root.render(React.createElement(Harness)); + }); + return async () => { + await act(async () => { + root.unmount(); + }); + }; +} + +describe("useObserverAgentPin — mounted viewer lifecycle", () => { + beforeEach(() => { + resetAgentObserverStore(); + }); + + it("test_mounting_useObserverEvents_pins_agent_then_unmount_truncates", async () => { + const unmount = await mountViewer(useObserverEvents); + + await injectSequential(AGENT, OVER_TAIL); + assert.equal( + windowLength(AGENT), + OVER_TAIL, + "a mounted session-panel viewer pins the agent — the full window accumulates", + ); + + await unmount(); + assert.equal( + windowLength(AGENT), + TAIL, + "unmounting the viewer unpins and bounds the window back to the tail", + ); + }); + + it("test_mounting_useAgentTranscript_pins_agent_then_unmount_truncates", async () => { + // useAgentTranscript is the activity-bar entry point (BotActivityBar), so + // it must pin too — otherwise a visible activity bar would show a + // tail-truncated transcript. + const unmount = await mountViewer(useAgentTranscript); + + await injectSequential(AGENT, OVER_TAIL); + assert.equal( + windowLength(AGENT), + OVER_TAIL, + "a mounted activity-bar viewer pins the agent — the full window accumulates", + ); + + await unmount(); + assert.equal( + windowLength(AGENT), + TAIL, + "unmounting the activity-bar viewer bounds the window back to the tail", + ); + }); +}); diff --git a/desktop/src/features/agents/ui/useObserverEvents.ts b/desktop/src/features/agents/ui/useObserverEvents.ts index 873013897c..ecd09813d9 100644 --- a/desktop/src/features/agents/ui/useObserverEvents.ts +++ b/desktop/src/features/agents/ui/useObserverEvents.ts @@ -7,8 +7,10 @@ import { getArchivedChannelEvents, ingestArchivedObserverEvents, pinArchiveChannel, + pinObserverAgent, subscribeAgentObserverStore, unpinArchiveChannel, + unpinObserverAgent, } from "@/features/agents/observerRelayStore"; import { listSaveSubscriptions, @@ -33,10 +35,27 @@ export type { ArchivePagingState } from "./archivePagingState"; const subscribeToStore = (onStoreChange: () => void) => subscribeAgentObserverStore(onStoreChange); +/** + * Pin an agent's full live-event window while a session viewer is mounted, so + * the deep scroll-back it displays is not tail-bounded out from under it. + * Refcounted in the store, so the two session panels and the activity bar + * compose when viewing the same agent. Keyed on pubkey only; `enabled` gates + * the relay subscription, not the display, so an idle-agent viewer still pins. + */ +function useObserverAgentPin(agentPubkey: string | null | undefined): void { + React.useEffect(() => { + if (!agentPubkey) return; + pinObserverAgent(agentPubkey); + return () => unpinObserverAgent(agentPubkey); + }, [agentPubkey]); +} + export function useObserverEvents( enabled: boolean, agentPubkey?: string | null, ) { + useObserverAgentPin(agentPubkey); + const getSnapshot = React.useCallback( () => getAgentObserverSnapshot(agentPubkey, enabled), [agentPubkey, enabled], @@ -57,6 +76,8 @@ export function useAgentTranscript( enabled: boolean, agentPubkey?: string | null, ): TranscriptItem[] { + useObserverAgentPin(agentPubkey); + const getSnapshot = React.useCallback( () => getAgentTranscript(agentPubkey, enabled), [agentPubkey, enabled],