diff --git a/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs index 4a1978a28..eb38da9a8 100644 --- a/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs +++ b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs @@ -17,6 +17,7 @@ import { getAgentObserverSnapshot, resetAgentObserverStore, _testRegisterKnownAgents, + _testGetArchivedChannelEvents, } from "@/features/agents/observerRelayStore.ts"; // ── Constants ───────────────────────────────────────────────────────────────── @@ -145,71 +146,86 @@ describe("ingestArchivedObserverEvents", () => { _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); const obs = makeObserverEvent({ seq: 1 }); await ingestArchivedObserverEvents([makeRawEvent()], makeDecrypt(obs)); + // Archived events with a channelId are stored in the channel-scoped archive + // window (not in the per-agent live snapshot). Read via raw events for tests. + const archivedEvents = _testGetArchivedChannelEvents( + AGENT_PUBKEY, + "chan-1", + ); + assert.equal(archivedEvents.length, 1, "archive must contain 1 raw event"); + assert.equal(archivedEvents[0].seq, 1); + // Also verify the live snapshot is untouched — archive separation. const snap = getAgentObserverSnapshot(AGENT_PUBKEY, true); - assert.equal(snap.events.length, 1); - assert.equal(snap.events[0].seq, 1); + assert.equal( + snap.events.length, + 0, + "live snapshot must be empty for archived events", + ); }); it("test_dedup_does_not_add_live_present_event", async () => { - // Pre-seed a live event via E2E injection. - const liveObs = makeObserverEvent({ - seq: 5, - timestamp: "2026-01-01T00:00:05.000Z", - }); - injectObserverEventsForE2E(AGENT_PUBKEY, [liveObs]); - _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); - // Try to ingest an archived event with the SAME (seq, timestamp) — must be deduped. + // Ingest the same archived event twice — the channel archive window must dedup + // so (seq, timestamp) pairs are only stored once. const archivedObs = makeObserverEvent({ seq: 5, timestamp: "2026-01-01T00:00:05.000Z", }); - await ingestArchivedObserverEvents( - [makeRawEvent()], - makeDecrypt(archivedObs), + await ingestArchivedObserverEvents([makeRawEvent(), makeRawEvent()], () => + Promise.resolve(archivedObs), ); - const snap = getAgentObserverSnapshot(AGENT_PUBKEY, true); + const archiveEvents = _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-1"); assert.equal( - snap.events.length, + archiveEvents.length, 1, - "dedup: duplicate seq+timestamp must not add a second entry", + "dedup: identical (seq, timestamp) must produce exactly 1 entry in the archive window", ); }); it("test_older_archived_event_sorts_before_live", async () => { - // Pre-seed a newer live event. + // Pre-seed a newer live event (no channelId → goes to live path). const liveObs = makeObserverEvent({ seq: 2, timestamp: "2026-01-01T00:00:02.000Z", + channelId: null, }); injectObserverEventsForE2E(AGENT_PUBKEY, [liveObs]); _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); - // Ingest an older archived event. + // Ingest an older archived event (has channelId → goes to archive window). const archivedObs = makeObserverEvent({ seq: 1, timestamp: "2026-01-01T00:00:01.000Z", + channelId: "chan-1", }); await ingestArchivedObserverEvents( [makeRawEvent()], makeDecrypt(archivedObs), ); + // Live snapshot has seq=2; archive window for chan-1 has seq=1. + // After merge they should appear in ascending time order. const snap = getAgentObserverSnapshot(AGENT_PUBKEY, true); - assert.equal(snap.events.length, 2); - // Ascending time order: older first. assert.equal( - snap.events[0].seq, + snap.events.length, + 1, + "live snapshot must have 1 event (no-channelId frame)", + ); + assert.equal(snap.events[0].seq, 2, "live event must be seq=2"); + + const archiveEvents = _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-1"); + assert.equal( + archiveEvents.length, 1, - "older archived event must sort before newer live event", + "archive window must have 1 event (older frame)", ); - assert.equal(snap.events[1].seq, 2); + assert.equal(archiveEvents[0].seq, 1, "older archived event must be seq=1"); }); it("test_multiple_events_ingested_in_order", async () => { _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); - // Three events: seq 3, 1, 2 — must end up sorted 1, 2, 3. + // Three events with channelId — all go to archive window, not live snapshot. const events = [ makeObserverEvent({ seq: 3, timestamp: "2026-01-01T00:00:03.000Z" }), makeObserverEvent({ seq: 1, timestamp: "2026-01-01T00:00:01.000Z" }), @@ -222,11 +238,25 @@ describe("ingestArchivedObserverEvents", () => { [makeRawEvent(), makeRawEvent(), makeRawEvent()], decryptFn, ); - const snap = getAgentObserverSnapshot(AGENT_PUBKEY, true); - assert.equal(snap.events.length, 3); + // All have channelId "chan-1" — verify archive window, not live snapshot. + const archiveEvents = _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-1"); + assert.equal( + archiveEvents.length, + 3, + "archive must have 3 ingested events", + ); + // Events must be sorted ascending by timestamp (compareObserverEvents order). assert.deepEqual( - snap.events.map((e) => e.seq), + archiveEvents.map((e) => e.seq), [1, 2, 3], + "archive events must be sorted ascending by timestamp", + ); + // Live snapshot must be empty (all events were channeled). + const snap = getAgentObserverSnapshot(AGENT_PUBKEY, true); + assert.equal( + snap.events.length, + 0, + "live snapshot must be empty for channeled archived events", ); }); @@ -234,6 +264,10 @@ describe("ingestArchivedObserverEvents", () => { // archived rows in the store must render those rows, scoped to the viewed // channel. Prior to the fix, getAgentObserverSnapshot returned IDLE_SNAPSHOT // when enabled=false, discarding ingested archived events. + // + // Updated: archived events now go to the channel-scoped archive window + // (getArchivedChannelTranscript), not the live snapshot. The channel-scoping + // is by construction — cross-channel contamination is impossible. it("test_idle_agent_archived_events_readable_when_enabled_false", async () => { _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); // Ingest two archived events: one for channel-A, one for channel-B. @@ -257,34 +291,39 @@ describe("ingestArchivedObserverEvents", () => { () => Promise.resolve(events[callIdx++]), ); - // With enabled=false (simulating isManagedAgentActive=false for idle agent): - // getAgentObserverSnapshot must still return stored events. - const snap = getAgentObserverSnapshot(AGENT_PUBKEY, false); + // Channel-A archive window must have at least one item. + const channelAEvents = _testGetArchivedChannelEvents( + AGENT_PUBKEY, + "channel-A", + ); assert.equal( - snap.events.length, - 2, - "idle agent (enabled=false) must still read archived events from store", + channelAEvents.length, + 1, + "idle agent: channel-A archive window must contain 1 event", ); - // scopeByChannel on channel-A must return only the channel-A frame. - const { scopeByChannel } = await import( - "@/features/agents/ui/agentSessionPanelLayout.ts" + // Channel-B archive window must have at least one item. + const channelBEvents = _testGetArchivedChannelEvents( + AGENT_PUBKEY, + "channel-B", ); - const scopedA = scopeByChannel(snap.events, "channel-A"); assert.equal( - scopedA.length, + channelBEvents.length, 1, - "scopeByChannel(channel-A) must include only channel-A frames", + "idle agent: channel-B archive window must contain 1 event", ); - assert.equal(scopedA[0].channelId, "channel-A"); - // scopeByChannel on channel-A must exclude channel-B frames — the core - // cross-channel-contamination guard. - const channelBFrames = scopedA.filter((e) => e.channelId === "channel-B"); + // Cross-channel contamination guard: channel-A events must not appear in channel-B window. + const channelASeqSet = new Set( + channelAEvents.map((e) => `${e.seq}:${e.timestamp}`), + ); + const contaminated = channelBEvents.some((e) => + channelASeqSet.has(`${e.seq}:${e.timestamp}`), + ); assert.equal( - channelBFrames.length, - 0, - "channel-B frames must NOT appear in channel-A scoped view", + contaminated, + false, + "channel-B archive must NOT contain channel-A events (cross-channel contamination guard)", ); }); }); @@ -445,3 +484,437 @@ describe("archive paging state reset on channel change", () => { ); }); }); + +// ── Archive window beyond MAX_OBSERVER_EVENTS cap (regression) ──────────────── +// +// The live observer relay store caps per-agent events at MAX_OBSERVER_EVENTS +// (3,000). Before this fix, archived events flowed through the same capped path, +// so loading more than 3,000 archived frames silently discarded the oldest ones. +// The channel-scoped archive window is uncapped — all loaded history persists. +// +// This describe block injects 3,100 archived events and verifies every one +// is preserved in the archive window without truncation. + +describe("archive window holds more than MAX_OBSERVER_EVENTS (3000) frames", () => { + beforeEach(() => { + resetAgentObserverStore(); + }); + + it("test_archive_window_retains_all_events_beyond_3000_cap", async () => { + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + + const OVER_CAP = 3100; + const rawEvents = Array.from({ length: OVER_CAP }, (_, i) => ({ + id: `e${String(i).padStart(63, "0")}`, + pubkey: AGENT_PUBKEY, + created_at: 1000 + i, + kind: 24200, + tags: [ + ["p", OTHER_PUBKEY], + ["agent", AGENT_PUBKEY], + ["frame", "telemetry"], + ], + content: "encrypted", + sig: "s".repeat(128), + })); + const observerEvents = Array.from({ length: OVER_CAP }, (_, i) => + makeObserverEvent({ + seq: i + 1, + timestamp: new Date(1000000 + i * 1000).toISOString(), + channelId: "chan-1", + }), + ); + let callIdx = 0; + await ingestArchivedObserverEvents(rawEvents, () => + Promise.resolve(observerEvents[callIdx++]), + ); + + const archiveEvents = _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-1"); + assert.equal( + archiveEvents.length, + OVER_CAP, + `archive window must hold all ${OVER_CAP} events without truncation (cap was 3000)`, + ); + + // The live snapshot must be empty — separation is strict. + const snap = getAgentObserverSnapshot(AGENT_PUBKEY, true); + assert.equal( + snap.events.length, + 0, + "live snapshot must not contain any archived events", + ); + + // All 3100 events must be sorted ascending. + assert.equal( + archiveEvents[0].seq, + 1, + "first archived event must be seq=1 (oldest)", + ); + assert.equal( + archiveEvents[OVER_CAP - 1].seq, + OVER_CAP, + `last archived event must be seq=${OVER_CAP} (newest)`, + ); + }); + + it("test_resetAgentObserverStore_clears_archive_window", async () => { + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + + const obs = makeObserverEvent({ seq: 1, channelId: "chan-1" }); + await ingestArchivedObserverEvents([makeRawEvent()], () => + Promise.resolve(obs), + ); + + // Confirm events are present before reset. + assert.equal( + _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-1").length, + 1, + "pre-reset: archive must have 1 event", + ); + + // Reset must wipe the archive window. + resetAgentObserverStore(); + + assert.equal( + _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-1").length, + 0, + "post-reset: archive window must be empty after resetAgentObserverStore", + ); + }); +}); + +// ── Session-boundary key stability across prepend (regression) ──────────────── +// +// getDisplayBlockKey for session-boundary blocks previously used `runIndex` (the +// run's array-position), which shifts when older sessions are prepended before +// existing runs — causing React to remount unchanged boundaries and churn the +// virtual list. The fix uses `firstItemId` (the id of the first item in the +// following run), which is invariant across prepend. +// +// The full key-stability test suite lives in agentSessionTranscriptGrouping.test.mjs +// where proper TranscriptItem fixtures are already defined. See +// buildTranscriptDisplayBlocks_sessionBoundary_emitsFirstItemId and +// buildTranscriptDisplayBlocks_sessionBoundary_keyStableAcrossPrepend there. + +// ── Raw-event-level merge: subscription notification + stateful aggregates ─────── +// +// These tests verify the revised design where: +// - A full archive page notifies useSyncExternalStore subscribers. +// - Tool start (archive) + tool_call_update (live) yields one complete row. +// - Permission request (archive) + response (live) yields the resolved outcome. +// - The combined raw event window feeds raw rail / header count. +// - The >3,000-frame retention guarantee still holds. + +import { + subscribeAgentObserverStore, + getArchivedChannelEvents, +} from "@/features/agents/observerRelayStore.ts"; +import { buildTranscriptState } from "@/features/agents/ui/agentSessionTranscript.ts"; +import { mergeObserverEventWindows } from "@/features/agents/ui/agentSessionPanelLayout.ts"; + +describe("archive page subscription notification", () => { + beforeEach(() => { + resetAgentObserverStore(); + }); + + it("test_full_archive_page_notifies_subscribers", async () => { + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + + let notifyCount = 0; + const unsubscribe = subscribeAgentObserverStore(() => { + notifyCount++; + }); + + // Ingest a full 50-row page of archived events. + const PAGE_SIZE = 50; + const rawEvents = Array.from({ length: PAGE_SIZE }, (_, i) => + makeRawEvent({ id: `p${String(i).padStart(63, "0")}` }), + ); + const observerEvents = Array.from({ length: PAGE_SIZE }, (_, i) => + makeObserverEvent({ + seq: i + 1, + timestamp: new Date(1_000_000 + i * 1000).toISOString(), + channelId: "chan-1", + }), + ); + let callIdx = 0; + await ingestArchivedObserverEvents(rawEvents, () => + Promise.resolve(observerEvents[callIdx++]), + ); + + unsubscribe(); + + // Must have fired exactly one batched notification for the full page. + assert.equal( + notifyCount, + 1, + `expected exactly 1 batched subscriber notification after full archive page, got ${notifyCount}`, + ); + + // Production getter must reflect the added events. + const archiveEvents = getArchivedChannelEvents(AGENT_PUBKEY, "chan-1"); + assert.equal( + archiveEvents.length, + PAGE_SIZE, + "production getter must return all 50 archive events", + ); + }); + + it("test_all_duplicate_page_does_not_notify", async () => { + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + + // Seed one event. + const obs = makeObserverEvent({ seq: 1, channelId: "chan-1" }); + await ingestArchivedObserverEvents([makeRawEvent()], () => + Promise.resolve(obs), + ); + + // Subscribe AFTER the first ingest so we only track the second ingest. + let notifyCount = 0; + const unsubscribe = subscribeAgentObserverStore(() => { + notifyCount++; + }); + + // Re-ingest the same event — all duplicates, no new state. + await ingestArchivedObserverEvents([makeRawEvent()], () => + Promise.resolve(obs), + ); + + unsubscribe(); + + assert.equal( + notifyCount, + 0, + "a page of pure duplicates must not notify subscribers", + ); + }); +}); + +describe("raw-event-level merge: stateful aggregates across live/archive boundary", () => { + beforeEach(() => { + resetAgentObserverStore(); + }); + + it("test_tool_start_in_archive_plus_update_in_live_yields_complete_row", () => { + // Simulates the boundary scenario: tool_call frame is older than the + // live cap and lives only in the archive; tool_call_update (completion) is + // in the live window. Merging at the raw-event level and running a single + // buildTranscriptState must yield one complete tool row with the update's + // status, not two separate rows. + const TOOL_ID = "tool-abc-123"; + const CHANNEL = "chan-merge"; + + // Archive: tool_call (start, executing) + const toolStart = makeObserverEvent({ + seq: 1, + timestamp: "2026-01-01T00:00:01.000Z", + kind: "acp_read", + channelId: CHANNEL, + sessionId: "sess-1", + turnId: "turn-1", + payload: { + method: "session/update", + params: { + update: { + sessionUpdate: "tool_call", + toolCallId: TOOL_ID, + toolName: "read_file", + status: "executing", + arguments: { path: "/tmp/x" }, + }, + }, + }, + }); + + // Live: tool_call_update (completion) + const toolUpdate = makeObserverEvent({ + seq: 2, + timestamp: "2026-01-01T00:00:02.000Z", + kind: "acp_read", + channelId: CHANNEL, + sessionId: "sess-1", + turnId: "turn-1", + payload: { + method: "session/update", + params: { + update: { + sessionUpdate: "tool_call_update", + toolCallId: TOOL_ID, + toolName: "read_file", + status: "completed", + result: "file content", + }, + }, + }, + }); + + const archiveEvents = [toolStart]; + const liveEvents = [toolUpdate]; + + const combined = mergeObserverEventWindows(liveEvents, archiveEvents); + assert.equal(combined.length, 2, "combined must have 2 raw events"); + + const state = buildTranscriptState(combined); + const toolRows = state.items.filter( + (item) => + item.type === "tool" && item.id === `tool:${CHANNEL}:${TOOL_ID}`, + ); + assert.equal(toolRows.length, 1, "must produce exactly 1 tool row"); + assert.equal( + toolRows[0].status, + "completed", + "tool row must carry the completed status from the update frame", + ); + }); + + it("test_permission_request_in_archive_plus_response_in_live_yields_resolved_outcome", () => { + // Simulates: permission request is older than the live cap (archive only); + // permission response arrived live. Single-pass buildTranscriptState must + // resolve the outcome and update the row. + const CHANNEL = "chan-perm"; + const RPC_ID = "rpc-perm-42"; + + // Archive: permission request — option uses production optionId + kind fields + // so the optionId→kind map is populated and the outcome label is correct. + const permRequest = makeObserverEvent({ + seq: 10, + timestamp: "2026-01-01T00:10:00.000Z", + kind: "acp_read", + channelId: CHANNEL, + sessionId: "sess-1", + turnId: "turn-2", + payload: { + id: RPC_ID, + method: "session/request_permission", + params: { + description: "Read /etc/passwd", + options: [ + { optionId: "allow_once", kind: "allow_once", name: "Allow once" }, + ], + }, + }, + }); + + // Live: permission response — result.outcome carries outcome:"selected" and + // the selected optionId so describePermissionOutcome builds "Approved (allow_once)". + const permResponse = makeObserverEvent({ + seq: 11, + timestamp: "2026-01-01T00:10:01.000Z", + kind: "acp_write", + channelId: CHANNEL, + sessionId: "sess-1", + turnId: "turn-2", + payload: { + id: RPC_ID, + result: { + outcome: { + outcome: "selected", + optionId: "allow_once", + }, + }, + }, + }); + + const combined = mergeObserverEventWindows([permResponse], [permRequest]); + assert.equal(combined.length, 2, "combined must have 2 raw events"); + + const state = buildTranscriptState(combined); + const permRows = state.items.filter( + (item) => item.type === "lifecycle" && item.renderClass === "permission", + ); + assert.equal(permRows.length, 1, "must produce exactly 1 permission row"); + // The row must carry the fully-resolved production label. + assert.equal( + permRows[0].outcome, + "Approved (allow_once)", + "permission row outcome must be the production-shaped label when request+response are in the combined window", + ); + }); + + it("test_raw_rail_count_includes_archived_events_after_reload", async () => { + // Simulates a reload-shaped scenario: only archived events are ingested + // (no live events delivered yet). The combined raw window used for the + // raw rail and header count must include the archived rows. + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + const CHANNEL = "chan-raw"; + + const archiveObsEvents = Array.from({ length: 5 }, (_, i) => + makeObserverEvent({ + seq: i + 1, + timestamp: new Date(1_000_000 + i * 1000).toISOString(), + channelId: CHANNEL, + }), + ); + let callIdx = 0; + await ingestArchivedObserverEvents( + Array.from({ length: 5 }, (_, i) => + makeRawEvent({ id: `r${String(i).padStart(63, "0")}` }), + ), + () => Promise.resolve(archiveObsEvents[callIdx++]), + ); + + // Production getter returns archived raw events. + const archived = getArchivedChannelEvents(AGENT_PUBKEY, CHANNEL); + // Live events for this agent are empty (no live relay ingest). + const liveSnap = getAgentObserverSnapshot(AGENT_PUBKEY, true); + const liveScoped = liveSnap.events.filter((e) => e.channelId === CHANNEL); + + const combined = mergeObserverEventWindows(liveScoped, archived); + + assert.equal( + combined.length, + 5, + "combined window must contain all 5 archived events for raw rail/count", + ); + assert.equal( + liveScoped.length, + 0, + "live scoped events must be empty after archive-only ingest", + ); + }); + + it("test_header_last_updated_is_non_null_after_archive_only_ingest", async () => { + // Regression for AgentSessionThreadPanel header: after a reload where only + // archived events exist (no live relay yet), the merged raw window must + // contain events with valid timestamps so the header shows "Last updated …" + // rather than "No updates yet". + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + const CHANNEL = "chan-header"; + const TIMESTAMP = "2026-06-01T12:00:00.000Z"; + + await ingestArchivedObserverEvents( + [makeRawEvent({ id: "h".repeat(64) })], + () => + Promise.resolve( + makeObserverEvent({ + seq: 1, + timestamp: TIMESTAMP, + channelId: CHANNEL, + }), + ), + ); + + // Simulate what AgentSessionThreadPanel computes for the header timestamp: + // merged combined events (live=[], archive=[1 event]) → max timestamp. + const archived = getArchivedChannelEvents(AGENT_PUBKEY, CHANNEL); + const combined = mergeObserverEventWindows([], archived); + + assert.equal( + combined.length, + 1, + "combined must contain the archived event", + ); + + // latestActivityAt equivalent: max of finite Date.parse values. + const latestMs = combined.reduce((acc, e) => { + const parsed = Date.parse(e.timestamp); + return Number.isFinite(parsed) ? Math.max(acc ?? -Infinity, parsed) : acc; + }, /** @type {number | null} */ null); + + assert.ok( + latestMs !== null && Number.isFinite(latestMs), + "header latest timestamp must be non-null after archive-only ingest (header must not say 'No updates yet')", + ); + assert.equal(latestMs, Date.parse(TIMESTAMP)); + }); +}); diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index 775ad00b0..cba397abf 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -35,6 +35,7 @@ const IDLE_SNAPSHOT: ObserverSnapshot = { events: [], }; +const EMPTY_EVENTS: ObserverEvent[] = []; const EMPTY_TRANSCRIPT: TranscriptItem[] = []; const listeners = new Set<() => void>(); @@ -42,6 +43,15 @@ const eventsByAgent = new Map(); const transcriptByAgent = new Map(); const snapshotByAgent = new Map(); +// 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 +// `eventsByAgent` (per-agent, capped) and this map is NEVER written by live +// events — separation is strict so loading deep history can never evict live frames +// or vice versa. UI consumers merge the raw events from both sources, then derive +// TranscriptState once over the combined window. +const archiveEventsByChannel = new Map(); + // Per-agent, per-channel latest-live-session-id. // Key: `${normalizePubkey(agentPubkey)}:${channelId}`. // Set when a live relay observer event with a sessionId arrives. @@ -204,6 +214,75 @@ function appendAgentEvent(agentPubkey: string, event: ObserverEvent) { notifyListeners(); } +/** + * Compose the map key for the channel-scoped archive transcript. + * Separates agent identity from channel with `:` — the same delimiter used by + * liveSessionKey so all composite keys in this module are consistently shaped. + */ +function archiveChannelKey(agentPubkey: string, channelId: string): string { + return `${normalizePubkey(agentPubkey)}:${channelId}`; +} + +/** + * Append a decoded archived observer event to the channel-scoped archive + * event journal. Unlike `appendAgentEvent`, this path does NOT cap or trim — + * the channel archive window grows only by explicit paged loads from SQLite, + * so unbounded growth from live relay events is impossible. + * + * Deduplicates on `(seq, timestamp)` — identical to `appendAgentEvent` — so + * events that arrive on the live relay before the archive page is loaded are + * silently skipped. The archive window and the live transcript are kept + * strictly separate: live events never write here. + * + * Returns `true` if the event was added (state changed), `false` if it was a + * duplicate and was skipped. The caller batches notifications. + */ +function appendArchivedChannelEvent( + agentPubkey: string, + channelId: string, + event: ObserverEvent, +): boolean { + const key = archiveChannelKey(agentPubkey, channelId); + const current = archiveEventsByChannel.get(key) ?? []; + + // Dedup: skip if (seq, timestamp) already present in the archive window. + if ( + current.some( + (existing) => + existing.seq === event.seq && existing.timestamp === event.timestamp, + ) + ) { + return false; + } + + // Archive pages arrive newest-first from SQLite, so each new event sorts + // BEFORE the existing entries. Sort the combined array to maintain ascending + // order for consumers that call buildTranscriptState over the window. + const sorted = [...current, event].sort(compareObserverEvents); + archiveEventsByChannel.set(key, sorted); + return true; +} + +/** + * Read the channel-scoped archive raw events for a given (agent, channel) + * pair. Returns an empty array when no archive has been loaded yet. + * + * Called by `useArchivedChannelEvents` so UI components can reactively + * subscribe to archive loads and derive transcript state from the combined + * live + archive raw event window without touching the live-capped per-agent + * store. + */ +export function getArchivedChannelEvents( + agentPubkey: string | null | undefined, + channelId: string | null | undefined, +): ObserverEvent[] { + if (!agentPubkey || !channelId) return EMPTY_EVENTS; + return ( + archiveEventsByChannel.get(archiveChannelKey(agentPubkey, channelId)) ?? + EMPTY_EVENTS + ); +} + export function compareObserverEvents( left: ObserverEvent, right: ObserverEvent, @@ -535,6 +614,7 @@ export async function ingestArchivedObserverEvents( rawEvents: RelayEvent[], _decryptFn: (event: RelayEvent) => Promise = decryptObserverEvent, ): Promise { + let archiveChanged = false; for (const event of rawEvents) { const agentPubkey = observerTag(event, "agent"); const frame = observerTag(event, "frame"); @@ -549,11 +629,31 @@ export async function ingestArchivedObserverEvents( } try { const parsed = (await _decryptFn(event)) as ObserverEvent; - appendAgentEvent(agentPubkey, parsed); + // Route archived events to the channel-scoped archive window (no cap) + // rather than the per-agent live-relay store (MAX_OBSERVER_EVENTS cap). + // Events without a channelId fall through to the live store so they + // remain visible in the agent's general transcript. + if (parsed.channelId) { + const added = appendArchivedChannelEvent( + agentPubkey, + parsed.channelId, + parsed, + ); + if (added) archiveChanged = true; + } else { + // Live path already calls notifyListeners() inside appendAgentEvent. + appendAgentEvent(agentPubkey, parsed); + } } catch { // Silently drop decrypt failures — same as live path error handling. } } + // 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) { + notifyListeners(); + } } /** @@ -597,6 +697,7 @@ export function resetAgentObserverStore() { eventsByAgent.clear(); transcriptByAgent.clear(); snapshotByAgent.clear(); + archiveEventsByChannel.clear(); knownAgentPubkeys.clear(); knownAgentsBySubscription.clear(); latestLiveSessionByAgentChannel.clear(); @@ -618,3 +719,17 @@ export function _testRegisterKnownAgents( ): void { registerKnownAgents(subscriptionId, pubkeys); } + +/** + * Test-only: read the raw archived observer events for a (agent, channel) pair. + * Production callers should use `getArchivedChannelEvents`. + * Only call from tests — never from production code. + */ +export function _testGetArchivedChannelEvents( + agentPubkey: string, + channelId: string, +): ObserverEvent[] { + return ( + archiveEventsByChannel.get(archiveChannelKey(agentPubkey, channelId)) ?? [] + ); +} diff --git a/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx b/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx index f7914db55..0c0717aae 100644 --- a/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx +++ b/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx @@ -338,7 +338,9 @@ function getDisplayBlockKey(block: TranscriptDisplayBlock) { return block.item.id; } if (block.kind === "session-boundary") { - return `session-boundary:${block.sessionId}:${block.runIndex}`; + // Use firstItemId (stable across prepend) rather than runIndex (shifts when + // older sessions are prepended, causing unnecessary boundary remounts). + return `session-boundary:${block.sessionId}:${block.firstItemId}`; } return `turn:${block.turnId}`; } @@ -933,12 +935,11 @@ function TurnSetupStatus({ /** * Horizontal rule rendered between session runs in the observer transcript. * - * Three label states: - * - `"current"` — live session, agent is actively running in this context. - * - `"most-recent"` — newest visible session but no live match (archived-only - * or session ended). Shown as "Most recent observed session" - * so users know it is history, not active context. - * - `"earlier"` — an older session. Always labeled with the archive signal. + * Three label states (based on live-frame observation, not harness affinity): + * - `"current"` — most recent session observed via the live relay subscription. + * - `"most-recent"` — newest visible session with no matching live frames + * (loaded from archive or session ended before observation). + * - `"earlier"` — an older session preceding the most-recent one. */ function SessionBoundaryDivider({ labelState, @@ -949,10 +950,10 @@ function SessionBoundaryDivider({ }) { const label = labelState === "current" - ? "Current session" + ? "Latest live-observed session" : labelState === "most-recent" - ? "Most recent observed session — not in current context" - : "Earlier session — not in current context"; + ? "Most recent observed session" + : "Earlier observed session"; const formattedDate = new Date(sessionStartTimestamp).toLocaleString(); return (
& { @@ -80,22 +85,42 @@ export function ManagedAgentSessionPanel({ hasObserver, agent.pubkey, ); - const transcript = useAgentTranscript(hasObserver, agent.pubkey); - const scopedTranscript = React.useMemo( - () => scopeByChannel(transcript, channelId), - [channelId, transcript], + // Channel-scoped live events (capped at MAX_OBSERVER_EVENTS) and uncapped + // archived events from SQLite paging. Both are raw ObserverEvent[] — we merge + // them at the raw-event level and derive a single TranscriptState, so stateful + // aggregates (tool start/update, plan replacement, permission request/response) + // are never split across two independent state machines. + const archivedChannelEvents = useArchivedChannelEvents( + agent.pubkey, + channelId, ); - const displayTranscript = transcriptOverride ?? scopedTranscript; - - const scopedEvents = React.useMemo( + const scopedLiveEvents = React.useMemo( () => scopeByChannel(events, channelId), [channelId, events], ); + + // Combined raw window: live (scoped) + archive merged by (seq, timestamp), + // sorted ascending. Used as the single source for both the transcript and the + // raw event rail / header count. + const combinedEvents = React.useMemo( + () => mergeObserverEventWindows(scopedLiveEvents, archivedChannelEvents), + [scopedLiveEvents, archivedChannelEvents], + ); + + // Derive transcript once from the combined raw window. When transcriptOverride + // is set (e.g. E2E snapshot specs), bypass both — the caller supplies the full + // transcript directly. + const derivedTranscript = React.useMemo( + () => buildTranscriptState(combinedEvents).items, + [combinedEvents], + ); + const displayTranscript = transcriptOverride ?? derivedTranscript; + const displayEvents = React.useMemo( - () => resolveDisplayEvents(scopedEvents, rawEventsOverride), - [rawEventsOverride, scopedEvents], + () => resolveDisplayEvents(combinedEvents, rawEventsOverride), + [rawEventsOverride, combinedEvents], ); const latestSessionId = React.useMemo( diff --git a/desktop/src/features/agents/ui/agentSessionPanelLayout.ts b/desktop/src/features/agents/ui/agentSessionPanelLayout.ts index 34b6f32cf..52896dd50 100644 --- a/desktop/src/features/agents/ui/agentSessionPanelLayout.ts +++ b/desktop/src/features/agents/ui/agentSessionPanelLayout.ts @@ -12,6 +12,47 @@ export function scopeByChannel( return items.filter((item) => item.channelId === channelId); } +/** + * Merge live and archived raw `ObserverEvent[]` arrays into a single + * deduplicated, chronologically-sorted array. + * + * The live event window is capped at MAX_OBSERVER_EVENTS (3000) and holds the + * most recent events for the agent/channel. The archive window is channel-scoped + * paged history loaded from SQLite — it extends the visible range beyond the cap. + * + * Deduplication: events present in both (e.g. a frame that arrived live and was + * also loaded from the archive) are collapsed to one entry by `(seq, timestamp)`. + * The live copy is preferred when a duplicate exists, since the live path may + * have applied incremental transcript mutations via `processTranscriptEvent`. + * + * Sorting: ascending `compareObserverEvents` order (timestamp then seq). + * Callers should pass the result directly to `buildTranscriptState()`. + */ +export function mergeObserverEventWindows( + liveEvents: readonly ObserverEvent[], + archivedEvents: readonly ObserverEvent[], +): ObserverEvent[] { + if (archivedEvents.length === 0) return liveEvents as ObserverEvent[]; + if (liveEvents.length === 0) return archivedEvents as ObserverEvent[]; + + // Dedup key: same as appendAgentEvent / appendArchivedChannelEvent. + const liveKeySet = new Set(liveEvents.map((e) => `${e.seq}:${e.timestamp}`)); + const uniqueArchived = archivedEvents.filter( + (e) => !liveKeySet.has(`${e.seq}:${e.timestamp}`), + ); + if (uniqueArchived.length === 0) return liveEvents as ObserverEvent[]; + + const merged = [...liveEvents, ...uniqueArchived]; + // compareObserverEvents: timestamp diff then seq diff (ascending). + merged.sort((a, b) => { + const ta = Date.parse(a.timestamp); + const tb = Date.parse(b.timestamp); + if (Number.isFinite(ta) && Number.isFinite(tb) && ta !== tb) return ta - tb; + return a.seq - b.seq; + }); + return merged; +} + /** * Derive the most recent session id from a list of observer events by * scanning from the end. Returns null when no event carries a sessionId. diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs index 09e2ad286..8da3e52a1 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs +++ b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs @@ -1328,3 +1328,81 @@ test("splitIntoSessionRuns: system-prompt renders before turn blocks when no use `toolA (${toolAIdx}) must be before boundary (${boundaryIdx})`, ); }); + +// ── Session-boundary firstItemId key stability (regression) ─────────────────── +// +// Previously getDisplayBlockKey keyed boundaries as +// `session-boundary:${sessionId}:${runIndex}`. `runIndex` is the run's +// position in the ordered array — it SHIFTS when older sessions are prepended +// (archive page load), causing React to remount unchanged boundaries and churn +// the virtual list. +// +// The fix replaces `runIndex` with `firstItemId` — the id of the first +// TranscriptItem in the run — which is invariant across prepend. + +test("buildTranscriptDisplayBlocks_sessionBoundary_emitsFirstItemId", () => { + // Two sessions: sess-1 (older) then sess-2 (newer). + const items = [ + sessionItem("a", "sess-1", "2026-07-08T00:00:01.000Z"), + sessionItem("b", "sess-2", "2026-07-08T00:00:02.000Z"), + sessionItem("c", "sess-2", "2026-07-08T00:00:03.000Z"), + ]; + const blocks = buildTranscriptDisplayBlocks(items, null); + const boundary = blocks.find((b) => b.kind === "session-boundary"); + assert.ok(boundary, "boundary must be present between two sessions"); + assert.equal( + boundary.sessionId, + "sess-2", + "boundary labels the newer session", + ); + // firstItemId must equal the id of the first item in sess-2's run. + assert.ok(boundary.firstItemId, "boundary must carry firstItemId"); + // "b" is the first item in sess-2's run (from sessionItem("b", "sess-2")). + assert.equal( + boundary.firstItemId, + "b", + "firstItemId must equal the id of the first item in the following session run", + ); +}); + +test("buildTranscriptDisplayBlocks_sessionBoundary_keyStableAcrossPrepend", () => { + // Before: sess-1 then sess-2. + const before = [ + sessionItem("a", "sess-1", "2026-07-08T00:00:02.000Z"), + sessionItem("b", "sess-2", "2026-07-08T00:00:03.000Z"), + ]; + const blocksBefore = buildTranscriptDisplayBlocks(before, null); + const boundaryBefore = blocksBefore.find( + (b) => b.kind === "session-boundary" && b.sessionId === "sess-2", + ); + assert.ok(boundaryBefore, "must have sess-2 boundary before prepend"); + const keyBefore = `session-boundary:${boundaryBefore.sessionId}:${boundaryBefore.firstItemId}`; + + // After: prepend an older sess-0 before sess-1. Now [sess-0, sess-1, sess-2]. + const after = [ + sessionItem("z", "sess-0", "2026-07-08T00:00:01.000Z"), // oldest — prepended + sessionItem("a", "sess-1", "2026-07-08T00:00:02.000Z"), + sessionItem("b", "sess-2", "2026-07-08T00:00:03.000Z"), + ]; + const blocksAfter = buildTranscriptDisplayBlocks(after, null); + const boundaryAfter = blocksAfter.find( + (b) => b.kind === "session-boundary" && b.sessionId === "sess-2", + ); + assert.ok(boundaryAfter, "must still have sess-2 boundary after prepend"); + const keyAfter = `session-boundary:${boundaryAfter.sessionId}:${boundaryAfter.firstItemId}`; + + // firstItemId-based key must be identical before and after prepend. + assert.equal( + keyBefore, + keyAfter, + "session-boundary React key must not change when an older session is prepended", + ); + + // Sanity check: runIndex DID shift (from 1 to 2) confirming the old key would have changed. + assert.equal(boundaryBefore.runIndex, 1, "runIndex before prepend must be 1"); + assert.equal( + boundaryAfter.runIndex, + 2, + "runIndex after prepend must be 2, confirming it is unstable", + ); +}); diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts index e1960faae..967e08adc 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts +++ b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts @@ -39,8 +39,23 @@ export type TranscriptDisplayBlock = * of the run it precedes, which is always ≥ 1). Used as a tiebreaker in * the React list key so that two non-contiguous runs sharing the same * sessionId produce DISTINCT keys and React never duplicates/omits them. + * + * @deprecated Use `firstItemId` for stable React keys instead. `runIndex` + * shifts when older sessions are prepended before existing runs, causing + * key churn and unnecessary remounting of unchanged boundary nodes. + * Kept for callers that use position for non-key purposes. */ runIndex: number; + /** + * The `id` of the first `TranscriptItem` in the run that follows this + * boundary. Stable across prepend: older runs inserted before this run + * do not change the first item of this run, so this field never churns + * on archive-page loads. + * + * Use this field (not `runIndex`) as the React list key component for + * session-boundary rows. + */ + firstItemId: string; }; export type TranscriptToolRunChildSegment = @@ -548,12 +563,18 @@ export function buildTranscriptDisplayBlocks( : isNewestRun ? "most-recent" : "earlier"; + // firstItemId: the id of the first TranscriptItem in this run. Stable + // across prepend — inserting older runs before this run does not change + // its first item. Falls back to sessionId when the run has no items + // (edge case for empty runs from pre-session null-id buffers). + const firstItemId = run.items[0]?.id ?? run.sessionId; allBlocks.push({ kind: "session-boundary", sessionId: run.sessionId, sessionStartTimestamp, labelState, runIndex: i, + firstItemId, }); } diff --git a/desktop/src/features/agents/ui/useObserverEvents.ts b/desktop/src/features/agents/ui/useObserverEvents.ts index a04eefbd0..8f4ecc103 100644 --- a/desktop/src/features/agents/ui/useObserverEvents.ts +++ b/desktop/src/features/agents/ui/useObserverEvents.ts @@ -4,6 +4,7 @@ import { ensureRelayObserverSubscription, getAgentObserverSnapshot, getAgentTranscript, + getArchivedChannelEvents, ingestArchivedObserverEvents, subscribeAgentObserverStore, } from "@/features/agents/observerRelayStore"; @@ -15,7 +16,7 @@ import { } from "@/shared/api/tauriArchive"; import { decryptObserverEvent } from "@/shared/api/tauriObserver"; import { useIdentityQuery } from "@/shared/api/hooks"; -import type { TranscriptItem } from "./agentSessionTypes"; +import type { ObserverEvent, TranscriptItem } from "./agentSessionTypes"; import type { RelayEvent } from "@/shared/api/types"; import { createArchivePagingState, @@ -61,6 +62,32 @@ export function useAgentTranscript( return React.useSyncExternalStore(subscribeToStore, getSnapshot); } +/** + * Reactively read the channel-scoped archive raw events for a given + * (agent, channel) pair. Returns an empty array until archive pages are loaded. + * + * Subscribes to `subscribeAgentObserverStore` so it re-renders whenever + * `ingestArchivedObserverEvents` writes new pages to the archive window — the + * same subscription used by the live event snapshot, keeping both in sync. + * + * UI consumers merge these events with the live event window and call + * `buildTranscriptState()` once over the combined sorted/deduplicated set, + * so stateful transcript relationships (tool start/update, plan replacement, + * permission request/response) are never split across two independent state + * machines. + */ +export function useArchivedChannelEvents( + agentPubkey: string | null | undefined, + channelId: string | null | undefined, +): ObserverEvent[] { + const getSnapshot = React.useCallback( + () => getArchivedChannelEvents(agentPubkey, channelId), + [agentPubkey, channelId], + ); + + return React.useSyncExternalStore(subscribeToStore, getSnapshot); +} + const ARCHIVED_EVENTS_PAGE_SIZE = 50; /** diff --git a/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx b/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx index 0992c3b17..1d3576014 100644 --- a/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx +++ b/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx @@ -10,14 +10,14 @@ import { toast } from "sonner"; import { useAgentWorking } from "@/features/agents/agentWorkingSignal"; import { isManagedAgentActive } from "@/features/agents/lib/managedAgentControlActions"; -import { scopeByChannel } from "@/features/agents/ui/agentSessionPanelLayout"; -import type { - ObserverEvent, - TranscriptItem, -} from "@/features/agents/ui/agentSessionTypes"; +import { + mergeObserverEventWindows, + scopeByChannel, +} from "@/features/agents/ui/agentSessionPanelLayout"; +import type { ObserverEvent } from "@/features/agents/ui/agentSessionTypes"; import { ManagedAgentSessionPanel } from "@/features/agents/ui/ManagedAgentSessionPanel"; import { - useAgentTranscript, + useArchivedChannelEvents, useObserverEvents, } from "@/features/agents/ui/useObserverEvents"; import { cancelManagedAgentTurn } from "@/shared/api/agentControl"; @@ -107,22 +107,24 @@ export function AgentSessionThreadPanel({ const topSentinelRef = React.useRef(null); const now = useNow(1000); const { events } = useObserverEvents(isLive, agent.pubkey); - const transcript = useAgentTranscript(isLive, agent.pubkey); const scopedEvents = React.useMemo( () => scopeByChannel(events, sessionChannelId), [events, sessionChannelId], ); - const scopedTranscript = React.useMemo( - () => scopeByChannel(transcript, sessionChannelId), - [sessionChannelId, transcript], + // Archived channel events merged with live scoped events so the header's + // "Last updated" timestamp reflects the full loaded history, not just the + // capped live window. Mirrors ManagedAgentSessionPanel's combinedEvents. + const archivedChannelEvents = useArchivedChannelEvents( + agent.pubkey, + sessionChannelId, + ); + const combinedHeaderEvents = React.useMemo( + () => mergeObserverEventWindows(scopedEvents, archivedChannelEvents), + [scopedEvents, archivedChannelEvents], ); const latestActivityAt = React.useMemo( - () => - getLatestActivityTimestamp({ - events: scopedEvents, - transcript: scopedTranscript, - }), - [scopedEvents, scopedTranscript], + () => getLatestActivityTimestamp(combinedHeaderEvents), + [combinedHeaderEvents], ); const lastUpdatedLabel = formatLastUpdatedLabel(latestActivityAt, now); const lastUpdatedTitle = @@ -425,13 +427,9 @@ export function AgentSessionThreadPanel({ ); } -function getLatestActivityTimestamp({ - events, - transcript, -}: { - events: readonly ObserverEvent[]; - transcript: readonly TranscriptItem[]; -}): number | null { +function getLatestActivityTimestamp( + events: readonly ObserverEvent[], +): number | null { let latest: number | null = null; const record = (timestamp: string) => { @@ -449,10 +447,6 @@ function getLatestActivityTimestamp({ record(event.timestamp); } - for (const item of transcript) { - record(item.timestamp); - } - return latest; }