perf(desktop): tier the observer live-event window by viewer pin

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 <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
2026-08-11 14:52:42 -04:00
parent 44d608416d
commit 5b10fe02d9
4 changed files with 511 additions and 4 deletions

View File

@@ -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<string, number>();
/**
* 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;

View File

@@ -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",
);
});
});

View File

@@ -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",
);
});
});

View File

@@ -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],