jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 20:41:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(desktop): preserve archived observer history (#1752)

Browse Source 
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Will Pfleger
2026-07-10 23:46:42 -04:00
committed by GitHub
parent 09be502159
commit 51234bd89f
9 changed files with 873 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

571
desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs
View File

@@ -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,
"older archived event must sort before newer live event",
"live snapshot must have 1 event (no-channelId frame)",
);
assert.equal(snap.events[1].seq, 2);
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,
"archive window must have 1 event (older frame)",
);
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);
assert.equal(
snap.events.length,
2,
"idle agent (enabled=false) must still read archived events from store",
// Channel-A archive window must have at least one item.
const channelAEvents = _testGetArchivedChannelEvents(
AGENT_PUBKEY,
"channel-A",
);
// scopeByChannel on channel-A must return only the channel-A frame.
const { scopeByChannel } = await import(
"@/features/agents/ui/agentSessionPanelLayout.ts"
);
const scopedA = scopeByChannel(snap.events, "channel-A");
assert.equal(
scopedA.length,
channelAEvents.length,
1,
"scopeByChannel(channel-A) must include only channel-A frames",
"idle agent: channel-A 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");
// Channel-B archive window must have at least one item.
const channelBEvents = _testGetArchivedChannelEvents(
AGENT_PUBKEY,
"channel-B",
);
assert.equal(
channelBFrames.length,
0,
"channel-B frames must NOT appear in channel-A scoped view",
channelBEvents.length,
1,
"idle agent: channel-B archive window must contain 1 event",
);
// 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(
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));
});
});

117
desktop/src/features/agents/observerRelayStore.ts
View File

@@ -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<string, ObserverEvent[]>();
const transcriptByAgent = new Map<string, TranscriptState>();
const snapshotByAgent = new Map<string, ObserverSnapshot>();
// 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<string, ObserverEvent[]>();
// 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<unknown> = decryptObserverEvent,
): Promise<void> {
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)) ?? []
);
}

21
desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx
View File

@@ -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 (
<div

45
desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx
View File

@@ -27,12 +27,17 @@ import type {
import type { AgentSessionTranscriptVariant } from "./agentSessionTranscriptContext";
import {
deriveLatestSessionId,
mergeObserverEventWindows,
resolveDisplayEvents,
resolveRawRailLayout,
scopeByChannel,
} from "./agentSessionPanelLayout";
import { shorten } from "./agentSessionUtils";
import { useObserverEvents, useAgentTranscript } from "./useObserverEvents";
import {
useObserverEvents,
useArchivedChannelEvents,
} from "./useObserverEvents";
import { buildTranscriptState } from "./agentSessionTranscript";
type ManagedAgentSessionPanelProps = {
agent: Pick<ManagedAgent, "pubkey" | "name" | "status"> & {
@@ -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(

41
desktop/src/features/agents/ui/agentSessionPanelLayout.ts
View File

@@ -12,6 +12,47 @@ export function scopeByChannel<T extends { channelId?: string | null }>(
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.

78
desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs
View File

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

21
desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts
View File

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

29
desktop/src/features/agents/ui/useObserverEvents.ts
View File

@@ -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;
/**

48
desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx
View File

@@ -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<HTMLDivElement>(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;
}