From 44d608416dddc595457c8e455fda98e81a5aa3fa Mon Sep 17 00:00:00 2001 From: Duncan Date: Tue, 11 Aug 2026 13:57:11 -0400 Subject: [PATCH] perf(desktop): bound observer archive to an LRU channel window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The channel-scoped observer archive (`archiveEventsByChannel`) grew only by explicit paged loads from SQLite and was never evicted, so a long session that visits many channels retains every channel's full scroll-back for the lifetime of the app — a renderer memory leak behind the v0.5.9 lag. Bound it to MAX_RETAINED_ARCHIVE_CHANNELS distinct channels via an LRU policy that evicts whole channels (every agent entry for a channel goes together). Channels with a mounted panel are pinned (refcounted) and never evicted; an evicted channel re-hydrates from SQLite on revisit, so eviction is a cache miss, not data loss. A post-unmount fence stops an in-flight page from resurrecting an evicted channel after its pin is released. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../lib/observerArchiveEviction.test.mjs | 148 ++++++++++ .../agents/lib/observerArchiveEviction.ts | 91 ++++++ .../observerArchiveEvictionStore.test.mjs | 260 ++++++++++++++++++ .../src/features/agents/observerRelayStore.ts | 102 ++++++- .../ui/useLoadArchivedObserverEvents.test.mjs | 77 ++++++ .../features/agents/ui/useObserverEvents.ts | 36 +++ 6 files changed, 713 insertions(+), 1 deletion(-) create mode 100644 desktop/src/features/agents/lib/observerArchiveEviction.test.mjs create mode 100644 desktop/src/features/agents/lib/observerArchiveEviction.ts create mode 100644 desktop/src/features/agents/observerArchiveEvictionStore.test.mjs diff --git a/desktop/src/features/agents/lib/observerArchiveEviction.test.mjs b/desktop/src/features/agents/lib/observerArchiveEviction.test.mjs new file mode 100644 index 0000000000..c08fb94a25 --- /dev/null +++ b/desktop/src/features/agents/lib/observerArchiveEviction.test.mjs @@ -0,0 +1,148 @@ +/** + * Unit tests for selectArchiveEvictionKeys — the pure LRU-with-pins policy that + * bounds the channel-scoped observer archive. + * + * The policy evicts at channel granularity (all agent entries for a channel go + * together) while respecting pins (a pinned channel is never evicted). These + * tests exercise the selection logic directly, with no React/Tauri runtime. + */ + +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; + +import { selectArchiveEvictionKeys } from "./observerArchiveEviction.ts"; + +/** Build one archive entry meta. */ +function meta(agent, channelId, accessSeq) { + return { key: `${agent}:${channelId}`, channelId, accessSeq }; +} + +const NO_PINS = new Set(); + +describe("selectArchiveEvictionKeys", () => { + it("test_empty_entries_returns_no_eviction", () => { + assert.deepEqual(selectArchiveEvictionKeys([], NO_PINS, 8), []); + }); + + it("test_channel_count_below_cap_returns_no_eviction", () => { + const entries = [ + meta("agentA", "chan-1", 1), + meta("agentA", "chan-2", 2), + meta("agentA", "chan-3", 3), + ]; + assert.deepEqual(selectArchiveEvictionKeys(entries, NO_PINS, 8), []); + }); + + it("test_channel_count_exactly_at_cap_returns_no_eviction", () => { + const entries = Array.from({ length: 8 }, (_, i) => + meta("agentA", `chan-${i}`, i), + ); + assert.deepEqual(selectArchiveEvictionKeys(entries, NO_PINS, 8), []); + }); + + it("test_over_cap_evicts_least_recently_accessed_channels", () => { + // 10 channels, cap 8 → evict the 2 lowest-recency (chan-0, chan-1). + const entries = Array.from({ length: 10 }, (_, i) => + meta("agentA", `chan-${i}`, i), + ); + const evicted = selectArchiveEvictionKeys(entries, NO_PINS, 8); + assert.deepEqual(evicted.sort(), ["agentA:chan-0", "agentA:chan-1"].sort()); + }); + + it("test_channel_recency_is_max_access_across_its_agent_entries", () => { + // chan-old has two agent entries; its most-recent access (seq 100) keeps it + // alive even though its other entry (seq 0) is the global minimum. + const entries = [ + meta("agentA", "chan-old", 0), + meta("agentB", "chan-old", 100), + meta("agentA", "chan-1", 1), + meta("agentA", "chan-2", 2), + ]; + // cap 2 → keep the 2 most-recent channels (chan-old@100, chan-2@2), + // evict chan-1@1. + const evicted = selectArchiveEvictionKeys(entries, NO_PINS, 2); + assert.deepEqual(evicted, ["agentA:chan-1"]); + }); + + it("test_evicting_a_channel_removes_all_its_agent_entries", () => { + // chan-evict has 3 agent entries; all three keys must be returned together. + const entries = [ + meta("agentA", "chan-evict", 0), + meta("agentB", "chan-evict", 0), + meta("agentC", "chan-evict", 0), + meta("agentA", "chan-keep", 9), + ]; + const evicted = selectArchiveEvictionKeys(entries, NO_PINS, 1); + assert.deepEqual( + evicted.sort(), + ["agentA:chan-evict", "agentB:chan-evict", "agentC:chan-evict"].sort(), + ); + }); + + it("test_pinned_channel_is_never_evicted_even_when_least_recent", () => { + // chan-pinned is the least-recently accessed but pinned → must survive; + // chan-1 (next lowest) is evicted instead. + const entries = [ + meta("agentA", "chan-pinned", 0), + meta("agentA", "chan-1", 1), + meta("agentA", "chan-2", 2), + meta("agentA", "chan-3", 3), + ]; + const evicted = selectArchiveEvictionKeys( + entries, + new Set(["chan-pinned"]), + 3, + ); + assert.deepEqual(evicted, ["agentA:chan-1"]); + }); + + it("test_pins_exceeding_cap_evict_all_unpinned_but_keep_all_pinned", () => { + // 3 pinned + 2 unpinned, cap 2. Pins alone exceed the cap, so the unpinned + // budget is 0 → both unpinned channels evicted, all pinned retained. + const entries = [ + meta("agentA", "pin-1", 10), + meta("agentA", "pin-2", 11), + meta("agentA", "pin-3", 12), + meta("agentA", "free-1", 1), + meta("agentA", "free-2", 2), + ]; + const evicted = selectArchiveEvictionKeys( + entries, + new Set(["pin-1", "pin-2", "pin-3"]), + 2, + ); + assert.deepEqual(evicted.sort(), ["agentA:free-1", "agentA:free-2"].sort()); + }); + + it("test_pinned_channels_count_against_the_cap_for_unpinned_budget", () => { + // 1 pinned + 3 unpinned, cap 2 → unpinned budget = 2 - 1 = 1. Keep the + // single most-recent unpinned channel (free-3), evict free-1 and free-2. + const entries = [ + meta("agentA", "pin-1", 5), + meta("agentA", "free-1", 1), + meta("agentA", "free-2", 2), + meta("agentA", "free-3", 3), + ]; + const evicted = selectArchiveEvictionKeys(entries, new Set(["pin-1"]), 2); + assert.deepEqual(evicted.sort(), ["agentA:free-1", "agentA:free-2"].sort()); + }); + + it("test_tie_broken_deterministically_by_channel_id", () => { + // Two channels share recency 1 at the eviction boundary. The keep-sort + // breaks ties by ascending channelId, so "chan-a" is retained and the + // higher "chan-b" is evicted — deterministic regardless of input order. + const entries = [ + meta("agentA", "chan-b", 1), + meta("agentA", "chan-a", 1), + meta("agentA", "chan-keep", 9), + ]; + const evicted = selectArchiveEvictionKeys(entries, NO_PINS, 2); + assert.deepEqual(evicted, ["agentA:chan-b"]); + }); + + it("test_zero_cap_evicts_all_unpinned_channels", () => { + const entries = [meta("agentA", "chan-1", 1), meta("agentA", "chan-2", 2)]; + const evicted = selectArchiveEvictionKeys(entries, NO_PINS, 0); + assert.deepEqual(evicted.sort(), ["agentA:chan-1", "agentA:chan-2"].sort()); + }); +}); diff --git a/desktop/src/features/agents/lib/observerArchiveEviction.ts b/desktop/src/features/agents/lib/observerArchiveEviction.ts new file mode 100644 index 0000000000..d38b176171 --- /dev/null +++ b/desktop/src/features/agents/lib/observerArchiveEviction.ts @@ -0,0 +1,91 @@ +/** + * Pure LRU-with-pins eviction policy for the channel-scoped observer archive. + * + * The archive store (`observerRelayStore`) keys paged history by + * `${normalizedAgentPubkey}:${channelId}`, so a single channel observed for + * several agents occupies several map entries. Memory pressure, however, scales + * with the number of distinct *channels* whose scroll-back is retained — the + * user opens channels, not (agent, channel) pairs. This module therefore evicts + * at channel granularity: when the count of distinct retained channels exceeds + * `cap`, the least-recently-accessed unpinned channels are dropped whole (every + * agent entry for that channel). Eviction is not data loss — the next visit + * re-hydrates the channel from SQLite through the normal paging path. + * + * Pins are inviolable: a pinned channel is never selected for eviction, even + * when pins alone exceed `cap`. A pin marks a channel whose panel is mounted and + * whose eviction would blank a live view, so the cap is enforced only against + * unpinned channels. + * + * Pure and React/Tauri-free so the selection logic is unit-testable in + * isolation, mirroring the `archivePagingState` extraction. + */ + +export interface ArchiveEntryMeta { + /** Composite map key `${normalizedAgentPubkey}:${channelId}`. */ + key: string; + /** Channel the entry belongs to; the unit of eviction. */ + channelId: string; + /** Monotonic access counter; higher = more recently read or written. */ + accessSeq: number; +} + +/** + * Select the composite map keys to evict so that no more than `cap` distinct + * unpinned channels are retained. Returns the keys for every entry belonging to + * an evicted channel; pinned channels never appear in the result. + * + * @param entries One meta per archive map entry (composite key granularity). + * @param pinnedChannels Channel ids whose entries must be retained regardless of cap. + * @param cap Maximum distinct unpinned channels to keep. + */ +export function selectArchiveEvictionKeys( + entries: readonly ArchiveEntryMeta[], + pinnedChannels: ReadonlySet, + cap: number, +): string[] { + // Recency of each channel = the most recent access across all its entries, + // so touching any agent's view of a channel refreshes the whole channel. + const channelRecency = new Map(); + for (const entry of entries) { + const prev = channelRecency.get(entry.channelId); + if (prev === undefined || entry.accessSeq > prev) { + channelRecency.set(entry.channelId, entry.accessSeq); + } + } + + // Only unpinned channels are eviction candidates; pins claim retained slots. + const unpinned: Array<{ channelId: string; recency: number }> = []; + let pinnedPresent = 0; + for (const [channelId, recency] of channelRecency) { + if (pinnedChannels.has(channelId)) { + pinnedPresent += 1; + } else { + unpinned.push({ channelId, recency }); + } + } + + const unpinnedBudget = Math.max(0, cap - pinnedPresent); + if (unpinned.length <= unpinnedBudget) { + return []; + } + + // Keep the most-recent `unpinnedBudget` channels; evict the rest. Sort by + // recency descending, breaking ties by channelId so selection is deterministic. + unpinned.sort( + (a, b) => + b.recency - a.recency || + (a.channelId < b.channelId ? -1 : a.channelId > b.channelId ? 1 : 0), + ); + const evictChannels = new Set( + unpinned.slice(unpinnedBudget).map((channel) => channel.channelId), + ); + + // Expand evicted channels back to every composite map key they cover. + const keys: string[] = []; + for (const entry of entries) { + if (evictChannels.has(entry.channelId)) { + keys.push(entry.key); + } + } + return keys; +} diff --git a/desktop/src/features/agents/observerArchiveEvictionStore.test.mjs b/desktop/src/features/agents/observerArchiveEvictionStore.test.mjs new file mode 100644 index 0000000000..bd485c6e58 --- /dev/null +++ b/desktop/src/features/agents/observerArchiveEvictionStore.test.mjs @@ -0,0 +1,260 @@ +/** + * Store-level eviction tests for the channel-scoped observer archive. + * + * These exercise the REAL ingest path (ingestArchivedObserverEvents with an + * injected decrypt fn) plus the public pin API, asserting the LRU bound and its + * invariants at the store boundary — the layer above the pure policy unit: + * + * - the retained-channel cap is enforced across ingest pages; + * - eviction is least-recently-loaded first; + * - a pinned channel is never evicted (even when least recent); + * - evicting a channel drops every (agent, channel) entry it owns; + * - an evicted channel reads back empty (cache miss → re-hydrate on revisit); + * - resetAgentObserverStore clears the LRU bookkeeping so counts don't leak. + * + * No React/Tauri runtime: security guards are satisfied via _testRegisterKnownAgents + * and decrypt is injected. The cap (MAX_RETAINED_ARCHIVE_CHANNELS = 8) is a + * private constant; tests drive enough distinct channels to cross it and assert + * behaviour relative to that threshold rather than importing the value. + */ + +import assert from "node:assert/strict"; +import { beforeEach, describe, it } from "node:test"; + +import { + ingestArchivedObserverEvents, + resetAgentObserverStore, + pinArchiveChannel, + unpinArchiveChannel, + _testRegisterKnownAgents, + _testGetArchivedChannelEvents, +} from "@/features/agents/observerRelayStore.ts"; + +const CAP = 8; // MAX_RETAINED_ARCHIVE_CHANNELS (private constant, mirrored here) +const SUB_ID = "test-eviction-sub"; + +/** 64-hex agent pubkey from a short label. */ +function agentPubkey(label) { + return label.padEnd(64, "0"); +} + +const AGENT_A = agentPubkey("a"); +const AGENT_B = agentPubkey("b"); + +/** Raw kind-24200 telemetry row for (agent, channel); content carries the event. */ +function makeRawEvent(agent, channelId, seq) { + return { + id: `${seq}`.padStart(64, "e"), + pubkey: agent, + created_at: 1000 + seq, + kind: 24200, + tags: [ + ["p", "d".repeat(64)], + ["agent", agent], + ["frame", "telemetry"], + ], + content: JSON.stringify({ + seq, + timestamp: new Date(1_000_000 + seq * 1000).toISOString(), + kind: "acp_write", + agentIndex: 0, + channelId, + sessionId: "sess-1", + turnId: "turn-1", + payload: {}, + }), + sig: "s".repeat(128), + }; +} + +/** Decrypt fn: content is the JSON-encoded ObserverEvent. */ +const decrypt = (event) => Promise.resolve(JSON.parse(event.content)); + +/** Ingest one telemetry frame for (agent, channel), awaiting the store write. */ +async function ingestOne(agent, channelId, seq) { + await ingestArchivedObserverEvents( + [makeRawEvent(agent, channelId, seq)], + decrypt, + ); +} + +describe("observer archive eviction — store level", () => { + beforeEach(() => { + resetAgentObserverStore(); + _testRegisterKnownAgents(SUB_ID, [AGENT_A, AGENT_B]); + }); + + it("test_channels_up_to_cap_are_all_retained", async () => { + for (let i = 0; i < CAP; i++) { + await ingestOne(AGENT_A, `chan-${i}`, i); + } + for (let i = 0; i < CAP; i++) { + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, `chan-${i}`).length, + 1, + `chan-${i} must be retained at exactly cap`, + ); + } + }); + + it("test_exceeding_cap_evicts_least_recently_loaded_channel", async () => { + // Load CAP+1 distinct channels in ascending order. chan-0 is the oldest + // load and unpinned, so it is the one evicted when the cap is crossed. + for (let i = 0; i <= CAP; i++) { + await ingestOne(AGENT_A, `chan-${i}`, i); + } + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-0").length, + 0, + "least-recently-loaded channel must be evicted", + ); + // The newest CAP channels survive. + for (let i = 1; i <= CAP; i++) { + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, `chan-${i}`).length, + 1, + `chan-${i} must survive`, + ); + } + }); + + it("test_pinned_channel_survives_even_when_least_recently_loaded", async () => { + // chan-0 is loaded first (oldest) but pinned; loading CAP more channels + // crosses the cap. The eviction must fall on the oldest UNPINNED channel + // (chan-1), never the pinned chan-0. + await ingestOne(AGENT_A, "chan-0", 0); + pinArchiveChannel("chan-0"); + for (let i = 1; i <= CAP; i++) { + await ingestOne(AGENT_A, `chan-${i}`, i); + } + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-0").length, + 1, + "pinned channel must never be evicted", + ); + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-1").length, + 0, + "oldest unpinned channel must be evicted instead of the pinned one", + ); + }); + + it("test_unpinning_makes_channel_evictable_again", async () => { + // Pin protects chan-0 across a first over-cap wave; after unpinning, a + // second wave that again exceeds the cap must be free to evict chan-0. + await ingestOne(AGENT_A, "chan-0", 0); + pinArchiveChannel("chan-0"); + for (let i = 1; i <= CAP; i++) { + await ingestOne(AGENT_A, `chan-${i}`, i); + } + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-0").length, + 1, + "pinned channel survives the first wave", + ); + + unpinArchiveChannel("chan-0"); + // chan-0 is now the least-recently-loaded unpinned channel. Load fresh + // channels to push the retained set past the cap again. + for (let i = CAP + 1; i <= CAP * 2; i++) { + await ingestOne(AGENT_A, `chan-${i}`, i); + } + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-0").length, + 0, + "unpinned channel becomes evictable and is dropped", + ); + }); + + it("test_refcounted_pin_holds_until_last_release", async () => { + // Two mounted consumers pin chan-0; one unpins. The channel must remain + // protected until the SECOND unpin — refcount, not a boolean flag. + await ingestOne(AGENT_A, "chan-0", 0); + pinArchiveChannel("chan-0"); + pinArchiveChannel("chan-0"); + unpinArchiveChannel("chan-0"); + + for (let i = 1; i <= CAP; i++) { + await ingestOne(AGENT_A, `chan-${i}`, i); + } + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-0").length, + 1, + "channel stays pinned while one consumer still holds a pin", + ); + + unpinArchiveChannel("chan-0"); + for (let i = CAP + 1; i <= CAP * 2; i++) { + await ingestOne(AGENT_A, `chan-${i}`, i); + } + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-0").length, + 0, + "channel is evictable only after the last pin is released", + ); + }); + + it("test_eviction_drops_all_agent_entries_for_the_channel", async () => { + // chan-shared is populated for two agents (fan-out: one channel view + // populates multiple (agent, channel) keys). Both keys must be evicted + // together when the channel is dropped. + await ingestOne(AGENT_A, "chan-shared", 0); + await ingestOne(AGENT_B, "chan-shared", 1); + // Push past the cap with unrelated channels; chan-shared is oldest. + for (let i = 0; i < CAP; i++) { + await ingestOne(AGENT_A, `chan-other-${i}`, 100 + i); + } + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-shared").length, + 0, + "agent-A entry for the evicted channel must be gone", + ); + assert.equal( + _testGetArchivedChannelEvents(AGENT_B, "chan-shared").length, + 0, + "agent-B entry for the same evicted channel must be gone too", + ); + }); + + it("test_evicted_channel_reads_back_empty_for_rehydration", async () => { + // The revisit contract: an evicted channel returns an empty window, which + // the paging hook treats as a cache miss and re-hydrates from SQLite. Here + // we assert the empty read; re-ingesting the same page restores it. + for (let i = 0; i <= CAP; i++) { + await ingestOne(AGENT_A, `chan-${i}`, i); + } + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-0").length, + 0, + "evicted channel reads back empty", + ); + // Re-hydrate: re-ingesting restores the window (proves eviction is not a + // permanent tombstone — the key is fully reusable). + await ingestOne(AGENT_A, "chan-0", 0); + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-0").length, + 1, + "re-ingest after eviction restores the channel window", + ); + }); + + it("test_reset_clears_pin_state_so_counts_do_not_leak", async () => { + // A pin outstanding at reset must not survive it — otherwise a phantom pin + // would protect a channel id in the next session. After reset we re-pin + // NOTHING and confirm the same channel id is freely evictable. + await ingestOne(AGENT_A, "chan-0", 0); + pinArchiveChannel("chan-0"); + resetAgentObserverStore(); + _testRegisterKnownAgents(SUB_ID, [AGENT_A, AGENT_B]); + + // If reset leaked the pin, chan-0 would survive this over-cap wave. + for (let i = 0; i <= CAP; i++) { + await ingestOne(AGENT_A, `chan-${i}`, i); + } + assert.equal( + _testGetArchivedChannelEvents(AGENT_A, "chan-0").length, + 0, + "pin state must not survive reset — chan-0 is freely evictable again", + ); + }); +}); diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index 4a8313a1e1..f4ba1509df 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -25,10 +25,19 @@ import { createEmptyTranscriptState, processTranscriptEvent, } from "./ui/agentSessionTranscript"; +import { selectArchiveEvictionKeys } from "./lib/observerArchiveEviction"; const MAX_OBSERVER_EVENTS = 3000; const MAX_PENDING_UNKNOWN_AGENT_FRAMES = 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 +// history for the lifetime of the app. Eviction drops the least-recently-loaded +// unpinned channels whole; a revisit re-hydrates from SQLite (cache miss, not +// data loss). Channels with a mounted panel are pinned and never evicted. +const MAX_RETAINED_ARCHIVE_CHANNELS = 8; + export type ObserverSnapshot = { connectionState: ConnectionState; errorMessage: string | null; @@ -58,6 +67,53 @@ const snapshotByAgent = new Map(); // TranscriptState once over the combined window. const archiveEventsByChannel = new Map(); +// LRU bookkeeping for `archiveEventsByChannel`. `archiveChannelId` maps a +// composite key back to its channelId (the eviction unit) so the pure policy +// can group agent entries by channel. `archiveAccessSeq` records the value of +// `archiveAccessCounter` at each key's most recent read or write, giving a +// total order for least-recently-used selection. +const archiveChannelId = new Map(); +const archiveAccessSeq = new Map(); +let archiveAccessCounter = 0; + +// Channels whose archive must never be evicted because a panel is mounted on +// them. Refcounted so co-mounted panels (e.g. the channel screen and the +// profile panel viewing the same channel) compose: the channel stays pinned +// until the last panel unmounts. A channel with a positive count is pinned. +const pinnedArchiveChannelCounts = new Map(); + +/** + * Pin a channel's archive against eviction while a panel is mounted on it. + * Refcounted so multiple mounted consumers of the same channel compose; the + * channel is protected until every consumer calls `unpinArchiveChannel`. + * No-op for a null/empty channelId (panels open before a channel resolves). + */ +export function pinArchiveChannel(channelId: string | null | undefined): void { + if (!channelId) return; + pinnedArchiveChannelCounts.set( + channelId, + (pinnedArchiveChannelCounts.get(channelId) ?? 0) + 1, + ); +} + +/** + * Release one pin acquired via `pinArchiveChannel`. When the refcount reaches + * zero the channel becomes eligible for LRU eviction again. No-op for a + * null/empty channelId or a channel with no outstanding pins. + */ +export function unpinArchiveChannel( + channelId: string | null | undefined, +): void { + if (!channelId) return; + const current = pinnedArchiveChannelCounts.get(channelId); + if (current === undefined) return; + if (current <= 1) { + pinnedArchiveChannelCounts.delete(channelId); + } else { + pinnedArchiveChannelCounts.set(channelId, current - 1); + } +} + // Per-agent, per-channel latest-live-session-id. // Key: `${normalizePubkey(agentPubkey)}:${channelId}`. // Set when a live relay observer event with a sessionId arrives. @@ -298,9 +354,42 @@ function appendArchivedChannelEvent( // order for consumers that call buildTranscriptState over the window. const sorted = [...current, event].sort(compareObserverEvents); archiveEventsByChannel.set(key, sorted); + // Record LRU bookkeeping: which channel this key belongs to (the eviction + // unit) and its most-recent access, so the policy can pick the least-recently + // loaded unpinned channels when the retained-channel cap is exceeded. + archiveChannelId.set(key, channelId); + archiveAccessSeq.set(key, ++archiveAccessCounter); return true; } +/** + * Evict the least-recently-loaded unpinned channels from the archive window + * when the number of distinct retained channels exceeds + * `MAX_RETAINED_ARCHIVE_CHANNELS`. Pinned channels (a panel is mounted on them) + * are never evicted. Eviction removes every (agent, channel) entry for the + * dropped channels and their LRU bookkeeping; a later revisit re-hydrates the + * channel from SQLite through the normal paging path. Returns true if any key + * was evicted so the caller can decide whether a notify is warranted. + */ +function evictArchiveChannelsIfNeeded(): boolean { + const entries = Array.from(archiveEventsByChannel.keys(), (key) => ({ + key, + channelId: archiveChannelId.get(key) ?? "", + accessSeq: archiveAccessSeq.get(key) ?? 0, + })); + const evictKeys = selectArchiveEvictionKeys( + entries, + new Set(pinnedArchiveChannelCounts.keys()), + MAX_RETAINED_ARCHIVE_CHANNELS, + ); + for (const key of evictKeys) { + archiveEventsByChannel.delete(key); + archiveChannelId.delete(key); + archiveAccessSeq.delete(key); + } + return evictKeys.length > 0; +} + /** * Read the channel-scoped archive raw events for a given (agent, channel) * pair. Returns an empty array when no archive has been loaded yet. @@ -763,7 +852,14 @@ export async function ingestArchivedObserverEvents( // Batch-notify once for the whole page of archive events. appendAgentEvent // already notifies individually for live/no-channelId events above, so we // only need one extra notify here for the archive path. - if (archiveChanged) { + // + // Enforce the retained-channel bound after the page lands: the channel just + // ingested has the highest access seq, so it is safe from its own eviction; + // only older unpinned channels are dropped. Fold the eviction result into the + // notify decision so a page that only triggers eviction still refreshes any + // panel reading an evicted channel. + const evicted = evictArchiveChannelsIfNeeded(); + if (archiveChanged || evicted) { notifyListeners(); } } @@ -809,6 +905,10 @@ export function resetAgentObserverStore() { transcriptByAgent.clear(); snapshotByAgent.clear(); archiveEventsByChannel.clear(); + archiveChannelId.clear(); + archiveAccessSeq.clear(); + archiveAccessCounter = 0; + pinnedArchiveChannelCounts.clear(); knownAgentPubkeys.clear(); knownAgentsBySubscription.clear(); pendingUnknownAgentFrames.length = 0; diff --git a/desktop/src/features/agents/ui/useLoadArchivedObserverEvents.test.mjs b/desktop/src/features/agents/ui/useLoadArchivedObserverEvents.test.mjs index c3ca1f678b..49d9b3720f 100644 --- a/desktop/src/features/agents/ui/useLoadArchivedObserverEvents.test.mjs +++ b/desktop/src/features/agents/ui/useLoadArchivedObserverEvents.test.mjs @@ -934,3 +934,80 @@ describe("useLoadArchivedObserverEvents — mounted hook lifecycle regressions", await unmount(); }); }); + +describe("useLoadArchivedObserverEvents — post-unmount ingest fence", () => { + beforeEach(() => { + resetAgentObserverStore(); + clearIpcHandlers(); + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + }); + + /** + * Post-unmount fence: a page whose Tauri read is still in flight when the + * panel unmounts must NOT write to the archive store. By unmount time the + * panel's pin has been released, so the channel may already be evicted; + * writing the late page would resurrect a dead channel as most-recently-used + * and silently defeat the archive bound. + * + * Sequence: + * 1. Mount on chan-a. Backfill returns immediately (no rows). The archive + * read is DEFERRED so it is still in flight at unmount. + * 2. Unmount the hook while the read is deferred (disposedRef → true). + * 3. Release the archive read. The resolved page reaches the fence check. + * 4. Assert chan-a's archive window is EMPTY — the fence skipped the write. + * + * Without the disposedRef fence, the late page ingests and the channel window + * is non-empty after unmount. + */ + it("test_page_resolving_after_unmount_does_not_write_to_store", async () => { + let resolveArchive; + const archiveDeferred = new Promise((resolve) => { + resolveArchive = resolve; + }); + + setIpcHandler("list_save_subscriptions", async () => + makeOwnerPSubResponse(), + ); + setIpcHandler("read_unindexed_observer_rows", async () => []); + setIpcHandler("index_observer_channel_id", async () => null); + setIpcHandler("read_archived_observer_events_for_channel", async (args) => { + if (args.channelId === "chan-a") { + await archiveDeferred; // hold the read in flight across unmount + return [JSON.stringify(makeArchivedRow(1, "chan-a"))]; + } + return []; + }); + setIpcHandler("decrypt_observer_event", async (args) => { + try { + const event = JSON.parse(args.eventJson); + return JSON.parse(event.content); + } catch { + return { kind: "telemetry", channelId: null }; + } + }); + + const qc = makeQueryClient(); + const { render, unmount } = mountHook("chan-a", qc); + + // Step 1: mount; hydration starts and blocks on the deferred archive read. + await render("chan-a"); + await act(async () => { + await new Promise((r) => setTimeout(r, 20)); + }); + + // Step 2: unmount while the read is still in flight. + await unmount(); + + // Step 3: release the archive read; let the resolved page reach the fence. + resolveArchive(); + await settle(10); + + // Step 4: the late page must NOT have been written to the store. + const archived = _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-a"); + assert.equal( + archived.length, + 0, + `page resolving after unmount must not write to the store — found ${archived.length} (disposedRef fence missing would leave this 1)`, + ); + }); +}); diff --git a/desktop/src/features/agents/ui/useObserverEvents.ts b/desktop/src/features/agents/ui/useObserverEvents.ts index 0c44a64d5f..873013897c 100644 --- a/desktop/src/features/agents/ui/useObserverEvents.ts +++ b/desktop/src/features/agents/ui/useObserverEvents.ts @@ -6,7 +6,9 @@ import { getAgentTranscript, getArchivedChannelEvents, ingestArchivedObserverEvents, + pinArchiveChannel, subscribeAgentObserverStore, + unpinArchiveChannel, } from "@/features/agents/observerRelayStore"; import { listSaveSubscriptions, @@ -81,6 +83,18 @@ export function useArchivedChannelEvents( agentPubkey: string | null | undefined, channelId: string | null | undefined, ): ObserverEvent[] { + // Pin this channel's archive against LRU eviction while the panel is mounted, + // so the history the user is looking at is never dropped out from under the + // view. Refcounted in the store, so co-mounted consumers of the same channel + // (this read hook plus the loader below, plus a second panel) compose. The + // pin is keyed on channel only — agent identity does not affect eviction, + // which is a per-channel operation. + React.useEffect(() => { + if (!channelId) return; + pinArchiveChannel(channelId); + return () => unpinArchiveChannel(channelId); + }, [channelId]); + const getSnapshot = React.useCallback( () => getArchivedChannelEvents(agentPubkey, channelId), [agentPubkey, channelId], @@ -133,6 +147,19 @@ export function useLoadArchivedObserverEvents( } const ps = pagingStateRef.current; + // Fence for in-flight fetches that resolve after unmount. Set true by the + // cleanup effect below; checked before every store write so a page still + // decrypting when the panel closes cannot repopulate a channel whose pin has + // just been released — which would otherwise defeat the archive bound by + // resurrecting a dead channel as most-recently-used. + const disposedRef = React.useRef(false); + React.useEffect(() => { + disposedRef.current = false; + return () => { + disposedRef.current = true; + }; + }, []); + // React state mirrors the fields callers observe so re-renders fire on change. const [hasSubscription, setHasSubscription] = React.useState( ps.hasSubscription, @@ -335,6 +362,15 @@ export function useLoadArchivedObserverEvents( createdAt: oldestEvent.created_at, id: oldestEvent.id, }; + // Fence post-unmount ingest: if the panel unmounted while this page + // was in flight, its pin has been released and the channel may already + // be evicted. Writing now would resurrect a dead channel as + // most-recently-used, silently defeating the archive bound. Skip the + // store write; the cursor advance above is harmless (the ref is + // discarded on unmount). + if (disposedRef.current) { + return; + } await ingestArchivedObserverEvents(events); }