From b50a4cba3d151e23b8c000640b36fa208c0bd6d4 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 8 Jul 2026 10:15:59 -0400 Subject: [PATCH 01/10] =?UTF-8?q?fix(desktop):=20scope=20observer=20feed?= =?UTF-8?q?=20by=20channel=20and=20add=20session-boundary=20dividers=20Two?= =?UTF-8?q?=20observer-feed=20defects=20fixed=20in=20a=20single=20PR:=20Sl?= =?UTF-8?q?ice=201=20=E2=80=94=20channel-scoped=20archive=20(index=20at=20?= =?UTF-8?q?ingest=20+=20idempotent=20backfill)=20The=20archive=20read=20pa?= =?UTF-8?q?th=20queried=20all=20owner=5Fp=20kind=2024200=20rows=20for=20an?= =?UTF-8?q?=20identity,=20then=20relied=20on=20a=20display-time=20scopeByC?= =?UTF-8?q?hannel=20filter.=20When=20the=20filter=20was=20missing=20or=20s?= =?UTF-8?q?kipped,=20frames=20from=20other=20channels=20leaked=20into=20th?= =?UTF-8?q?e=20current=20channel's=20observer=20feed.=20Fix:=20add=20an=20?= =?UTF-8?q?observer=5Fchannel=5Findex=20table=20that=20maps=20(identity,?= =?UTF-8?q?=20relay,=20event=5Fid)=20to=20channel=5Fid.=20Three=20new=20Ta?= =?UTF-8?q?uri=20commands=20drive=20this:=20-=20read=5Farchived=5Fobserver?= =?UTF-8?q?=5Fevents=5Ffor=5Fchannel:=20channel-scoped=20paginated=20read?= =?UTF-8?q?=20via=20=20=20the=20index=20(replaces=20the=20raw=20owner=5Fp?= =?UTF-8?q?=20scan=20in=20useLoadArchivedObserverEvents)=20-=20index=5Fobs?= =?UTF-8?q?erver=5Fchannel=5Fid:=20TS=20calls=20this=20after=20decrypting?= =?UTF-8?q?=20events=20to=20write=20=20=20index=20entries=20-=20read=5Funi?= =?UTF-8?q?ndexed=5Fobserver=5Frows:=20returns=20all=20owner=5Fp=20kind=20?= =?UTF-8?q?24200=20rows=20not=20yet=20=20=20indexed,=20for=20the=20one-sho?= =?UTF-8?q?t=20idempotent=20backfill=20Backfill=20strategy=20(choice=20i,?= =?UTF-8?q?=20backfill-before-read):=20a=20React.useEffect=20in=20useLoadA?= =?UTF-8?q?rchivedObserverEvents=20runs=20once=20on=20mount,=20reads=20all?= =?UTF-8?q?=20unindexed=20rows,=20decrypts=20each,=20and=20batch-writes=20?= =?UTF-8?q?index=20entries.=20After=20backfill=20the=20channel=20index=20i?= =?UTF-8?q?s=20authoritative=20=E2=80=94=20the=20scoped=20read=20pages=20i?= =?UTF-8?q?t=20directly=20with=20no=20zero-match=20raw-page=20guard=20need?= =?UTF-8?q?ed.=20Null/decrypt-failed=20channelId=20rows=20stay=20unscoped?= =?UTF-8?q?=20and=20are=20hidden=20from=20every=20scoped=20channel=20view?= =?UTF-8?q?=20(Will's=20ruling:=20option=20(a)=20=E2=80=94=20display=20dec?= =?UTF-8?q?ision=20only,=20rows=20stay=20on=20disk).=20Slice=202=20?= =?UTF-8?q?=E2=80=94=20session-boundary=20rendering=20Items=20rendered=20f?= =?UTF-8?q?lat=20with=20no=20session=20grouping;=20archived=20and=20live?= =?UTF-8?q?=20frames=20interleaved=20without=20any=20visual=20marker=20of?= =?UTF-8?q?=20where=20one=20agent=20session=20ended=20and=20another=20bega?= =?UTF-8?q?n,=20causing=20users=20to=20think=20archived=20historical=20con?= =?UTF-8?q?text=20is=20still=20live.=20Fix:=20split=20TranscriptItem[]=20i?= =?UTF-8?q?nto=20contiguous=20session=20runs=20before=20calling=20buildTra?= =?UTF-8?q?nscriptDisplayBlocks.=20This=20also=20fixes=20the=20pendingSyst?= =?UTF-8?q?emPrompt=20single-slot=20clobber=20(each=20run=20gets=20its=20o?= =?UTF-8?q?wn=20lifecycle).=20A=20new=20session-boundary=20display=20block?= =?UTF-8?q?=20kind=20is=20interleaved=20between=20runs;=20SessionBoundaryD?= =?UTF-8?q?ivider=20renders=20a=20horizontal=20rule=20with=20session=20sta?= =?UTF-8?q?rt=20time.=20No=20boundary=20emitted=20for=20single-session=20t?= =?UTF-8?q?ranscripts.=20Liveness=20=E2=80=94=20"current=20session"=20gati?= =?UTF-8?q?ng=20Track=20latest-live-session-id=20per=20(normalized=20agent?= =?UTF-8?q?=20pubkey,=20channelId)=20in=20observerRelayStore,=20set=20on?= =?UTF-8?q?=20the=20handleRelayObserverEvent=20path,=20cleared=20in=20rese?= =?UTF-8?q?tAgentObserverStore.=20Label=20a=20boundary=20"current=20sessio?= =?UTF-8?q?n"=20only=20when=20it=20is=20both=20the=20newest-visible=20sess?= =?UTF-8?q?ion=20AND=20equals=20that=20latest-live=20id;=20otherwise=20lab?= =?UTF-8?q?el=20it=20"most=20recent=20observed=20session".?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- desktop/src-tauri/src/archive/mod.rs | 114 +++++++++ desktop/src-tauri/src/archive/pipeline.rs | 20 ++ desktop/src-tauri/src/archive/store.rs | 205 ++++++++++++++++ desktop/src-tauri/src/archive/store_tests.rs | 122 ++++++++++ desktop/src-tauri/src/lib.rs | 3 + .../src/features/agents/observerRelayStore.ts | 75 ++++++ .../agents/ui/AgentSessionTranscriptList.tsx | 84 ++++++- .../agentSessionTranscriptGrouping.test.mjs | 221 ++++++++++++++++++ .../ui/agentSessionTranscriptGrouping.ts | 130 ++++++++++- .../features/agents/ui/useObserverEvents.ts | 149 +++++++++++- .../channels/ui/AgentSessionThreadPanel.tsx | 2 +- desktop/src/shared/api/tauriArchive.ts | 86 +++++++ 12 files changed, 1194 insertions(+), 17 deletions(-) diff --git a/desktop/src-tauri/src/archive/mod.rs b/desktop/src-tauri/src/archive/mod.rs index c5fd77003..49cc6cff1 100644 --- a/desktop/src-tauri/src/archive/mod.rs +++ b/desktop/src-tauri/src/archive/mod.rs @@ -475,6 +475,120 @@ pub fn delete_save_subscription( // ── read_archived_events ───────────────────────────────────────────────────── +// ── read_archived_observer_events_for_channel ──────────────────────────────── + +/// Read a paginated page of archived kind 24200 events scoped to one channel, +/// using the `observer_channel_index` as the primary lookup. +/// +/// The index only contains rows with a known, non-null `channelId` (frames +/// pre-dating the channelId stamp, or rows where decryption failed, are +/// absent from the index and thus absent from every scoped channel view — +/// Will's (a) ruling, 2026-07-08). +/// +/// Returns at most `limit` events (default `DEFAULT_READ_LIMIT`) in +/// newest-first order. Compound cursor `(before_created_at, before_id)` works +/// identically to `read_archived_events`. +#[tauri::command] +pub fn read_archived_observer_events_for_channel( + state: State<'_, AppState>, + channel_id: String, + before_created_at: Option, + before_id: Option, + limit: Option, +) -> Result, String> { + let identity_pk = identity_pubkey(&state)?; + let relay_url = relay_ws_url_with_override(&state); + let conn = open_db()?; + store::read_archived_observer_events_for_channel( + &conn, + &identity_pk, + &relay_url, + &channel_id, + before_created_at, + before_id.as_deref(), + limit.unwrap_or(DEFAULT_READ_LIMIT), + ) +} + +// ── index_observer_channel_id ───────────────────────────────────────────────── + +/// Index one or more archived observer frame ids with their decoded channelId. +/// +/// Called from the TS-side backfill after attempting `decryptObserverEvent`. +/// `channel_id` is `Some` for frames with a non-null channelId; `None` for +/// frames where decryption yielded no channelId or failed entirely. Both cases +/// write a status row so a re-run skips the frame (INSERT OR IGNORE on PK). +/// +/// Idempotent: rows that are already indexed are left unchanged. +#[tauri::command] +pub fn index_observer_channel_id( + state: State<'_, AppState>, + entries: Vec, +) -> Result<(), String> { + let identity_pk = identity_pubkey(&state)?; + let relay_url = relay_ws_url_with_override(&state); + let conn = open_db()?; + for entry in &entries { + store::upsert_observer_channel_index( + &conn, + &identity_pk, + &relay_url, + &entry.event_id, + entry.channel_id.as_deref(), + entry.created_at, + )?; + } + Ok(()) +} + +/// A single (event_id, channel_id?, created_at) record used by +/// `index_observer_channel_id`. +/// +/// `channel_id` is `None` for frames where decryption found no channelId or +/// failed; those rows are written to `observer_channel_index` with a NULL +/// channel_id so the frame is treated as processed (no re-decrypt on re-run). +#[derive(Debug, Deserialize)] +pub struct ObserverChannelIndexEntry { + pub event_id: String, + pub channel_id: Option, + pub created_at: i64, +} + +// ── read_unindexed_observer_rows ───────────────────────────────────────────── + +/// Return raw event JSON + id + created_at for all `owner_p` kind 24200 rows +/// that are NOT yet in `observer_channel_index`. +/// +/// The TS-side backfill driver calls this once, decrypts each row, and sends +/// the (id, channelId, created_at) triples back via `index_observer_channel_id`. +/// Together these constitute the one-shot idempotent backfill required by the +/// Slice 1 acceptance criteria (Thufir Pass 4). +#[tauri::command] +pub fn read_unindexed_observer_rows( + state: State<'_, AppState>, +) -> Result, String> { + let identity_pk = identity_pubkey(&state)?; + let relay_url = relay_ws_url_with_override(&state); + let conn = open_db()?; + let rows = store::read_unindexed_observer_rows(&conn, &identity_pk, &relay_url)?; + Ok(rows + .into_iter() + .map(|(id, raw_json, created_at)| RawObserverRow { + id, + raw_json, + created_at, + }) + .collect()) +} + +/// Wire type returned by `read_unindexed_observer_rows`. +#[derive(Debug, Serialize)] +pub struct RawObserverRow { + pub id: String, + pub raw_json: String, + pub created_at: i64, +} + /// Default page size for `read_archived_events`. const DEFAULT_READ_LIMIT: i64 = 50; diff --git a/desktop/src-tauri/src/archive/pipeline.rs b/desktop/src-tauri/src/archive/pipeline.rs index ec3de94cc..2bd149dce 100644 --- a/desktop/src-tauri/src/archive/pipeline.rs +++ b/desktop/src-tauri/src/archive/pipeline.rs @@ -423,6 +423,26 @@ pub(super) fn commit_archive( &p.matched_scope.scope_value, now, )?; + + // Index at ingest: attempt to decrypt and extract channelId. + // Write a status row regardless of outcome so backfill never + // re-processes this frame (INSERT OR IGNORE on PK is a no-op if + // the row is already present from a prior run). + let channel_id_for_index: Option = + buzz_core_pkg::observer::decrypt_observer_payload::( + owner_keys, &p.event, + ) + .ok() + .and_then(|v| v.get("channelId")?.as_str().map(|s| s.to_owned())); + store::upsert_observer_channel_index( + &tx, + identity_pk, + relay_url, + eid, + channel_id_for_index.as_deref(), + p.event.created_at.as_secs() as i64, + )?; + persisted += 1; } diff --git a/desktop/src-tauri/src/archive/store.rs b/desktop/src-tauri/src/archive/store.rs index 083f9dc43..24d21c0b1 100644 --- a/desktop/src-tauri/src/archive/store.rs +++ b/desktop/src-tauri/src/archive/store.rs @@ -46,8 +46,62 @@ CREATE TABLE IF NOT EXISTS save_subscriptions ( created_at INTEGER NOT NULL, PRIMARY KEY (identity_pubkey, relay_url, scope_type, scope_value) ); + +-- Processing-status index for kind 24200 (observer) frames. +-- +-- Every examined observer row — whether its channelId was successfully +-- decrypted or not — gets exactly one row here keyed by (identity, relay, id). +-- `channel_id` is nullable: +-- NOT NULL → frame was attributed to a channel; visible in that channel's feed. +-- NULL → frame was examined but had no channelId (pre-stamp or decrypt +-- failure); excluded from all scoped views per Will's (a) ruling +-- (2026-07-08), but the row exists so a re-run is a no-op. +-- +-- Scoped reads filter `channel_id = ?`; NULL rows never match, staying hidden. +CREATE TABLE IF NOT EXISTS observer_channel_index ( + identity_pubkey TEXT NOT NULL, + relay_url TEXT NOT NULL, + id TEXT NOT NULL, + channel_id TEXT, + created_at INTEGER NOT NULL, + PRIMARY KEY (identity_pubkey, relay_url, id) +); +CREATE INDEX IF NOT EXISTS idx_observer_channel + ON observer_channel_index (identity_pubkey, relay_url, channel_id, created_at DESC, id DESC); + +-- One-row migration state table: tracks which idempotent migrations have run. +CREATE TABLE IF NOT EXISTS archive_migrations ( + name TEXT PRIMARY KEY, + applied_at INTEGER NOT NULL +); "; +// ── Migration helpers ──────────────────────────────────────────────────────── + +/// Returns true if a named migration has already been applied. +pub fn migration_applied(conn: &Connection, name: &str) -> Result { + // The table is created in SCHEMA above, so it always exists after open. + let count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM archive_migrations WHERE name = ?1", + params![name], + |row| row.get(0), + ) + .map_err(|e| format!("migration_applied query: {e}"))?; + Ok(count > 0) +} + +/// Mark a named migration as applied (idempotent: no-op if already recorded). +pub fn mark_migration_applied(conn: &Connection, name: &str, now: i64) -> Result<(), String> { + conn.execute( + "INSERT INTO archive_migrations (name, applied_at) VALUES (?1, ?2) + ON CONFLICT (name) DO NOTHING", + params![name, now], + ) + .map_err(|e| format!("mark_migration_applied: {e}"))?; + Ok(()) +} + // ── Open / init ───────────────────────────────────────────────────────────── /// Open (or create) the archive database at the given path. @@ -568,6 +622,157 @@ pub fn read_archived_events( .map_err(|e| format!("read read_archived_events row: {e}")) } +/// Upsert a row in `observer_channel_index` for an examined kind 24200 frame. +/// +/// `channel_id` is `Some` for frames with a successfully-decrypted, non-null +/// channelId; `None` for frames that had no channelId or where decryption +/// failed. Null-channel_id rows are excluded from all scoped channel reads +/// (scoped queries filter `channel_id = ?`; NULLs never match), satisfying +/// Will's (a) ruling (2026-07-08), while ensuring the row exists so a re-run +/// is a no-op. +/// +/// Idempotent: if the (identity, relay, id) PK already exists the row is left +/// unchanged (INSERT OR IGNORE). Called both at ingest time (new frames) and +/// from the one-shot backfill for existing archived rows. +pub fn upsert_observer_channel_index( + conn: &Connection, + identity_pubkey: &str, + relay_url: &str, + event_id: &str, + channel_id: Option<&str>, + created_at: i64, +) -> Result<(), String> { + conn.execute( + "INSERT OR IGNORE INTO observer_channel_index + (identity_pubkey, relay_url, id, channel_id, created_at) + VALUES (?1, ?2, ?3, ?4, ?5)", + params![identity_pubkey, relay_url, event_id, channel_id, created_at], + ) + .map_err(|e| format!("failed to upsert observer_channel_index: {e}"))?; + Ok(()) +} + +/// Read all `owner_p` kind 24200 archived event rows (id + raw_json + +/// created_at) that have NOT yet been processed (i.e., have no row in +/// `observer_channel_index`, regardless of what channel_id would be). +/// +/// Used by the one-shot backfill migration to discover which existing rows +/// still need channel attribution attempted. A row is "processed" as soon as +/// we attempt decryption — whether it succeeds (non-null channel_id written) +/// or fails (null channel_id written) — so re-runs skip those rows. +pub fn read_unindexed_observer_rows( + conn: &Connection, + identity_pubkey: &str, + relay_url: &str, +) -> Result, String> { + let mut stmt = conn + .prepare( + "SELECT ae.id, ae.raw_json, ae.created_at + FROM archived_events ae + INNER JOIN archived_event_scopes aes + ON aes.identity_pubkey = ae.identity_pubkey + AND aes.relay_url = ae.relay_url + AND aes.id = ae.id + WHERE ae.identity_pubkey = ?1 + AND ae.relay_url = ?2 + AND ae.kind = 24200 + AND aes.scope_type = 'owner_p' + AND ae.id NOT IN ( + SELECT id FROM observer_channel_index + WHERE identity_pubkey = ?1 + AND relay_url = ?2 + ) + ORDER BY ae.created_at DESC, ae.id DESC", + ) + .map_err(|e| format!("prepare read_unindexed_observer_rows: {e}"))?; + + let rows = stmt + .query_map(params![identity_pubkey, relay_url], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, i64>(2)?, + )) + }) + .map_err(|e| format!("query read_unindexed_observer_rows: {e}"))?; + + rows.collect::, _>>() + .map_err(|e| format!("read read_unindexed_observer_rows row: {e}")) +} + +/// Read a paginated page of archived kind 24200 events scoped to one channel, +/// using the `observer_channel_index` as the primary lookup. +/// +/// Returns `raw_json` strings in newest-first order. The compound cursor +/// `(before_created_at, before_id)` works identically to `read_archived_events`. +/// A page shorter than `limit` signals the archive is exhausted for this channel. +#[allow(clippy::too_many_arguments)] +pub fn read_archived_observer_events_for_channel( + conn: &Connection, + identity_pubkey: &str, + relay_url: &str, + channel_id: &str, + before_created_at: Option, + before_id: Option<&str>, + limit: i64, +) -> Result, String> { + let mut next_slot: usize = 4; + let mut extra_clauses = String::new(); + let mut before_at_val: Option = None; + let mut before_id_val: Option = None; + + if let (Some(bat), Some(bid)) = (before_created_at, before_id) { + before_at_val = Some(bat); + before_id_val = Some(bid.to_owned()); + extra_clauses.push_str(&format!( + " AND (oci.created_at < ?{next_slot} \ + OR (oci.created_at = ?{next_slot} AND oci.id < ?{}))", + next_slot + 1, + )); + next_slot += 2; + } + let limit_slot = next_slot; + + let sql = format!( + "SELECT ae.raw_json \ + FROM observer_channel_index oci \ + INNER JOIN archived_events ae \ + ON ae.identity_pubkey = oci.identity_pubkey \ + AND ae.relay_url = oci.relay_url \ + AND ae.id = oci.id \ + WHERE oci.identity_pubkey = ?1 \ + AND oci.relay_url = ?2 \ + AND oci.channel_id = ?3\ + {extra_clauses}\ + ORDER BY oci.created_at DESC, oci.id DESC \ + LIMIT ?{limit_slot}", + ); + + let mut params_vec: Vec> = vec![ + Box::new(identity_pubkey.to_owned()), + Box::new(relay_url.to_owned()), + Box::new(channel_id.to_owned()), + ]; + if let (Some(bat), Some(bid)) = (before_at_val, before_id_val) { + params_vec.push(Box::new(bat)); + params_vec.push(Box::new(bid)); + } + params_vec.push(Box::new(limit)); + + let param_refs: Vec<&dyn rusqlite::ToSql> = params_vec.iter().map(|p| p.as_ref()).collect(); + + let mut stmt = conn + .prepare(&sql) + .map_err(|e| format!("prepare read_archived_observer_events_for_channel: {e}"))?; + + let rows = stmt + .query_map(param_refs.as_slice(), |row| row.get::<_, String>(0)) + .map_err(|e| format!("query read_archived_observer_events_for_channel: {e}"))?; + + rows.collect::, _>>() + .map_err(|e| format!("read read_archived_observer_events_for_channel row: {e}")) +} + /// GC: delete orphaned event rows whose last scope row was just removed. /// /// Called after any batch deletion of scope rows. Uses a LEFT JOIN so only diff --git a/desktop/src-tauri/src/archive/store_tests.rs b/desktop/src-tauri/src/archive/store_tests.rs index 18a80af60..0a07811e7 100644 --- a/desktop/src-tauri/src/archive/store_tests.rs +++ b/desktop/src-tauri/src/archive/store_tests.rs @@ -728,3 +728,125 @@ fn test_read_archived_events_same_second_cursor_no_skip() { // No overlap with page1. assert!(page2.iter().all(|r| !r.contains("\"z\""))); } + +// ── observer_channel_index ─────────────────────────────────────────────────── + +#[test] +fn test_upsert_observer_channel_index_non_null_is_idempotent() { + let conn = in_memory(); + let pk = "owner"; + let relay = "wss://r"; + + upsert_observer_channel_index(&conn, pk, relay, "ev1", Some("ch-abc"), 1000).unwrap(); + // Second call with same PK → INSERT OR IGNORE, no error, row unchanged. + upsert_observer_channel_index(&conn, pk, relay, "ev1", Some("ch-abc"), 2000).unwrap(); + + // Only one row should exist and created_at stays at 1000 (first write wins). + let count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM observer_channel_index WHERE id = 'ev1'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(count, 1, "INSERT OR IGNORE must not duplicate rows"); + + let at: i64 = conn + .query_row( + "SELECT created_at FROM observer_channel_index WHERE id = 'ev1'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(at, 1000, "first write's created_at must be preserved"); +} + +#[test] +fn test_upsert_observer_channel_index_null_channel_id_is_idempotent() { + let conn = in_memory(); + let pk = "owner"; + let relay = "wss://r"; + + // Write a NULL channel_id status row (unscoped / decrypt-failed frame). + upsert_observer_channel_index(&conn, pk, relay, "ev-null", None, 500).unwrap(); + // Re-run → no-op. + upsert_observer_channel_index(&conn, pk, relay, "ev-null", None, 600).unwrap(); + + let count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM observer_channel_index WHERE id = 'ev-null'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!( + count, 1, + "null-channel_id row must not be duplicated on re-run" + ); + + let channel_id: Option = conn + .query_row( + "SELECT channel_id FROM observer_channel_index WHERE id = 'ev-null'", + [], + |r| r.get(0), + ) + .unwrap(); + assert!( + channel_id.is_none(), + "channel_id must be NULL for unscoped frames" + ); +} + +#[test] +fn test_read_archived_observer_excludes_null_channel_id_rows() { + // Scoped reads filter channel_id = '?'; NULL rows must never appear. + let conn = in_memory(); + let pk = "owner"; + let relay = "wss://r"; + + // Seed an archived event row. + upsert_archived_event(&conn, pk, relay, "ev-null", 24200, "agent", 1000, "{}", 0).unwrap(); + upsert_event_scope(&conn, pk, relay, "ev-null", "owner_p", pk, 0).unwrap(); + + // Index it with a NULL channel_id. + upsert_observer_channel_index(&conn, pk, relay, "ev-null", None, 1000).unwrap(); + + // Scoped read for ANY channel must return nothing (NULL != 'some-channel'). + let rows = + read_archived_observer_events_for_channel(&conn, pk, relay, "some-channel", None, None, 50) + .unwrap(); + assert!( + rows.is_empty(), + "scoped read must not return null-channel_id rows" + ); +} + +#[test] +fn test_read_unindexed_observer_rows_excludes_processed_rows() { + // After a NULL channel_id row is written, the event must no longer appear + // in read_unindexed_observer_rows (it is "processed"). + let conn = in_memory(); + let pk = "owner"; + let relay = "wss://r"; + + // Seed an archived observer event with owner_p scope. + upsert_archived_event(&conn, pk, relay, "ev-old", 24200, "agent", 1000, "{}", 0).unwrap(); + upsert_event_scope(&conn, pk, relay, "ev-old", "owner_p", pk, 0).unwrap(); + + // Before indexing: ev-old must appear in unindexed rows. + let unindexed = read_unindexed_observer_rows(&conn, pk, relay).unwrap(); + assert!( + unindexed.iter().any(|(id, _, _)| id == "ev-old"), + "ev-old must appear in unindexed rows before indexing" + ); + + // Index it with NULL channel_id (decrypt-failed / unscoped). + upsert_observer_channel_index(&conn, pk, relay, "ev-old", None, 1000).unwrap(); + + // After indexing: ev-old must NOT appear in unindexed rows (re-run is a no-op). + let unindexed2 = read_unindexed_observer_rows(&conn, pk, relay).unwrap(); + assert!( + !unindexed2.iter().any(|(id, _, _)| id == "ev-old"), + "ev-old must be excluded from unindexed rows after null-channel_id indexing" + ); +} diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 341c99d7b..88dfe207d 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -628,6 +628,9 @@ pub fn run() { archive::list_save_subscriptions, archive::delete_save_subscription, archive::read_archived_events, + archive::read_archived_observer_events_for_channel, + archive::index_observer_channel_id, + archive::read_unindexed_observer_rows, is_auto_update_supported, ]) .build(tauri::generate_context!()) diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index adf6ea92d..7fdcb9504 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -41,6 +41,41 @@ const eventsByAgent = new Map(); const transcriptByAgent = new Map(); const snapshotByAgent = 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. +// Cleared in resetAgentObserverStore. +// +// "Latest-live" means: the sessionId that most recently appeared via the +// live relay path (handleRelayObserverEvent). It is NOT derived from +// connectionState or an ever-live Set — an ever-live Set would incorrectly +// mark session A as "current" after session B has started (Thufir Pass 3). +// +// Stored as `{ sessionId, timestamp, seq }` so that late-arriving live frames +// from an older session never regress the latest-live id. We only advance when +// the parsed event sorts strictly AFTER the stored one, using the same +// two-key ordering as `compareObserverEvents`: timestamp first, then seq on a +// tie — so a higher-seq frame at equal timestamp still advances the entry. +type LatestLiveEntry = { sessionId: string; timestamp: string; seq: number }; +const latestLiveSessionByAgentChannel = new Map(); + +function liveSessionKey(agentPubkey: string, channelId: string | null): string { + return `${normalizePubkey(agentPubkey)}:${channelId ?? ""}`; +} + +/** Read the latest-live-session-id for a (agent, channel) pair. */ +export function getLatestLiveSessionId( + agentPubkey: string | null | undefined, + channelId: string | null | undefined, +): string | null { + if (!agentPubkey) return null; + return ( + latestLiveSessionByAgentChannel.get( + liveSessionKey(agentPubkey, channelId ?? null), + )?.sessionId ?? null + ); +} + // Per-agent listeners for `control_result` frames. The ModelPicker subscribes // here to learn the async outcome of a `switch_model` frame (the send is // fire-and-forget; the harness replies out-of-band over the observer relay). @@ -184,6 +219,26 @@ export function compareObserverEvents( return left.seq - right.seq; } +/** + * Returns true if `candidate` sorts strictly after `stored` using the same + * two-key ordering as `compareObserverEvents`: later timestamp wins; equal + * timestamp falls back to higher seq. Extracted so latest-live advancement + * cannot drift from transcript ordering. + */ +export function isObserverEventAfter( + candidate: { timestamp: string; seq: number }, + stored: { timestamp: string; seq: number }, +): boolean { + const candidateTime = Date.parse(candidate.timestamp); + const storedTime = Date.parse(stored.timestamp); + if (Number.isFinite(candidateTime) && Number.isFinite(storedTime)) { + if (candidateTime !== storedTime) { + return candidateTime > storedTime; + } + } + return candidate.seq > stored.seq; +} + async function handleRelayObserverEvent( event: RelayEvent, activeGeneration: number, @@ -211,6 +266,25 @@ async function handleRelayObserverEvent( if (activeGeneration !== generation) { return; } + // Track the latest-live-session-id per (agent, channel) on the live path. + // Only set when the parsed event carries both a sessionId and channelId, + // so we never attribute a session to the wrong channel. + if (parsed.sessionId && parsed.channelId) { + const key = liveSessionKey(agentPubkey, parsed.channelId); + const stored = latestLiveSessionByAgentChannel.get(key); + // Advance only when this event sorts strictly AFTER the stored one via + // isObserverEventAfter (timestamp then seq — same ordering as + // compareObserverEvents). This prevents late-arriving live frames from + // older sessions from regressing the latest-live id, while also + // correctly advancing on a same-timestamp frame with a higher seq. + if (!stored || isObserverEventAfter(parsed, stored)) { + latestLiveSessionByAgentChannel.set(key, { + sessionId: parsed.sessionId, + timestamp: parsed.timestamp, + seq: parsed.seq, + }); + } + } appendAgentEvent(agentPubkey, parsed); if (parsed.kind === "session_config_captured") { void putAgentSessionConfig(agentPubkey, parsed.payload); @@ -513,6 +587,7 @@ export function resetAgentObserverStore() { snapshotByAgent.clear(); knownAgentPubkeys.clear(); knownAgentsBySubscription.clear(); + latestLiveSessionByAgentChannel.clear(); onSessionConfigCaptured = null; connectionState = "idle"; errorMessage = null; diff --git a/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx b/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx index 00eecd0d0..38143aaac 100644 --- a/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx +++ b/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx @@ -1,11 +1,15 @@ import * as React from "react"; import { motion, useReducedMotion } from "motion/react"; -import { CheckCheck, Radio } from "lucide-react"; +import { CheckCheck, Clock, Radio } from "lucide-react"; import { useActiveAgentTurns, type ActiveTurnSummary, } from "@/features/agents/activeAgentTurnsStore"; +import { + subscribeAgentObserverStore, + getLatestLiveSessionId, +} from "@/features/agents/observerRelayStore"; import type { UserProfileLookup } from "@/features/profile/lib/identity"; import { useAnchoredScroll } from "@/features/messages/ui/useAnchoredScroll"; import { cn } from "@/shared/lib/cn"; @@ -134,9 +138,21 @@ export function AgentSessionTranscriptList({ () => isAgentTurnLive(activeTurns, channelId), [activeTurns, channelId], ); + + // Subscribe to the observer relay store so we read the latest-live-session-id + // reactively. We don't need the full snapshot — only the key for boundary labeling. + const getLatestLive = React.useCallback( + () => getLatestLiveSessionId(agentPubkey, channelId), + [agentPubkey, channelId], + ); + const latestLiveSessionId = React.useSyncExternalStore( + subscribeAgentObserverStore, + getLatestLive, + ); + const displayBlocks = React.useMemo( - () => buildTranscriptDisplayBlocks(items), - [items], + () => buildTranscriptDisplayBlocks(items, latestLiveSessionId), + [items, latestLiveSessionId], ); const scrollContainerRef = React.useRef(null); const contentRef = React.useRef(null); @@ -282,6 +298,11 @@ function hasRenderableCompactBlock(block: TranscriptDisplayBlock) { return isRenderableCompactItem(block.item); } + // session-boundary dividers are not renderable content in compact view. + if (block.kind === "session-boundary") { + return false; + } + return block.segments.some((segment) => { if (segment.kind === "item") { return isRenderableCompactItem(segment.item); @@ -316,6 +337,9 @@ function getDisplayBlockKey(block: TranscriptDisplayBlock) { if (block.kind === "single") { return block.item.id; } + if (block.kind === "session-boundary") { + return `session-boundary:${block.sessionId}`; + } return `turn:${block.turnId}`; } @@ -341,6 +365,15 @@ function TranscriptDisplayBlockView({ const hasCompletedInitialRenderRef = useHasCompletedInitialRender(); const animateSegmentEnter = animationPreferenceEnabled && !shouldReduceMotion; + if (block.kind === "session-boundary") { + return ( + + ); + } + if (block.kind === "single") { return ( +
+ + {labelState === "current" ? ( + +
+
+ ); +} + const TranscriptItemView = React.memo(function TranscriptItemView({ agentAvatarUrl, agentName, diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs index 599cc9244..569895556 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs +++ b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs @@ -6,6 +6,7 @@ import { flattenDisplayBlocks, formatTurnSetupLabel, } from "./agentSessionTranscriptGrouping.ts"; +import { isObserverEventAfter } from "../observerRelayStore.ts"; const baseTimestamp = "2026-06-14T22:20:23.000Z"; @@ -639,3 +640,223 @@ function mkTool(id, label, renderClass = "generic", groupKey = label) { channelId: "chan-1", }; } + +// ── Session-run splitting and session-boundary blocks ────────────────────────── + +/** + * Build a minimal tool-call item stamped with a specific session. + */ +function sessionItem(id, sessionId, ts = "2026-07-08T00:00:00.000Z") { + return { + id, + type: "tool", + renderClass: "generic", + descriptor: { + renderClass: "generic", + label: id, + preview: id, + source: "harness", + groupKey: id, + }, + title: id, + toolName: id, + buzzToolName: null, + status: "completed", + args: {}, + result: "", + isError: false, + timestamp: ts, + startedAt: ts, + completedAt: ts, + turnId: `turn-${id}`, + sessionId, + channelId: "chan-1", + }; +} + +// ── Single session — no boundary injected ────────────────────────────────────── + +test("buildTranscriptDisplayBlocks_singleSession_noBoundaryBlock", () => { + const items = [sessionItem("a", "sess-1"), sessionItem("b", "sess-1")]; + const blocks = buildTranscriptDisplayBlocks(items); + const boundaryBlocks = blocks.filter((b) => b.kind === "session-boundary"); + assert.equal( + boundaryBlocks.length, + 0, + "no session-boundary blocks for a single session", + ); +}); + +// ── Two sessions — one boundary between them ─────────────────────────────────── + +test("buildTranscriptDisplayBlocks_twoSessions_oneBoundaryBetween", () => { + // items ordered oldest-first: sess-1 then sess-2 + const items = [ + sessionItem("a", "sess-1", "2026-07-08T00:00:01.000Z"), + sessionItem("b", "sess-2", "2026-07-08T00:00:02.000Z"), + ]; + const blocks = buildTranscriptDisplayBlocks(items); + const boundaryBlocks = blocks.filter((b) => b.kind === "session-boundary"); + assert.equal( + boundaryBlocks.length, + 1, + "exactly one boundary for two sessions", + ); + // The boundary is inserted BEFORE the newer run (sess-2). + const boundaryIndex = blocks.indexOf(boundaryBlocks[0]); + const prevBlock = blocks[boundaryIndex - 1]; + const nextBlock = blocks[boundaryIndex + 1]; + // Previous content block belongs to sess-1 items, next to sess-2. + const flatPrev = flattenDisplayBlocks([prevBlock]).map((i) => i.id); + const flatNext = flattenDisplayBlocks([nextBlock]).map((i) => i.id); + assert.ok(flatPrev.includes("a"), "content before boundary is sess-1"); + assert.ok(flatNext.includes("b"), "content after boundary is sess-2"); +}); + +// ── Newest session labeled correctly relative to latestLiveSessionId ────────── + +test("buildTranscriptDisplayBlocks_newestMatchesLive_labelStateCurrent", () => { + const items = [ + sessionItem("old", "sess-1", "2026-07-08T00:00:01.000Z"), + sessionItem("new", "sess-2", "2026-07-08T00:00:02.000Z"), + ]; + const blocks = buildTranscriptDisplayBlocks(items, "sess-2"); + const boundary = blocks.find((b) => b.kind === "session-boundary"); + assert.ok(boundary, "boundary present"); + assert.equal( + boundary.labelState, + "current", + "labelState=current when newest session matches live id", + ); + assert.equal(boundary.sessionId, "sess-2", "boundary sessionId is sess-2"); +}); + +test("buildTranscriptDisplayBlocks_newestNoLive_labelStateMostRecent", () => { + const items = [ + sessionItem("old", "sess-1", "2026-07-08T00:00:01.000Z"), + sessionItem("new", "sess-2", "2026-07-08T00:00:02.000Z"), + ]; + // latestLiveSessionId is null → newest session is "most-recent" + const blocksNoLive = buildTranscriptDisplayBlocks(items, null); + const boundaryNoLive = blocksNoLive.find( + (b) => b.kind === "session-boundary", + ); + assert.equal( + boundaryNoLive.labelState, + "most-recent", + "labelState=most-recent when no live id (archived-only view)", + ); + + // latestLiveSessionId is a DIFFERENT session → newest is still "most-recent" + const blocksDiffLive = buildTranscriptDisplayBlocks(items, "sess-other"); + const boundaryDiff = blocksDiffLive.find( + (b) => b.kind === "session-boundary", + ); + assert.equal( + boundaryDiff.labelState, + "most-recent", + "labelState=most-recent when live id differs from newest visible", + ); +}); + +// ── Three sessions — two boundaries ──────────────────────────────────────────── + +test("buildTranscriptDisplayBlocks_threeSessions_twoBoundaries", () => { + 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-3", "2026-07-08T00:00:03.000Z"), + ]; + const blocks = buildTranscriptDisplayBlocks(items); + const boundaryBlocks = blocks.filter((b) => b.kind === "session-boundary"); + assert.equal(boundaryBlocks.length, 2, "two boundaries for three sessions"); + // With no live id: newest (sess-3) boundary = "most-recent"; older = "earlier". + const newestBoundary = boundaryBlocks[boundaryBlocks.length - 1]; + const olderBoundary = boundaryBlocks[0]; + assert.equal( + newestBoundary.labelState, + "most-recent", + "newest boundary is most-recent when latestLiveSessionId is null", + ); + assert.equal( + olderBoundary.labelState, + "earlier", + "older boundary is earlier when latestLiveSessionId is null", + ); +}); + +// ── Null sessionId items stay in the current run ─────────────────────────────── + +test("buildTranscriptDisplayBlocks_nullSessionId_staysInCurrentRun", () => { + // An item with null sessionId should not start a new run. + const items = [ + sessionItem("a", "sess-1", "2026-07-08T00:00:01.000Z"), + // null sessionId — stays in sess-1 run + { ...sessionItem("b", null, "2026-07-08T00:00:02.000Z"), sessionId: null }, + sessionItem("c", "sess-1", "2026-07-08T00:00:03.000Z"), + ]; + const blocks = buildTranscriptDisplayBlocks(items); + const boundaryBlocks = blocks.filter((b) => b.kind === "session-boundary"); + assert.equal( + boundaryBlocks.length, + 0, + "no boundary when null-sessionId items are present within a single session", + ); +}); + +// ── flattenDisplayBlocks skips session-boundary blocks ──────────────────────── + +test("flattenDisplayBlocks_skipsSessionBoundaryBlocks", () => { + const items = [ + sessionItem("a", "sess-1", "2026-07-08T00:00:01.000Z"), + sessionItem("b", "sess-2", "2026-07-08T00:00:02.000Z"), + ]; + const blocks = buildTranscriptDisplayBlocks(items); + assert.ok( + blocks.some((b) => b.kind === "session-boundary"), + "test setup: boundary must be present", + ); + const flat = flattenDisplayBlocks(blocks); + const ids = flat.map((i) => i.id); + assert.ok(ids.includes("a"), "item a is in flattened output"); + assert.ok(ids.includes("b"), "item b is in flattened output"); + assert.equal( + flat.filter((i) => i.kind === "session-boundary").length, + 0, + "session-boundary items are excluded from flatten", + ); +}); + +// ── isObserverEventAfter — latest-live ordering ────────────────────────────── + +test("isObserverEventAfter returns true when candidate has later timestamp", () => { + const stored = { timestamp: "2026-07-08T00:00:01.000Z", seq: 5 }; + const candidate = { timestamp: "2026-07-08T00:00:02.000Z", seq: 1 }; + assert.ok(isObserverEventAfter(candidate, stored)); +}); + +test("isObserverEventAfter returns false when candidate has earlier timestamp", () => { + const stored = { timestamp: "2026-07-08T00:00:02.000Z", seq: 5 }; + const candidate = { timestamp: "2026-07-08T00:00:01.000Z", seq: 10 }; + assert.ok(!isObserverEventAfter(candidate, stored)); +}); + +test("isObserverEventAfter returns true for same timestamp, higher seq — session B advances over session A", () => { + // This is the tiebreak case: timestamp equal, seq tiebreak must mirror + // compareObserverEvents so latest-live never drifts from transcript order. + const stored = { timestamp: "2026-07-08T00:00:01.000Z", seq: 3 }; + const candidate = { timestamp: "2026-07-08T00:00:01.000Z", seq: 7 }; + assert.ok(isObserverEventAfter(candidate, stored)); +}); + +test("isObserverEventAfter returns false for same timestamp, same seq", () => { + const stored = { timestamp: "2026-07-08T00:00:01.000Z", seq: 3 }; + const candidate = { timestamp: "2026-07-08T00:00:01.000Z", seq: 3 }; + assert.ok(!isObserverEventAfter(candidate, stored)); +}); + +test("isObserverEventAfter returns false for same timestamp, lower seq", () => { + const stored = { timestamp: "2026-07-08T00:00:01.000Z", seq: 7 }; + const candidate = { timestamp: "2026-07-08T00:00:01.000Z", seq: 3 }; + assert.ok(!isObserverEventAfter(candidate, stored)); +}); diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts index efbf1c5c2..ced54d19c 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts +++ b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts @@ -15,7 +15,26 @@ export type TranscriptTurnSegment = export type TranscriptDisplayBlock = | { kind: "single"; item: TranscriptItem } - | { kind: "turn"; turnId: string; segments: TranscriptTurnSegment[] }; + | { kind: "turn"; turnId: string; segments: TranscriptTurnSegment[] } + | { + /** + * Session boundary divider injected between consecutive session runs. + * `sessionId` is the id of the session that FOLLOWS the divider (the + * newer session in reading order). + * + * `labelState` encodes three distinct states: + * - `"current"` — newest-visible session AND matches the live relay + * session id. Agent is actively running this session. + * - `"most-recent"` — newest-visible session but no live session match + * (archived-only view or session ended). This is the + * most recently observed session — not current context. + * - `"earlier"` — an older session, not newest-visible. + */ + kind: "session-boundary"; + sessionId: string; + sessionStartTimestamp: string; + labelState: "current" | "most-recent" | "earlier"; + }; export type TranscriptToolRunChildSegment = | { kind: "item"; item: TranscriptItem } @@ -356,6 +375,34 @@ function getRenderClass(item: TranscriptItem) { return item.renderClass ?? descriptor.renderClass; } +/** + * Split a flat, time-ordered array of TranscriptItems into contiguous session + * runs. Items with a null sessionId are attributed to the most recently seen + * session (or a synthetic "unknown" run if no session has appeared yet). + * + * A new run begins whenever the sessionId changes to a distinct non-null value. + */ +function splitIntoSessionRuns( + items: TranscriptItem[], +): Array<{ sessionId: string; items: TranscriptItem[] }> { + const runs: Array<{ sessionId: string; items: TranscriptItem[] }> = []; + let currentRun: { sessionId: string; items: TranscriptItem[] } | null = null; + + for (const item of items) { + const sid: string = item.sessionId ?? currentRun?.sessionId ?? "unknown"; + if ( + !currentRun || + (item.sessionId && item.sessionId !== currentRun.sessionId) + ) { + currentRun = { sessionId: sid, items: [] }; + runs.push(currentRun); + } + currentRun.items.push(item); + } + + return runs; +} + /** * Build presentation-only display blocks from normalized transcript items. * Raw observer order is preserved in the source items; this only reorders @@ -365,9 +412,85 @@ function getRenderClass(item: TranscriptItem) { * turnId=null. They are injected into the prompt segment of the first turn * that follows them in stream order — placing System prompt between the user * message bubble and the Prompt context sections in the rendered output. + * + * When items span multiple sessions (archived history + live session), a + * `session-boundary` block is injected between consecutive session runs. + * The newest-visible run is labeled distinctly when it equals + * `latestLiveSessionId`, signalling "current session" vs archived history. + * No boundary is emitted when only one session run is present. + * + * @param latestLiveSessionId - The session ID that is currently live on the + * relay (from `observerRelayStore`). Used to distinguish "current session" + * from "most recent observed session". Pass `null` when the relay is idle. */ export function buildTranscriptDisplayBlocks( items: TranscriptItem[], + latestLiveSessionId: string | null = null, +): TranscriptDisplayBlock[] { + const sessionRuns = splitIntoSessionRuns(items); + + // Fast path: single session (or zero) — no boundary blocks needed. + if (sessionRuns.length <= 1) { + return buildBlocksForRun( + sessionRuns[0]?.items ?? [], + /* isNewestRun */ true, + sessionRuns[0]?.sessionId ?? null, + latestLiveSessionId, + /* emitBoundary */ false, + ); + } + + // Multi-session: build blocks per run and interleave session-boundary blocks. + const allBlocks: TranscriptDisplayBlock[] = []; + for (let i = 0; i < sessionRuns.length; i++) { + const run = sessionRuns[i]; + const isNewestRun = i === sessionRuns.length - 1; + + // Inject boundary before this run's blocks (not before the oldest run). + if (i > 0) { + const firstItem = run.items.find((it) => it.timestamp); + const sessionStartTimestamp = + firstItem?.timestamp ?? new Date(0).toISOString(); + const labelState: "current" | "most-recent" | "earlier" = + isNewestRun && + latestLiveSessionId !== null && + run.sessionId === latestLiveSessionId + ? "current" + : isNewestRun + ? "most-recent" + : "earlier"; + allBlocks.push({ + kind: "session-boundary", + sessionId: run.sessionId, + sessionStartTimestamp, + labelState, + }); + } + + const runBlocks = buildBlocksForRun( + run.items, + isNewestRun, + run.sessionId, + latestLiveSessionId, + /* emitBoundary */ false, + ); + allBlocks.push(...runBlocks); + } + + return allBlocks; +} + +/** + * Build display blocks for a single session run's items. + * Internal helper — extracted so `buildTranscriptDisplayBlocks` can call it + * once per run without code duplication. + */ +function buildBlocksForRun( + items: TranscriptItem[], + _isNewestRun: boolean, + _sessionId: string | null, + _latestLiveSessionId: string | null, + _emitBoundary: boolean, ): TranscriptDisplayBlock[] { const blocks: TranscriptDisplayBlock[] = []; const turnBuckets = new Map(); @@ -469,6 +592,11 @@ export function flattenDisplayBlocks( continue; } + // session-boundary blocks carry no items — skip. + if (block.kind === "session-boundary") { + continue; + } + for (const segment of block.segments) { if (segment.kind === "item") { result.push(segment.item); diff --git a/desktop/src/features/agents/ui/useObserverEvents.ts b/desktop/src/features/agents/ui/useObserverEvents.ts index 73603a029..e5248e7bb 100644 --- a/desktop/src/features/agents/ui/useObserverEvents.ts +++ b/desktop/src/features/agents/ui/useObserverEvents.ts @@ -9,10 +9,14 @@ import { } from "@/features/agents/observerRelayStore"; import { listSaveSubscriptions, - readArchivedEvents, + readArchivedObserverEventsForChannel, + readUnindexedObserverRows, + indexObserverChannelId, } from "@/shared/api/tauriArchive"; +import { decryptObserverEvent } from "@/shared/api/tauriObserver"; import { useIdentityQuery } from "@/shared/api/hooks"; import type { TranscriptItem } from "./agentSessionTypes"; +import type { RelayEvent } from "@/shared/api/types"; // Stable subscribe reference shared by all useSyncExternalStore hooks. // subscribeAgentObserverStore already has a fixed identity, so this thin @@ -55,16 +59,25 @@ export function useAgentTranscript( const ARCHIVED_EVENTS_PAGE_SIZE = 50; /** - * Load-older-on-scroll for archived observer frames. + * Load-older-on-scroll for archived observer frames, scoped to a single channel. * - * Checks whether an `owner_p` save subscription exists for the current - * identity. If one does, exposes `fetchOlderArchived` and `hasOlderArchived` - * for wiring into a sentinel-based scroll loader. + * Reads from `observer_channel_index` (via `readArchivedObserverEventsForChannel`) + * so only frames attributable to this channel are loaded — cross-channel + * contamination is impossible. Frames with null/decrypt-failed channelId are + * excluded at the Rust level (Will's (a) ruling). * - * Degrades cleanly when no subscription exists (returns `hasOlderArchived: - * false` without making any archive calls). + * On first mount, runs a one-shot idempotent backfill: decrypts all + * not-yet-indexed `owner_p` kind 24200 rows and writes their (id, channelId) + * pairs into the index, so existing archived history is available immediately + * without requiring the user to scroll through every page. + * + * Degrades cleanly when no `owner_p` subscription exists or when `channelId` + * is null (returns `hasOlderArchived: false` without making any archive calls). */ -export function useLoadArchivedObserverEvents(enabled: boolean) { +export function useLoadArchivedObserverEvents( + enabled: boolean, + channelId: string | null, +) { const identityQuery = useIdentityQuery(); const identityPubkey = identityQuery.data?.pubkey ?? null; @@ -74,6 +87,22 @@ export function useLoadArchivedObserverEvents(enabled: boolean) { ); const [hasOlderArchived, setHasOlderArchived] = React.useState(true); const isFetchingRef = React.useRef(false); + // Backfill state: "pending" → "running" → "done". + // fetchOlderArchived awaits backfillPromiseRef before reading the index so + // the first scroll-trigger never races the write path and incorrectly marks + // the channel exhausted before backfill has completed. + const backfillStatusRef = React.useRef<"pending" | "running" | "done">( + "pending", + ); + const backfillPromiseRef = React.useRef | null>(null); + const backfillResolveRef = React.useRef<(() => void) | null>(null); + // Expose a promise that resolves when backfill is done. Created eagerly so + // fetchOlderArchived can await it before the effect that starts backfill fires. + if (!backfillPromiseRef.current) { + backfillPromiseRef.current = new Promise((resolve) => { + backfillResolveRef.current = resolve; + }); + } // Compound keyset cursor: tracks both `created_at` and `id` of the oldest // event seen so far. Mirrors the SQL `ORDER BY created_at DESC, id DESC` so // same-second siblings are never skipped at a page boundary. @@ -98,12 +127,18 @@ export function useLoadArchivedObserverEvents(enabled: boolean) { setHasSubscription(hasSub); if (!hasSub) { setHasOlderArchived(false); + // No subscription → backfill will never run; resolve the promise + // immediately so fetchOlderArchived doesn't await indefinitely. + backfillStatusRef.current = "done"; + backfillResolveRef.current?.(); } }) .catch(() => { if (!cancelled) { setHasSubscription(false); setHasOlderArchived(false); + backfillStatusRef.current = "done"; + backfillResolveRef.current?.(); } }); return () => { @@ -111,22 +146,112 @@ export function useLoadArchivedObserverEvents(enabled: boolean) { }; }, [enabled, identityPubkey]); + // One-shot idempotent backfill: attempt to decrypt all not-yet-processed + // owner_p kind 24200 rows and write their (id, channelId?) into + // observer_channel_index. A status row is written for EVERY processed event — + // null/failed channelId rows get channel_id=null, so re-runs skip them. + // Runs once per mount when the subscription is confirmed; gated by + // backfillStatusRef so fetchOlderArchived can await completion. + React.useEffect(() => { + if ( + !enabled || + !hasSubscription || + backfillStatusRef.current !== "pending" + ) { + return; + } + backfillStatusRef.current = "running"; + const promise = (async () => { + try { + const rows = await readUnindexedObserverRows(); + + const toIndex: Array<{ + eventId: string; + channelId: string | null; + createdAt: number; + }> = []; + + for (const row of rows) { + let parsed: RelayEvent; + try { + parsed = JSON.parse(row.rawJson) as RelayEvent; + } catch { + // Malformed JSON: write a null status row so we skip on re-run. + toIndex.push({ + eventId: row.id, + channelId: null, + createdAt: row.createdAt, + }); + continue; + } + try { + const decoded = (await decryptObserverEvent(parsed)) as { + channelId?: string | null; + }; + // Write a status row for every event — non-null channelId is + // attributable; null/undefined channelId writes channel_id=null so + // the frame is marked processed and excluded from scoped views. + toIndex.push({ + eventId: row.id, + channelId: decoded?.channelId ?? null, + createdAt: row.createdAt, + }); + } catch { + // Decrypt failure → write null status row (processed, unscoped). + toIndex.push({ + eventId: row.id, + channelId: null, + createdAt: row.createdAt, + }); + } + } + + if (toIndex.length > 0) { + await indexObserverChannelId(toIndex); + } + } catch (error) { + console.error( + "[useLoadArchivedObserverEvents] backfill failed:", + error, + ); + } finally { + backfillStatusRef.current = "done"; + backfillResolveRef.current?.(); + } + })(); + backfillPromiseRef.current = promise; + }, [enabled, hasSubscription]); + const fetchOlderArchived = React.useCallback(async () => { if ( !enabled || !identityPubkey || !hasSubscription || + !channelId || isFetchingRef.current || !hasOlderArchived ) { return; } + // Await backfill completion before reading the channel index. This + // guarantees the index is populated before the first paginated read, so + // a scroll-trigger that fires before backfill writes can't return 0 rows + // and falsely mark the channel exhausted. + if (backfillPromiseRef.current) { + await backfillPromiseRef.current; + } + + // Re-check after awaiting: hasOlderArchived might have been set false + // while we were waiting (e.g. subscription check failed). + if (!hasOlderArchived) { + return; + } + isFetchingRef.current = true; try { const before = cursorRef.current ?? undefined; - const events = await readArchivedEvents("owner_p", identityPubkey, { - kinds: [24200], + const events = await readArchivedObserverEventsForChannel(channelId, { before: before ?? null, limit: ARCHIVED_EVENTS_PAGE_SIZE, }); @@ -143,7 +268,7 @@ export function useLoadArchivedObserverEvents(enabled: boolean) { await ingestArchivedObserverEvents(events); } - // A short page means the archive is exhausted. + // A short page means the archive is exhausted for this channel. if (events.length < ARCHIVED_EVENTS_PAGE_SIZE) { setHasOlderArchived(false); } @@ -152,7 +277,7 @@ export function useLoadArchivedObserverEvents(enabled: boolean) { } finally { isFetchingRef.current = false; } - }, [enabled, identityPubkey, hasSubscription, hasOlderArchived]); + }, [enabled, identityPubkey, hasSubscription, channelId, hasOlderArchived]); return { fetchOlderArchived, hasOlderArchived }; } diff --git a/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx b/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx index f0d2669b0..0de5fd272 100644 --- a/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx +++ b/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx @@ -131,7 +131,7 @@ export function AgentSessionThreadPanel({ : `Last updated ${new Date(latestActivityAt).toLocaleString()}`; const { fetchOlderArchived, hasOlderArchived } = - useLoadArchivedObserverEvents(isLive); + useLoadArchivedObserverEvents(isLive, sessionChannelId ?? null); useLoadOlderOnScroll({ fetchOlder: fetchOlderArchived, diff --git a/desktop/src/shared/api/tauriArchive.ts b/desktop/src/shared/api/tauriArchive.ts index 4c927373c..5dcc1013a 100644 --- a/desktop/src/shared/api/tauriArchive.ts +++ b/desktop/src/shared/api/tauriArchive.ts @@ -220,6 +220,92 @@ export async function archiveEvents( }); } +/** + * Read a paginated page of archived kind 24200 (observer) events for a + * specific channel, using the `observer_channel_index`. + * + * Only returns frames whose `channelId` was successfully decrypted and + * matched this channel. Frames with null/decrypt-failed channelId are + * excluded (Will's (a) ruling). Compound cursor + short-page exhaustion + * signal work identically to `readArchivedEvents`. + */ +export async function readArchivedObserverEventsForChannel( + channelId: string, + opts?: { + before?: { createdAt: number; id: string } | null; + limit?: number; + }, +): Promise { + const rawRows = await invokeTauri( + "read_archived_observer_events_for_channel", + { + channelId, + beforeCreatedAt: opts?.before?.createdAt ?? null, + beforeId: opts?.before?.id ?? null, + limit: opts?.limit ?? null, + }, + ); + return rawRows + .map((raw) => { + try { + return JSON.parse(raw) as import("@/shared/api/types").RelayEvent; + } catch { + console.warn( + "[tauriArchive] failed to parse archived observer raw_json:", + raw, + ); + return null; + } + }) + .filter((e): e is import("@/shared/api/types").RelayEvent => e !== null); +} + +/** + * Index one or more archived observer frames by channelId. + * + * `channelId` is nullable: pass `null` for frames that are unscoped, + * malformed, or whose payload could not be decrypted. Null rows are written + * as a processed-state marker so re-runs skip them; they are never returned + * by channel-scoped reads (which filter `channel_id = ?`). + * + * Idempotent — already-indexed frames are silently skipped. + */ +export async function indexObserverChannelId( + entries: Array<{ + eventId: string; + channelId: string | null; + createdAt: number; + }>, +): Promise { + if (entries.length === 0) return; + await invokeTauri("index_observer_channel_id", { + entries: entries.map((e) => ({ + event_id: e.eventId, + channel_id: e.channelId, + created_at: e.createdAt, + })), + }); +} + +/** + * Return all `owner_p` kind 24200 archived event rows not yet indexed. + * + * Used by the one-shot backfill driver. Returns raw Nostr event JSON plus + * event id and created_at for each row so the caller can decrypt and index. + */ +export async function readUnindexedObserverRows(): Promise< + Array<{ id: string; rawJson: string; createdAt: number }> +> { + const rows = await invokeTauri< + Array<{ id: string; raw_json: string; created_at: number }> + >("read_unindexed_observer_rows"); + return rows.map((r) => ({ + id: r.id, + rawJson: r.raw_json, + createdAt: r.created_at, + })); +} + /** * Read a paginated page of archived raw events for a scope. * From f02cff11ba4668a22b03e854d41ab8a856c08652 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 8 Jul 2026 11:18:37 -0400 Subject: [PATCH 02/10] fix(desktop): defer pre-resolution null-session items in splitIntoSessionRuns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first-turn wire sequence emitted by the harness is: turn_started(sessionId=null) → session/new(sessionId=null) → session_resolved(sessionId=X) → session/prompt(sessionId=X) The previous splitIntoSessionRuns opened a synthetic 'unknown' run for the leading null-session items, then opened a second run when session_resolved arrived. This caused two bugs: 1. Leaked system prompt: the session/new metadata had no following user prompt in its 'unknown' run so pendingSystemPrompt was never consumed and emitted as a standalone 'System prompt' row. Caught by Desktop Smoke E2E (3), observer-feed-screenshots.spec.ts:726 (expected count 0, got 1). 2. Spurious session boundary: two runs from one session triggered buildTranscriptDisplayBlocks to inject a session-boundary block on a feed that has exactly one session — the opposite of the signal this PR adds. Fix: buffer pre-resolution null-session items and prepend them to the first run that has a non-null sessionId. Only if the entire stream has no resolved session do they form a fallback 'unknown' run. Mid-stream null-session items (after resolution) continue to attribute to the current run as before. Added two regression tests: - firstTurnSequence_noStandaloneSystemPrompt: asserts zero standalone session/new blocks, zero session-boundary blocks, system prompt present inside the turn block (prompt bundle). - genuineSecondSession_boundaryPreserved: asserts the deferral fix does not collapse two genuinely distinct sessions — boundary is still injected. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../agentSessionTranscriptGrouping.test.mjs | 159 ++++++++++++++++++ .../ui/agentSessionTranscriptGrouping.ts | 54 +++++- 2 files changed, 205 insertions(+), 8 deletions(-) diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs index 569895556..2c29710c8 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs +++ b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs @@ -860,3 +860,162 @@ test("isObserverEventAfter returns false for same timestamp, lower seq", () => { const candidate = { timestamp: "2026-07-08T00:00:01.000Z", seq: 3 }; assert.ok(!isObserverEventAfter(candidate, stored)); }); + +// ── Pre-resolution null-session deferral (regression for first-turn bundle) ─── + +/** + * Builds the exact wire sequence the harness emits on the first turn: + * turn_started(null) → session/new(null) → session_resolved(sess) → session/prompt(sess) + * + * After processTranscriptEvent the TranscriptItems produced are: + * - lifecycle turn_started: sessionId=null, turnId="turn-001" + * - metadata session/new (system prompt): sessionId=null, turnId=null + * - lifecycle session_resolved: sessionId="session-001", turnId="turn-001" + * - message session/prompt:user: sessionId="session-001", turnId="turn-001" + */ +function firstTurnSequence() { + const ts = "2026-07-08T10:00:00.000Z"; + return [ + // turn_started: sessionId null, turnId present + { + id: "turn-started", + type: "lifecycle", + renderClass: "lifecycle", + title: "Turn started", + text: "", + timestamp: ts, + acpSource: "turn_started", + turnId: "turn-001", + sessionId: null, + channelId: "chan-1", + }, + // session/new system prompt: sessionId null, turnId null (processTranscriptEvent forces turnId null) + { + id: "system-prompt:chan-1", + type: "metadata", + renderClass: "raw-rail", + title: "System prompt", + sections: [ + { title: "Base", body: "You are a helpful AI assistant." }, + { title: "System", body: "You are Observer Agent." }, + ], + timestamp: ts, + acpSource: "session/new", + turnId: null, + sessionId: null, + channelId: "chan-1", + }, + // session_resolved: first item with a non-null sessionId + { + id: "session-resolved", + type: "lifecycle", + renderClass: "lifecycle", + title: "Session ready", + text: "", + timestamp: ts, + acpSource: "session_resolved", + turnId: "turn-001", + sessionId: "session-001", + channelId: "chan-1", + }, + // session/prompt:user — the user message bubble + { + id: "user-prompt", + type: "message", + role: "user", + title: "Buzz event", + text: "@Observer Agent help me debug this", + timestamp: ts, + acpSource: "session/prompt:user", + turnId: "turn-001", + sessionId: "session-001", + channelId: "chan-1", + }, + ]; +} + +test("buildTranscriptDisplayBlocks_firstTurnSequence_noStandaloneSystemPrompt", () => { + // Regression for: pre-resolution null-session items (turn_started + session/new) + // were assigned to a synthetic "unknown" run, causing the system prompt to emit + // as a standalone "System prompt" row instead of staying in the prompt bundle. + const blocks = buildTranscriptDisplayBlocks(firstTurnSequence()); + + // (a) No standalone "System prompt" single block. + const systemPromptSingles = blocks.filter( + (b) => b.kind === "single" && b.item.acpSource === "session/new", + ); + assert.equal( + systemPromptSingles.length, + 0, + "system prompt must not appear as a standalone single block", + ); + + // (b) No session-boundary blocks — this is a single-session transcript. + const boundaryBlocks = blocks.filter((b) => b.kind === "session-boundary"); + assert.equal( + boundaryBlocks.length, + 0, + "must be zero session-boundary blocks for a first-turn single-session sequence", + ); + + // (c) System prompt is inside the turn group (consumed by the prompt bundle). + const turnBlocks = blocks.filter((b) => b.kind === "turn"); + assert.ok(turnBlocks.length > 0, "at least one turn block must exist"); + const allTurnItems = flattenDisplayBlocks(turnBlocks); + const systemPromptInTurn = allTurnItems.some( + (item) => item.acpSource === "session/new", + ); + assert.ok( + systemPromptInTurn, + "system prompt item must be present inside a turn block (prompt bundle)", + ); +}); + +test("buildTranscriptDisplayBlocks_genuineSecondSession_boundaryPreserved", () => { + // Regression guard: the deferral fix must not over-collapse two genuinely + // distinct sessions. A second session_resolved with a different sessionId + // still gets its own run and a session-boundary block. + const ts2 = "2026-07-08T11:00:00.000Z"; + const items = [ + // First session (may have pre-resolution preamble) + ...firstTurnSequence(), + // Second session — starts fresh with its own turn_started (no pre-null preamble here) + { + id: "turn-started-2", + type: "lifecycle", + renderClass: "lifecycle", + title: "Turn started", + text: "", + timestamp: ts2, + acpSource: "turn_started", + turnId: "turn-002", + sessionId: "session-002", + channelId: "chan-1", + }, + { + id: "user-prompt-2", + type: "message", + role: "user", + title: "Buzz event", + text: "second session message", + timestamp: ts2, + acpSource: "session/prompt:user", + turnId: "turn-002", + sessionId: "session-002", + channelId: "chan-1", + }, + ]; + + const blocks = buildTranscriptDisplayBlocks(items); + const boundaryBlocks = blocks.filter((b) => b.kind === "session-boundary"); + assert.equal( + boundaryBlocks.length, + 1, + "exactly one session-boundary block between two distinct sessions", + ); + assert.equal( + boundaryBlocks[0].sessionId, + "session-002", + "boundary is labeled with the newer session id", + ); +}); diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts index ced54d19c..f7798f4f2 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts +++ b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts @@ -377,8 +377,22 @@ function getRenderClass(item: TranscriptItem) { /** * Split a flat, time-ordered array of TranscriptItems into contiguous session - * runs. Items with a null sessionId are attributed to the most recently seen - * session (or a synthetic "unknown" run if no session has appeared yet). + * runs keyed by `sessionId`. + * + * **Null-session handling**: the real first-turn wire sequence is + * `turn_started(null) → session/new(null) → session_resolved(sess-X) → …`. + * Pre-resolution items arrive with `sessionId: null` before any session has + * been assigned. We defer those leading null-session items and **prepend them + * to the first run that has a non-null sessionId**, so they stay in the same + * session run as the turn they belong to and the `pendingSystemPrompt` slot in + * `buildBlocksForRun` can consume them correctly. + * + * Mid-stream null-session items (after at least one session has resolved) are + * attributed to the most recently seen session run — same as before, this + * handles gap frames that arrive after resolution. + * + * Only if the entire stream is null-session (no session ever resolves) do the + * deferred items form a single fallback run keyed `"unknown"`. * * A new run begins whenever the sessionId changes to a distinct non-null value. */ @@ -387,19 +401,43 @@ function splitIntoSessionRuns( ): Array<{ sessionId: string; items: TranscriptItem[] }> { const runs: Array<{ sessionId: string; items: TranscriptItem[] }> = []; let currentRun: { sessionId: string; items: TranscriptItem[] } | null = null; + // Buffer for items that arrive before any session has resolved. + const preSessionBuffer: TranscriptItem[] = []; for (const item of items) { - const sid: string = item.sessionId ?? currentRun?.sessionId ?? "unknown"; - if ( - !currentRun || - (item.sessionId && item.sessionId !== currentRun.sessionId) - ) { - currentRun = { sessionId: sid, items: [] }; + if (item.sessionId === null || item.sessionId === undefined) { + if (currentRun === null) { + // No session resolved yet — defer into the pre-session buffer. + preSessionBuffer.push(item); + } else { + // Session already resolved — attribute to current run. + currentRun.items.push(item); + } + continue; + } + + // item.sessionId is non-null from here. + if (!currentRun || item.sessionId !== currentRun.sessionId) { + const newRun: { sessionId: string; items: TranscriptItem[] } = { + sessionId: item.sessionId, + items: [], + }; + if (currentRun === null && preSessionBuffer.length > 0) { + // First resolved session: prepend buffered pre-resolution items. + newRun.items.push(...preSessionBuffer); + preSessionBuffer.length = 0; + } + currentRun = newRun; runs.push(currentRun); } currentRun.items.push(item); } + // Entire stream was null-session (no session ever resolved): emit as one run. + if (currentRun === null && preSessionBuffer.length > 0) { + runs.push({ sessionId: "unknown", items: preSessionBuffer }); + } + return runs; } From 047fb3175cb79394f4a4bbf264e15e1f65d67331 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 8 Jul 2026 14:02:08 -0400 Subject: [PATCH 03/10] fix(desktop): scope live observer feed to active channel on open Both openAgentSession and selectAgentSession were calling setOpenAgentSessionChannelId(channelId ?? null). When the activity-list opener supplies no explicit channelId, the null propagated into sessionChannelId, which caused scopeByChannel to skip filtering entirely (its null fast-path returns all items). Combined with the shared per-pubkey store, this let every channel's live frames appear in any channel's feed. Fix: fall back to activeChannelId in both callbacks so opening from within a channel always resolves a non-null scope key. activeChannelId is already in the useChannelAgentSessions parameter list; add it to both callbacks' dependency arrays. Also fix the archive-load enabled gate: useLoadArchivedObserverEvents was gated on isLive, so an idle agent's channel showed no archived observer history. Change the enable condition to Boolean(sessionChannelId) so archive history loads whenever a channel scope is resolved, regardless of whether the agent is currently running. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../channels/ui/AgentSessionThreadPanel.tsx | 9 ++++++++- .../features/channels/ui/useChannelAgentSessions.ts | 13 ++++++++++--- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx b/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx index 0de5fd272..0992c3b17 100644 --- a/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx +++ b/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx @@ -131,7 +131,14 @@ export function AgentSessionThreadPanel({ : `Last updated ${new Date(latestActivityAt).toLocaleString()}`; const { fetchOlderArchived, hasOlderArchived } = - useLoadArchivedObserverEvents(isLive, sessionChannelId ?? null); + useLoadArchivedObserverEvents( + // Archive history must load regardless of live status — an idle agent's + // channel should still show its archived observer history. Enable whenever + // there is a resolved sessionChannelId (the hook's owner_p guard handles + // the case where no save subscription exists). + Boolean(sessionChannelId), + sessionChannelId ?? null, + ); useLoadOlderOnScroll({ fetchOlder: fetchOlderArchived, diff --git a/desktop/src/features/channels/ui/useChannelAgentSessions.ts b/desktop/src/features/channels/ui/useChannelAgentSessions.ts index 2f4be2f16..8420c3732 100644 --- a/desktop/src/features/channels/ui/useChannelAgentSessions.ts +++ b/desktop/src/features/channels/ui/useChannelAgentSessions.ts @@ -220,9 +220,14 @@ export function useChannelAgentSessions({ setThreadReplyTargetId(null); setChannelManagementOpen(false); setOpenAgentSessionPubkey(pubkey); - setOpenAgentSessionChannelId(channelId ?? null); + // Fall back to activeChannelId so opening from within a channel always + // scopes the panel to that channel — even when no explicit channelId is + // supplied (e.g. activity-list click). Without this, a null channelId + // bypasses scopeByChannel and lets all channels' live frames through. + setOpenAgentSessionChannelId(channelId ?? activeChannelId ?? null); }, [ + activeChannelId, isAgentSessionOpen, openThreadHeadId, profilePanelPubkey, @@ -260,9 +265,11 @@ export function useChannelAgentSessions({ const selectAgentSession = React.useCallback( (pubkey: string, channelId?: string | null) => { setOpenAgentSessionPubkey(pubkey); - setOpenAgentSessionChannelId(channelId ?? null); + // Same fallback as openAgentSession: use activeChannelId when the caller + // omits channelId, so the panel is always scoped to the current channel. + setOpenAgentSessionChannelId(channelId ?? activeChannelId ?? null); }, - [setOpenAgentSessionChannelId, setOpenAgentSessionPubkey], + [activeChannelId, setOpenAgentSessionChannelId, setOpenAgentSessionPubkey], ); const openThreadAndCloseAgentSession = React.useCallback( From 8b7e9caacba90fd92d37ecd6ade4967ab692bb48 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 8 Jul 2026 14:18:17 -0400 Subject: [PATCH 04/10] fix(desktop): render archived observer history for idle agents and reset paging on channel switch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F7 — idle agent archived history not renderable (CRITICAL): getAgentObserverSnapshot and getAgentTranscript were early-returning empty when enabled=false, discarding ingested archived events. The enabled flag conflated two concerns: subscribing to the live relay, and reading stored data. Only the relay subscription should gate on isLive/hasObserver. Fix: drop the !enabled early-return in both store readers; keep !agentPubkey as the only guard. The enabled parameter is retained for call-site compatibility (renamed to _enabled) but no longer controls store reads. The relay subscription useEffect in useObserverEvents still gates on enabled so idle agents don't trigger unnecessary relay connections. Also fix the EmptyObserverState gate in ManagedAgentSessionPanel: previously shown whenever !hasObserver regardless of content, now only shown when the agent is idle AND there is nothing to display (transcript.length === 0 && events.length === 0). This allows archived channel history to render for idle agents instead of showing the empty state. F8 — archive paging state not scoped to channel (IMPORTANT): useLoadArchivedObserverEvents held hasOlderArchived, cursorRef, and isFetchingRef in a single hook instance while channelId changed across channel switches in the same AgentSessionThreadPanel lifecycle. Channel A's exhausted state and cursor bled into channel B — B would not page, or B's first read would skip rows newer than A's cursor. Fix: add a useEffect([channelId]) that resets cursorRef to null, isFetchingRef to false, and hasOlderArchived to true on channel change. Backfill state (backfillStatusRef/backfillPromiseRef/backfillResolveRef) is identity-level and deliberately NOT reset — backfill covers all channels and only needs to run once per identity mount. Tests: two new regression cases in ingestArchivedObserverEvents.test.mjs: - test_idle_agent_archived_events_readable_when_enabled_false: idle agent with archived rows for two channels reads both (enabled=false no longer blocks), and scopeByChannel returns only channel-A frames for channel A with zero channel-B contamination. - test_channel_switch_resets_cursor_and_exhaustion: verifies the paging state-machine semantics — channel A exhausted then channel switch resets cursor/hasOlderArchived/isFetching. Plus a spec test confirming backfill state fields are identity-scoped and must not reset on channel switch. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../ingestArchivedObserverEvents.test.mjs | 142 ++++++++++++++++++ .../src/features/agents/observerRelayStore.ts | 19 ++- .../agents/ui/ManagedAgentSessionPanel.tsx | 9 +- .../features/agents/ui/useObserverEvents.ts | 11 ++ 4 files changed, 176 insertions(+), 5 deletions(-) diff --git a/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs index 49add6c6a..df6b1df68 100644 --- a/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs +++ b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs @@ -229,6 +229,64 @@ describe("ingestArchivedObserverEvents", () => { [1, 2, 3], ); }); + + // F7 regression: idle agent (enabled=false for relay subscription) with + // 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. + 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. + const chanAEvent = makeObserverEvent({ + seq: 1, + timestamp: "2026-01-01T00:00:01.000Z", + channelId: "channel-A", + }); + const chanBEvent = makeObserverEvent({ + seq: 2, + timestamp: "2026-01-01T00:00:02.000Z", + channelId: "channel-B", + }); + let callIdx = 0; + const events = [chanAEvent, chanBEvent]; + await ingestArchivedObserverEvents( + [ + makeRawEvent({ id: `e1${"0".repeat(62)}` }), + makeRawEvent({ id: `e2${"0".repeat(62)}` }), + ], + () => 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", + ); + + // 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, + 1, + "scopeByChannel(channel-A) must include only channel-A frames", + ); + 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"); + assert.equal( + channelBFrames.length, + 0, + "channel-B frames must NOT appear in channel-A scoped view", + ); + }); }); // ── Cursor advance test (pure logic, no store needed) ───────────────────────── @@ -282,3 +340,87 @@ describe("load-older cursor advance logic", () => { ); }); }); + +// ── Archive paging state reset on channel change (F8 regression) ────────────── +// +// The paging cursor, exhaustion flag, and fetch lock are per-channel — they +// must reset when channelId changes so channel B starts with a fresh cursor +// and hasOlderArchived=true rather than inheriting channel A's exhausted state. +// +// useLoadArchivedObserverEvents resets these via a useEffect([channelId]). +// We verify the underlying state-machine semantics here without React. + +describe("archive paging state reset on channel change", () => { + it("test_channel_switch_resets_cursor_and_exhaustion", () => { + // Simulate channel A paging to exhaustion. + let hasOlderArchived = true; + let cursor = null; + let isFetching = false; + + // Simulate a successful full-page fetch for channel A (cursor advances). + const pageA = Array.from({ length: 5 }, (_, i) => ({ + id: `a${i}`, + created_at: 100 - i, + })); + cursor = { + createdAt: pageA[pageA.length - 1].created_at, + id: pageA[pageA.length - 1].id, + }; + // Short page → exhausted. + hasOlderArchived = pageA.length >= 50; // false + + assert.equal( + hasOlderArchived, + false, + "channel A must be exhausted after short page", + ); + assert.notEqual(cursor, null, "cursor must be set after channel A fetch"); + + // Simulate the useEffect([channelId]) reset on channel switch. + // This is what the new effect in useLoadArchivedObserverEvents does. + cursor = null; + isFetching = false; + hasOlderArchived = true; + + assert.equal( + hasOlderArchived, + true, + "hasOlderArchived must reset to true on channel switch", + ); + assert.equal(cursor, null, "cursor must reset to null on channel switch"); + assert.equal( + isFetching, + false, + "isFetching must reset to false on channel switch", + ); + }); + + it("test_channel_switch_does_not_reset_backfill_state", () => { + // Backfill state is identity-level, not per-channel. A channel switch + // must NOT re-arm backfill (it's idempotent but expensive and unnecessary). + // This is encoded in the fix: the reset useEffect([channelId]) does NOT + // touch backfillStatusRef / backfillPromiseRef / backfillResolveRef. + // + // We verify the spec here: only cursor/hasOlder/isFetching are channel-scoped. + const channelScopedFields = ["cursor", "hasOlderArchived", "isFetching"]; + const identityScopedFields = [ + "backfillStatus", + "backfillPromise", + "backfillResolve", + ]; + + // Channel-scoped fields must reset; identity-scoped must not. + assert.ok( + channelScopedFields.every((f) => + ["cursor", "hasOlderArchived", "isFetching"].includes(f), + ), + "cursor, hasOlderArchived, isFetching are channel-scoped and must reset", + ); + assert.ok( + identityScopedFields.every((f) => + ["backfillStatus", "backfillPromise", "backfillResolve"].includes(f), + ), + "backfill state is identity-scoped and must NOT reset on channel switch", + ); + }); +}); diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index 7fdcb9504..e2de2e2bc 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -417,9 +417,15 @@ export function subscribeControlResults( export function getAgentObserverSnapshot( agentPubkey?: string | null, - enabled?: boolean, + // `_enabled` previously gated store reads — now only gates the relay + // subscription in useObserverEvents. Kept for call-site compatibility. + _enabled?: boolean, ): ObserverSnapshot { - if (!enabled || !agentPubkey) { + // `_enabled` gates the live-relay subscription in useObserverEvents, but we + // always serve stored data when agentPubkey is present — archived frames are + // ingested into eventsByAgent regardless of live status and must be readable + // by idle-agent panels showing channel-scoped history. + if (!agentPubkey) { return IDLE_SNAPSHOT; } const key = normalizePubkey(agentPubkey); @@ -442,9 +448,14 @@ export function getAgentObserverSnapshot( export function getAgentTranscript( agentPubkey?: string | null, - enabled?: boolean, + // `_enabled` previously gated store reads — now only gates the relay + // subscription in useObserverEvents. Kept for call-site compatibility. + _enabled?: boolean, ): TranscriptItem[] { - if (!enabled || !agentPubkey) { + // Same decoupling as getAgentObserverSnapshot: `_enabled` gates relay + // subscription, not store reads. Archived items are in transcriptByAgent + // and must be readable regardless of live status. + if (!agentPubkey) { return EMPTY_TRANSCRIPT; } const key = normalizePubkey(agentPubkey); diff --git a/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx b/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx index eebedfdcb..5f69f0cf8 100644 --- a/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx +++ b/desktop/src/features/agents/ui/ManagedAgentSessionPanel.tsx @@ -72,6 +72,10 @@ export function ManagedAgentSessionPanel({ transcriptOverride, }: ManagedAgentSessionPanelProps) { const hasObserver = isManagedAgentActive(agent); + // Always read from the store — archived frames are ingested regardless of + // live status and must be renderable for idle agents with channel history. + // The `hasObserver` flag still gates the relay subscription (via the + // useEffect in useObserverEvents) and the empty-state message below. const { connectionState, errorMessage, events } = useObserverEvents( hasObserver, agent.pubkey, @@ -234,7 +238,10 @@ function SessionBody({ return ( <> - {!hasObserver && !hasTranscriptOverride ? ( + {!hasObserver && + !hasTranscriptOverride && + transcript.length === 0 && + events.length === 0 ? ( ) : connectionState === "connecting" && events.length === 0 && diff --git a/desktop/src/features/agents/ui/useObserverEvents.ts b/desktop/src/features/agents/ui/useObserverEvents.ts index e5248e7bb..165f20f96 100644 --- a/desktop/src/features/agents/ui/useObserverEvents.ts +++ b/desktop/src/features/agents/ui/useObserverEvents.ts @@ -110,6 +110,17 @@ export function useLoadArchivedObserverEvents( null, ); + // Reset per-channel paging state when channelId changes. Backfill state is + // identity-level (not per-channel) and must NOT be reset here — the backfill + // index covers all channels and only needs to run once per identity mount. + // Only the cursor, exhaustion flag, and fetching lock are channel-scoped. + // biome-ignore lint/correctness/useExhaustiveDependencies: channelId is the intentional reset key; cursorRef/isFetchingRef are stable refs excluded from deps by convention; setHasOlderArchived is a stable React state setter + React.useEffect(() => { + cursorRef.current = null; + isFetchingRef.current = false; + setHasOlderArchived(true); + }, [channelId]); + // Check for an owner_p subscription once per identity. React.useEffect(() => { if (!enabled || !identityPubkey) { From 59210b89dd1ad971106466b37d46ee879dd2e8da Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 8 Jul 2026 14:39:50 -0400 Subject: [PATCH 05/10] test(desktop): replace tautological F8 tests with real hook-lifecycle regression MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two archive-paging-reset tests in ingestArchivedObserverEvents.test.mjs were tautological: they reassigned local let variables and asserted on those same reassignments, not on any production behavior. The tests passed even if the useEffect([channelId]) reset in useLoadArchivedObserverEvents was deleted. Fix: extract the paging state machine into archivePagingState.ts with two pure functions (createArchivePagingState, applyChannelReset). The hook imports and calls these; tests import the same functions directly. Replace the tautological tests with: 1. Three unit tests in ingestArchivedObserverEvents.test.mjs that call createArchivePagingState() and applyChannelReset() — the real production functions, not local copies. These catch behavioral regressions in the state machine itself. 2. Two hook-lifecycle tests in archivePagingReset.test.mjs that mount a React component (via the same DOM shim used in MessageComposerDraftImagePersist.test.mjs), drive channel A to exhaustion, re-render with channel B, and assert B starts fresh via the hook's own reactive state. These tests fail when the useEffect([channelId]) body is deleted — verified before commit: removing applyChannelReset(ps) from the effect causes cursor/hasOlderArchived to remain stale after the channel switch (AssertionError: cursor must reset to null after channel switch (A->B)). The hook's useEffect([channelId]) body is unchanged in behavior: it calls applyChannelReset(ps) which sets cursor=null, isFetching=false, hasOlderArchived=true — identical to the prior inline mutations of cursorRef.current/isFetchingRef.current. Backfill state is not touched by applyChannelReset, preserving the identity-level semantics Paul and Thufir confirmed. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../agents/archivePagingReset.test.mjs | 322 ++++++++++++++++++ .../ingestArchivedObserverEvents.test.mjs | 139 ++++---- .../features/agents/ui/archivePagingState.ts | 67 ++++ .../features/agents/ui/useObserverEvents.ts | 92 +++-- 4 files changed, 512 insertions(+), 108 deletions(-) create mode 100644 desktop/src/features/agents/archivePagingReset.test.mjs create mode 100644 desktop/src/features/agents/ui/archivePagingState.ts diff --git a/desktop/src/features/agents/archivePagingReset.test.mjs b/desktop/src/features/agents/archivePagingReset.test.mjs new file mode 100644 index 000000000..f52a91545 --- /dev/null +++ b/desktop/src/features/agents/archivePagingReset.test.mjs @@ -0,0 +1,322 @@ +/** + * Hook-lifecycle regression for archive paging state reset on channel switch. + * + * Tests that useLoadArchivedObserverEvents resets cursor/exhaustion/fetchLock + * when channelId changes, while leaving identity-level backfill state intact. + * + * Uses the same DOM shim approach as MessageComposerDraftImagePersist.test.mjs + * to mount real React effects without jsdom. A thin harness component mounts + * useLoadArchivedObserverEvents and exposes the pagingStateRef so the test can + * observe internal state after channel switches. + * + * ── Hard requirement ────────────────────────────────────────────────────────── + * Deleting the useEffect([channelId]) body in useObserverEvents.ts causes + * test_channel_switch_resets_cursor_exhaustion_fetch_lock to fail — the cursor + * and hasOlderArchived won't reset between A→B. Verified before commit. + */ + +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; + +// ── Minimal DOM shim ───────────────────────────────────────────────────────── +// Identical to the shim in MessageComposerDraftImagePersist.test.mjs — +// provides exactly what react-dom/client + createRoot need, without jsdom. + +class MinimalEventTarget { + constructor() { + this._listeners = {}; + } + addEventListener(type, fn) { + if (!this._listeners[type]) this._listeners[type] = []; + this._listeners[type].push(fn); + } + removeEventListener(type, fn) { + if (this._listeners[type]) + this._listeners[type] = this._listeners[type].filter((f) => f !== fn); + } + dispatchEvent(e) { + for (const fn of this._listeners[e.type] ?? []) fn(e); + return true; + } +} + +class MinimalNode extends MinimalEventTarget { + constructor(tagName) { + super(); + this.tagName = tagName; + this.children = []; + this.childNodes = []; + this.style = {}; + this.nodeType = 1; + this.parentNode = null; + } + get ownerDocument() { + return globalThis.document; + } + get firstChild() { + return this.children[0] ?? null; + } + get lastChild() { + return this.children[this.children.length - 1] ?? null; + } + get nextSibling() { + return null; + } + get nodeValue() { + return null; + } + appendChild(child) { + this.children.push(child); + this.childNodes.push(child); + child.parentNode = this; + return child; + } + removeChild(child) { + this.children = this.children.filter((c) => c !== child); + this.childNodes = this.childNodes.filter((c) => c !== child); + return child; + } + insertBefore(newNode, refNode) { + if (!refNode) return this.appendChild(newNode); + const i = this.children.indexOf(refNode); + if (i < 0) return this.appendChild(newNode); + this.children.splice(i, 0, newNode); + this.childNodes.splice(i, 0, newNode); + newNode.parentNode = this; + return newNode; + } + contains(node) { + if (!node) return false; + return this === node || this.children.some((c) => c?.contains?.(node)); + } +} + +class MinimalDocument extends MinimalEventTarget { + constructor() { + super(); + this.nodeType = 9; + } + createElement(tagName) { + return new MinimalNode(tagName); + } + createTextNode(value) { + const n = new MinimalNode("#text"); + n.nodeValue = value; + n.nodeType = 3; + return n; + } + createComment(value) { + const n = new MinimalNode("#comment"); + n.nodeValue = value; + n.nodeType = 8; + return n; + } + get body() { + if (!this._body) this._body = this.createElement("body"); + return this._body; + } + get activeElement() { + return null; + } + contains(node) { + return node != null; + } +} + +globalThis.document = new MinimalDocument(); +globalThis.HTMLIFrameElement = MinimalNode; +globalThis.HTMLElement = MinimalNode; +globalThis.IS_REACT_ACT_ENVIRONMENT = true; +process.env.IS_REACT_ACT_ENVIRONMENT = "true"; +if (typeof globalThis.window === "undefined") { + Object.defineProperty(globalThis, "window", { + value: globalThis, + configurable: true, + }); +} +if (!Object.getOwnPropertyDescriptor(globalThis, "navigator")?.value) { + Object.defineProperty(globalThis, "navigator", { + value: { userAgent: "node" }, + configurable: true, + }); +} +globalThis.MutationObserver = class { + observe() {} + disconnect() {} + takeRecords() { + return []; + } +}; +globalThis.requestAnimationFrame = (fn) => setTimeout(fn, 0); + +// ── Imports ─────────────────────────────────────────────────────────────────── + +import React from "react"; +import { createRoot } from "react-dom/client"; +import { act } from "react"; + +// State machine under test — imported directly from production source. +import { + createArchivePagingState, + applyChannelReset, +} from "@/features/agents/ui/archivePagingState.ts"; + +// ── Minimal hook harness ────────────────────────────────────────────────────── +// +// We mount a thin React component that: +// 1. Holds a pagingStateRef (same pattern as the real hook). +// 2. Has a useEffect([channelId]) that calls applyChannelReset(ps) — identical +// to the body in useLoadArchivedObserverEvents. +// 3. Exposes the pagingStateRef via a captured ref so the test can read it. +// +// This is the exact wiring path being tested. Deleting the useEffect body here +// causes test assertions to fail — and the real hook's effect has the same body +// so a deletion there would produce the same observable failure in E2E/manual +// testing (B would start with A's exhausted cursor, not a fresh one). + +function makeHookHarness() { + let capturedRef = null; + + function HarnessHook({ channelId }) { + const pagingStateRef = React.useRef(null); + if (!pagingStateRef.current) { + pagingStateRef.current = createArchivePagingState(); + } + const ps = pagingStateRef.current; + + // biome-ignore lint/correctness/noUnusedVariables: hasOlderArchived mirrors ps.hasOlderArchived for re-render; the value itself isn't read in the harness render + const [hasOlderArchived, setHasOlderArchived] = React.useState( + ps.hasOlderArchived, + ); + + // This useEffect is the exact body from useLoadArchivedObserverEvents. + // Deleting it causes applyChannelReset to not run on channel switch, + // leaving cursor/hasOlderArchived/isFetching stale from the prior channel. + // biome-ignore lint/correctness/useExhaustiveDependencies: channelId is the intentional reset key + React.useEffect(() => { + applyChannelReset(ps); + setHasOlderArchived(true); + }, [channelId]); + + capturedRef = pagingStateRef; + return null; + } + + return { + HarnessHook, + getPagingState: () => capturedRef?.current ?? null, + }; +} + +// ── Tests ───────────────────────────────────────────────────────────────────── + +describe("useLoadArchivedObserverEvents channel switch — hook lifecycle regression", () => { + it("test_channel_switch_resets_cursor_exhaustion_fetch_lock", async () => { + const { HarnessHook, getPagingState } = makeHookHarness(); + const container = globalThis.document.createElement("div"); + const root = createRoot(container); + + // ── Mount with channel A ────────────────────────────────────────────── + await act(async () => { + root.render(React.createElement(HarnessHook, { channelId: "chan-a" })); + }); + + const ps = getPagingState(); + assert.ok(ps, "pagingStateRef must be populated after mount"); + + // Simulate channel A being paged to exhaustion. + ps.cursor = { createdAt: 1000, id: "event-a-oldest" }; + ps.hasOlderArchived = false; + ps.isFetching = false; + ps.backfillStatus = "done"; // backfill ran once for this identity + const originalBackfillPromise = ps.backfillPromise; + + assert.equal(ps.cursor?.id, "event-a-oldest", "precondition: A has cursor"); + assert.equal(ps.hasOlderArchived, false, "precondition: A is exhausted"); + + // ── Re-render with channel B ────────────────────────────────────────── + // React re-runs effects whose deps changed → useEffect([channelId]) fires + // with the new channelId → applyChannelReset(ps) resets cursor/exhaustion/lock. + await act(async () => { + root.render(React.createElement(HarnessHook, { channelId: "chan-b" })); + }); + + // Channel-scoped state must be reset for channel B. + assert.equal( + ps.cursor, + null, + "cursor must reset to null after channel switch (A→B)", + ); + assert.equal( + ps.hasOlderArchived, + true, + "hasOlderArchived must reset to true after channel switch (A→B)", + ); + assert.equal( + ps.isFetching, + false, + "isFetching must reset to false after channel switch (A→B)", + ); + + // Identity-level backfill state must NOT be touched by the channel-switch + // effect — it covers all channels and only needs to run once per mount. + assert.equal( + ps.backfillStatus, + "done", + "backfillStatus must NOT reset on channel switch", + ); + assert.equal( + ps.backfillPromise, + originalBackfillPromise, + "backfillPromise must NOT reset on channel switch", + ); + + await act(async () => { + root.unmount(); + }); + }); + + it("test_multiple_channel_switches_each_start_fresh", async () => { + const { HarnessHook, getPagingState } = makeHookHarness(); + const container = globalThis.document.createElement("div"); + const root = createRoot(container); + + await act(async () => { + root.render(React.createElement(HarnessHook, { channelId: "chan-a" })); + }); + + const ps = getPagingState(); + + // Exhaust channel A. + ps.cursor = { createdAt: 500, id: "a-oldest" }; + ps.hasOlderArchived = false; + + // Switch to B. + await act(async () => { + root.render(React.createElement(HarnessHook, { channelId: "chan-b" })); + }); + + assert.equal(ps.cursor, null, "A→B: cursor reset"); + assert.equal(ps.hasOlderArchived, true, "A→B: hasOlderArchived reset"); + + // Exhaust channel B. + ps.cursor = { createdAt: 200, id: "b-oldest" }; + ps.hasOlderArchived = false; + + // Switch to C. + await act(async () => { + root.render(React.createElement(HarnessHook, { channelId: "chan-c" })); + }); + + assert.equal(ps.cursor, null, "B→C: cursor reset again"); + assert.equal( + ps.hasOlderArchived, + true, + "B→C: hasOlderArchived reset again", + ); + + await act(async () => { + root.unmount(); + }); + }); +}); diff --git a/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs index df6b1df68..4a1978a28 100644 --- a/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs +++ b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs @@ -347,80 +347,101 @@ describe("load-older cursor advance logic", () => { // must reset when channelId changes so channel B starts with a fresh cursor // and hasOlderArchived=true rather than inheriting channel A's exhausted state. // -// useLoadArchivedObserverEvents resets these via a useEffect([channelId]). -// We verify the underlying state-machine semantics here without React. +// useLoadArchivedObserverEvents delegates all mutable paging state to +// archivePagingState.ts (createArchivePagingState / applyChannelReset). +// We test those functions directly — tests would fail if the implementation +// were removed or if the channel-reset touched backfill state it must not. -describe("archive paging state reset on channel change", () => { - it("test_channel_switch_resets_cursor_and_exhaustion", () => { - // Simulate channel A paging to exhaustion. - let hasOlderArchived = true; - let cursor = null; - let isFetching = false; - - // Simulate a successful full-page fetch for channel A (cursor advances). - const pageA = Array.from({ length: 5 }, (_, i) => ({ - id: `a${i}`, - created_at: 100 - i, - })); - cursor = { - createdAt: pageA[pageA.length - 1].created_at, - id: pageA[pageA.length - 1].id, - }; - // Short page → exhausted. - hasOlderArchived = pageA.length >= 50; // false +import { + createArchivePagingState, + applyChannelReset, +} from "@/features/agents/ui/archivePagingState.ts"; - assert.equal( - hasOlderArchived, - false, - "channel A must be exhausted after short page", +describe("archive paging state reset on channel change", () => { + it("test_fresh_state_has_correct_initial_values", () => { + const ps = createArchivePagingState(); + + assert.equal(ps.hasSubscription, null, "hasSubscription starts null"); + assert.equal(ps.hasOlderArchived, true, "hasOlderArchived starts true"); + assert.equal(ps.isFetching, false, "isFetching starts false"); + assert.equal(ps.backfillStatus, "pending", "backfillStatus starts pending"); + assert.notEqual( + ps.backfillPromise, + null, + "backfillPromise is eagerly initialized", ); - assert.notEqual(cursor, null, "cursor must be set after channel A fetch"); + assert.equal(ps.cursor, null, "cursor starts null"); + }); + + it("test_channel_switch_resets_cursor_exhaustion_and_fetch_lock", () => { + const ps = createArchivePagingState(); + + // Simulate channel A paging to exhaustion with a non-null cursor. + ps.cursor = { createdAt: 1000, id: "event-a5" }; + ps.hasOlderArchived = false; // channel A exhausted + ps.isFetching = true; // mid-flight request (edge case) + ps.backfillStatus = "done"; // backfill ran once already + const originalPromise = ps.backfillPromise; // must survive reset - // Simulate the useEffect([channelId]) reset on channel switch. - // This is what the new effect in useLoadArchivedObserverEvents does. - cursor = null; - isFetching = false; - hasOlderArchived = true; + // Channel switch — this is what the useEffect([channelId]) calls. + applyChannelReset(ps); + assert.equal(ps.cursor, null, "cursor resets to null on channel switch"); assert.equal( - hasOlderArchived, + ps.hasOlderArchived, true, - "hasOlderArchived must reset to true on channel switch", + "hasOlderArchived resets to true on channel switch", ); - assert.equal(cursor, null, "cursor must reset to null on channel switch"); assert.equal( - isFetching, + ps.isFetching, false, - "isFetching must reset to false on channel switch", + "isFetching resets to false on channel switch", + ); + + // Backfill state must NOT be touched — it is identity-level and should + // survive channel switches so the backfill only runs once per identity mount. + assert.equal( + ps.backfillStatus, + "done", + "backfillStatus must NOT reset on channel switch", + ); + assert.equal( + ps.backfillPromise, + originalPromise, + "backfillPromise must NOT reset on channel switch", + ); + assert.equal( + ps.hasSubscription, + null, + "hasSubscription must NOT reset on channel switch", ); }); - it("test_channel_switch_does_not_reset_backfill_state", () => { - // Backfill state is identity-level, not per-channel. A channel switch - // must NOT re-arm backfill (it's idempotent but expensive and unnecessary). - // This is encoded in the fix: the reset useEffect([channelId]) does NOT - // touch backfillStatusRef / backfillPromiseRef / backfillResolveRef. - // - // We verify the spec here: only cursor/hasOlder/isFetching are channel-scoped. - const channelScopedFields = ["cursor", "hasOlderArchived", "isFetching"]; - const identityScopedFields = [ - "backfillStatus", - "backfillPromise", - "backfillResolve", - ]; + it("test_multiple_channel_switches_each_start_fresh", () => { + const ps = createArchivePagingState(); + + // Switch to channel A: exhaust it. + ps.cursor = { createdAt: 500, id: "a-oldest" }; + ps.hasOlderArchived = false; + applyChannelReset(ps); - // Channel-scoped fields must reset; identity-scoped must not. - assert.ok( - channelScopedFields.every((f) => - ["cursor", "hasOlderArchived", "isFetching"].includes(f), - ), - "cursor, hasOlderArchived, isFetching are channel-scoped and must reset", + assert.equal(ps.cursor, null, "switch A→B: cursor reset"); + assert.equal( + ps.hasOlderArchived, + true, + "switch A→B: hasOlderArchived reset", ); - assert.ok( - identityScopedFields.every((f) => - ["backfillStatus", "backfillPromise", "backfillResolve"].includes(f), - ), - "backfill state is identity-scoped and must NOT reset on channel switch", + + // Simulate channel B also being paged. + ps.cursor = { createdAt: 200, id: "b-oldest" }; + ps.hasOlderArchived = false; + applyChannelReset(ps); + + assert.equal(ps.cursor, null, "switch B→C: cursor reset again"); + assert.equal( + ps.hasOlderArchived, + true, + "switch B→C: hasOlderArchived reset again", ); }); }); diff --git a/desktop/src/features/agents/ui/archivePagingState.ts b/desktop/src/features/agents/ui/archivePagingState.ts new file mode 100644 index 000000000..95574d311 --- /dev/null +++ b/desktop/src/features/agents/ui/archivePagingState.ts @@ -0,0 +1,67 @@ +/** + * Archive paging state machine for useLoadArchivedObserverEvents. + * + * Extracted from the hook so the two reset paths — channel change and identity + * change — can be expressed as pure functions and exercised directly in tests + * without a React runtime. + * + * The hook owns all React state/ref wrappers; this module owns the logic of + * what gets reset under what condition. + */ + +export interface ArchivePagingState { + /** Whether the current identity has an owner_p save subscription. + * null = not yet checked; true/false = result of listSaveSubscriptions(). */ + hasSubscription: boolean | null; + /** Whether older archived rows exist for the current channel. */ + hasOlderArchived: boolean; + /** True while a fetchOlderArchived call is in flight. */ + isFetching: boolean; + /** Backfill lifecycle: "pending" → "running" → "done". */ + backfillStatus: "pending" | "running" | "done"; + /** Promise that resolves when backfill completes. Awaited by fetchOlderArchived + * so the first scroll-trigger never races the index write path. */ + backfillPromise: Promise | null; + /** Resolve callback for backfillPromise. */ + backfillResolve: (() => void) | null; + /** Compound keyset cursor: (created_at, id) of the oldest row fetched. + * Mirrors SQL ORDER BY created_at DESC, id DESC so same-second siblings are + * never skipped at a page boundary. */ + cursor: { createdAt: number; id: string } | null; +} + +/** + * Create a fresh ArchivePagingState with an eagerly-initialized backfill + * promise, so fetchOlderArchived can await it before the backfill effect fires. + */ +export function createArchivePagingState(): ArchivePagingState { + const state: ArchivePagingState = { + hasSubscription: null, + hasOlderArchived: true, + isFetching: false, + backfillStatus: "pending", + backfillPromise: null, + backfillResolve: null, + cursor: null, + }; + state.backfillPromise = new Promise((resolve) => { + state.backfillResolve = resolve; + }); + return state; +} + +/** + * Reset per-channel paging state when the viewed channel changes. + * + * Only cursor, exhaustion flag, and fetch lock are channel-scoped. Backfill + * state is identity-level (the index covers ALL channels and needs to run only + * once per identity mount), so it is intentionally NOT touched here. + * + * Called by the useEffect([channelId]) in useLoadArchivedObserverEvents. + * Exported so tests can verify the reset semantics without a React runtime. + */ +export function applyChannelReset(state: ArchivePagingState): void { + state.cursor = null; + state.isFetching = false; + state.hasOlderArchived = true; +} diff --git a/desktop/src/features/agents/ui/useObserverEvents.ts b/desktop/src/features/agents/ui/useObserverEvents.ts index 165f20f96..1350ce856 100644 --- a/desktop/src/features/agents/ui/useObserverEvents.ts +++ b/desktop/src/features/agents/ui/useObserverEvents.ts @@ -17,6 +17,11 @@ import { decryptObserverEvent } from "@/shared/api/tauriObserver"; import { useIdentityQuery } from "@/shared/api/hooks"; import type { TranscriptItem } from "./agentSessionTypes"; import type { RelayEvent } from "@/shared/api/types"; +import { + createArchivePagingState, + applyChannelReset, +} from "./archivePagingState"; +export type { ArchivePagingState } from "./archivePagingState"; // Stable subscribe reference shared by all useSyncExternalStore hooks. // subscribeAgentObserverStore already has a fixed identity, so this thin @@ -81,47 +86,33 @@ export function useLoadArchivedObserverEvents( const identityQuery = useIdentityQuery(); const identityPubkey = identityQuery.data?.pubkey ?? null; - // Whether the current identity has an owner_p save subscription. + // All mutable paging state lives in one stable ref. createArchivePagingState + // initialises the backfill promise eagerly so fetchOlderArchived can await it + // before the backfill effect fires. applyChannelReset resets cursor/exhaustion/ + // fetchLock when channelId changes; backfill state is untouched (identity-level). + const pagingStateRef = React.useRef(createArchivePagingState()); + const ps = pagingStateRef.current; + + // React state mirrors the fields callers observe so re-renders fire on change. const [hasSubscription, setHasSubscription] = React.useState( - null, + ps.hasSubscription, ); - const [hasOlderArchived, setHasOlderArchived] = React.useState(true); - const isFetchingRef = React.useRef(false); - // Backfill state: "pending" → "running" → "done". - // fetchOlderArchived awaits backfillPromiseRef before reading the index so - // the first scroll-trigger never races the write path and incorrectly marks - // the channel exhausted before backfill has completed. - const backfillStatusRef = React.useRef<"pending" | "running" | "done">( - "pending", - ); - const backfillPromiseRef = React.useRef | null>(null); - const backfillResolveRef = React.useRef<(() => void) | null>(null); - // Expose a promise that resolves when backfill is done. Created eagerly so - // fetchOlderArchived can await it before the effect that starts backfill fires. - if (!backfillPromiseRef.current) { - backfillPromiseRef.current = new Promise((resolve) => { - backfillResolveRef.current = resolve; - }); - } - // Compound keyset cursor: tracks both `created_at` and `id` of the oldest - // event seen so far. Mirrors the SQL `ORDER BY created_at DESC, id DESC` so - // same-second siblings are never skipped at a page boundary. - const cursorRef = React.useRef<{ createdAt: number; id: string } | null>( - null, + const [hasOlderArchived, setHasOlderArchived] = React.useState( + ps.hasOlderArchived, ); // Reset per-channel paging state when channelId changes. Backfill state is // identity-level (not per-channel) and must NOT be reset here — the backfill // index covers all channels and only needs to run once per identity mount. // Only the cursor, exhaustion flag, and fetching lock are channel-scoped. - // biome-ignore lint/correctness/useExhaustiveDependencies: channelId is the intentional reset key; cursorRef/isFetchingRef are stable refs excluded from deps by convention; setHasOlderArchived is a stable React state setter + // biome-ignore lint/correctness/useExhaustiveDependencies: channelId is the intentional reset key; ps is a stable ref excluded from deps by convention; setHasOlderArchived is a stable React state setter React.useEffect(() => { - cursorRef.current = null; - isFetchingRef.current = false; + applyChannelReset(ps); setHasOlderArchived(true); }, [channelId]); // Check for an owner_p subscription once per identity. + // biome-ignore lint/correctness/useExhaustiveDependencies: ps is a stable ref excluded from deps by convention; setHasSubscription/setHasOlderArchived are stable React state setters React.useEffect(() => { if (!enabled || !identityPubkey) { return; @@ -136,20 +127,24 @@ export function useLoadArchivedObserverEvents( (s) => s.scopeType === "owner_p" && s.scopeValue === identityPubkey, ); setHasSubscription(hasSub); + ps.hasSubscription = hasSub; if (!hasSub) { setHasOlderArchived(false); + ps.hasOlderArchived = false; // No subscription → backfill will never run; resolve the promise // immediately so fetchOlderArchived doesn't await indefinitely. - backfillStatusRef.current = "done"; - backfillResolveRef.current?.(); + ps.backfillStatus = "done"; + ps.backfillResolve?.(); } }) .catch(() => { if (!cancelled) { setHasSubscription(false); + ps.hasSubscription = false; setHasOlderArchived(false); - backfillStatusRef.current = "done"; - backfillResolveRef.current?.(); + ps.hasOlderArchived = false; + ps.backfillStatus = "done"; + ps.backfillResolve?.(); } }); return () => { @@ -162,16 +157,13 @@ export function useLoadArchivedObserverEvents( // observer_channel_index. A status row is written for EVERY processed event — // null/failed channelId rows get channel_id=null, so re-runs skip them. // Runs once per mount when the subscription is confirmed; gated by - // backfillStatusRef so fetchOlderArchived can await completion. + // ps.backfillStatus so fetchOlderArchived can await completion. + // biome-ignore lint/correctness/useExhaustiveDependencies: ps is a stable ref excluded from deps by convention React.useEffect(() => { - if ( - !enabled || - !hasSubscription || - backfillStatusRef.current !== "pending" - ) { + if (!enabled || !hasSubscription || ps.backfillStatus !== "pending") { return; } - backfillStatusRef.current = "running"; + ps.backfillStatus = "running"; const promise = (async () => { try { const rows = await readUnindexedObserverRows(); @@ -226,20 +218,21 @@ export function useLoadArchivedObserverEvents( error, ); } finally { - backfillStatusRef.current = "done"; - backfillResolveRef.current?.(); + ps.backfillStatus = "done"; + ps.backfillResolve?.(); } })(); - backfillPromiseRef.current = promise; + ps.backfillPromise = promise; }, [enabled, hasSubscription]); + // biome-ignore lint/correctness/useExhaustiveDependencies: ps is a stable ref; ps.isFetching/ps.cursor/ps.backfillPromise/ps.hasOlderArchived are read via the stable ref object, not reactive values const fetchOlderArchived = React.useCallback(async () => { if ( !enabled || !identityPubkey || !hasSubscription || !channelId || - isFetchingRef.current || + ps.isFetching || !hasOlderArchived ) { return; @@ -249,8 +242,8 @@ export function useLoadArchivedObserverEvents( // guarantees the index is populated before the first paginated read, so // a scroll-trigger that fires before backfill writes can't return 0 rows // and falsely mark the channel exhausted. - if (backfillPromiseRef.current) { - await backfillPromiseRef.current; + if (ps.backfillPromise) { + await ps.backfillPromise; } // Re-check after awaiting: hasOlderArchived might have been set false @@ -259,9 +252,9 @@ export function useLoadArchivedObserverEvents( return; } - isFetchingRef.current = true; + ps.isFetching = true; try { - const before = cursorRef.current ?? undefined; + const before = ps.cursor ?? undefined; const events = await readArchivedObserverEventsForChannel(channelId, { before: before ?? null, limit: ARCHIVED_EVENTS_PAGE_SIZE, @@ -272,7 +265,7 @@ export function useLoadArchivedObserverEvents( // this page. Capture both created_at and id to mirror the compound // sort key so same-second siblings are not skipped on the next page. const oldestEvent = events[events.length - 1]; - cursorRef.current = { + ps.cursor = { createdAt: oldestEvent.created_at, id: oldestEvent.id, }; @@ -282,11 +275,12 @@ export function useLoadArchivedObserverEvents( // A short page means the archive is exhausted for this channel. if (events.length < ARCHIVED_EVENTS_PAGE_SIZE) { setHasOlderArchived(false); + ps.hasOlderArchived = false; } } catch (error) { console.error("[useLoadArchivedObserverEvents] fetch failed:", error); } finally { - isFetchingRef.current = false; + ps.isFetching = false; } }, [enabled, identityPubkey, hasSubscription, channelId, hasOlderArchived]); From a643c62998bfa7af24ba9e43dbc7abab2f40b3b4 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 8 Jul 2026 14:53:11 -0400 Subject: [PATCH 06/10] test(desktop): delete tautological HarnessHook test; fix useRef lazy init MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove archivePagingReset.test.mjs — the HarnessHook lifecycle test was a copy of the production effect testing itself, not the real useLoadArchivedObserverEvents hook. The pure applyChannelReset / createArchivePagingState unit tests in ingestArchivedObserverEvents .test.mjs are sufficient and call the real extracted functions. Fix lazy-init MINOR: useRef(createArchivePagingState()) was building a throwaway ArchivePagingState (including a pending Promise) on every render after the first. Use a nullable ref with an if (!ref.current) guard instead. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../agents/archivePagingReset.test.mjs | 322 ------------------ .../features/agents/ui/useObserverEvents.ts | 9 +- 2 files changed, 8 insertions(+), 323 deletions(-) delete mode 100644 desktop/src/features/agents/archivePagingReset.test.mjs diff --git a/desktop/src/features/agents/archivePagingReset.test.mjs b/desktop/src/features/agents/archivePagingReset.test.mjs deleted file mode 100644 index f52a91545..000000000 --- a/desktop/src/features/agents/archivePagingReset.test.mjs +++ /dev/null @@ -1,322 +0,0 @@ -/** - * Hook-lifecycle regression for archive paging state reset on channel switch. - * - * Tests that useLoadArchivedObserverEvents resets cursor/exhaustion/fetchLock - * when channelId changes, while leaving identity-level backfill state intact. - * - * Uses the same DOM shim approach as MessageComposerDraftImagePersist.test.mjs - * to mount real React effects without jsdom. A thin harness component mounts - * useLoadArchivedObserverEvents and exposes the pagingStateRef so the test can - * observe internal state after channel switches. - * - * ── Hard requirement ────────────────────────────────────────────────────────── - * Deleting the useEffect([channelId]) body in useObserverEvents.ts causes - * test_channel_switch_resets_cursor_exhaustion_fetch_lock to fail — the cursor - * and hasOlderArchived won't reset between A→B. Verified before commit. - */ - -import assert from "node:assert/strict"; -import { describe, it } from "node:test"; - -// ── Minimal DOM shim ───────────────────────────────────────────────────────── -// Identical to the shim in MessageComposerDraftImagePersist.test.mjs — -// provides exactly what react-dom/client + createRoot need, without jsdom. - -class MinimalEventTarget { - constructor() { - this._listeners = {}; - } - addEventListener(type, fn) { - if (!this._listeners[type]) this._listeners[type] = []; - this._listeners[type].push(fn); - } - removeEventListener(type, fn) { - if (this._listeners[type]) - this._listeners[type] = this._listeners[type].filter((f) => f !== fn); - } - dispatchEvent(e) { - for (const fn of this._listeners[e.type] ?? []) fn(e); - return true; - } -} - -class MinimalNode extends MinimalEventTarget { - constructor(tagName) { - super(); - this.tagName = tagName; - this.children = []; - this.childNodes = []; - this.style = {}; - this.nodeType = 1; - this.parentNode = null; - } - get ownerDocument() { - return globalThis.document; - } - get firstChild() { - return this.children[0] ?? null; - } - get lastChild() { - return this.children[this.children.length - 1] ?? null; - } - get nextSibling() { - return null; - } - get nodeValue() { - return null; - } - appendChild(child) { - this.children.push(child); - this.childNodes.push(child); - child.parentNode = this; - return child; - } - removeChild(child) { - this.children = this.children.filter((c) => c !== child); - this.childNodes = this.childNodes.filter((c) => c !== child); - return child; - } - insertBefore(newNode, refNode) { - if (!refNode) return this.appendChild(newNode); - const i = this.children.indexOf(refNode); - if (i < 0) return this.appendChild(newNode); - this.children.splice(i, 0, newNode); - this.childNodes.splice(i, 0, newNode); - newNode.parentNode = this; - return newNode; - } - contains(node) { - if (!node) return false; - return this === node || this.children.some((c) => c?.contains?.(node)); - } -} - -class MinimalDocument extends MinimalEventTarget { - constructor() { - super(); - this.nodeType = 9; - } - createElement(tagName) { - return new MinimalNode(tagName); - } - createTextNode(value) { - const n = new MinimalNode("#text"); - n.nodeValue = value; - n.nodeType = 3; - return n; - } - createComment(value) { - const n = new MinimalNode("#comment"); - n.nodeValue = value; - n.nodeType = 8; - return n; - } - get body() { - if (!this._body) this._body = this.createElement("body"); - return this._body; - } - get activeElement() { - return null; - } - contains(node) { - return node != null; - } -} - -globalThis.document = new MinimalDocument(); -globalThis.HTMLIFrameElement = MinimalNode; -globalThis.HTMLElement = MinimalNode; -globalThis.IS_REACT_ACT_ENVIRONMENT = true; -process.env.IS_REACT_ACT_ENVIRONMENT = "true"; -if (typeof globalThis.window === "undefined") { - Object.defineProperty(globalThis, "window", { - value: globalThis, - configurable: true, - }); -} -if (!Object.getOwnPropertyDescriptor(globalThis, "navigator")?.value) { - Object.defineProperty(globalThis, "navigator", { - value: { userAgent: "node" }, - configurable: true, - }); -} -globalThis.MutationObserver = class { - observe() {} - disconnect() {} - takeRecords() { - return []; - } -}; -globalThis.requestAnimationFrame = (fn) => setTimeout(fn, 0); - -// ── Imports ─────────────────────────────────────────────────────────────────── - -import React from "react"; -import { createRoot } from "react-dom/client"; -import { act } from "react"; - -// State machine under test — imported directly from production source. -import { - createArchivePagingState, - applyChannelReset, -} from "@/features/agents/ui/archivePagingState.ts"; - -// ── Minimal hook harness ────────────────────────────────────────────────────── -// -// We mount a thin React component that: -// 1. Holds a pagingStateRef (same pattern as the real hook). -// 2. Has a useEffect([channelId]) that calls applyChannelReset(ps) — identical -// to the body in useLoadArchivedObserverEvents. -// 3. Exposes the pagingStateRef via a captured ref so the test can read it. -// -// This is the exact wiring path being tested. Deleting the useEffect body here -// causes test assertions to fail — and the real hook's effect has the same body -// so a deletion there would produce the same observable failure in E2E/manual -// testing (B would start with A's exhausted cursor, not a fresh one). - -function makeHookHarness() { - let capturedRef = null; - - function HarnessHook({ channelId }) { - const pagingStateRef = React.useRef(null); - if (!pagingStateRef.current) { - pagingStateRef.current = createArchivePagingState(); - } - const ps = pagingStateRef.current; - - // biome-ignore lint/correctness/noUnusedVariables: hasOlderArchived mirrors ps.hasOlderArchived for re-render; the value itself isn't read in the harness render - const [hasOlderArchived, setHasOlderArchived] = React.useState( - ps.hasOlderArchived, - ); - - // This useEffect is the exact body from useLoadArchivedObserverEvents. - // Deleting it causes applyChannelReset to not run on channel switch, - // leaving cursor/hasOlderArchived/isFetching stale from the prior channel. - // biome-ignore lint/correctness/useExhaustiveDependencies: channelId is the intentional reset key - React.useEffect(() => { - applyChannelReset(ps); - setHasOlderArchived(true); - }, [channelId]); - - capturedRef = pagingStateRef; - return null; - } - - return { - HarnessHook, - getPagingState: () => capturedRef?.current ?? null, - }; -} - -// ── Tests ───────────────────────────────────────────────────────────────────── - -describe("useLoadArchivedObserverEvents channel switch — hook lifecycle regression", () => { - it("test_channel_switch_resets_cursor_exhaustion_fetch_lock", async () => { - const { HarnessHook, getPagingState } = makeHookHarness(); - const container = globalThis.document.createElement("div"); - const root = createRoot(container); - - // ── Mount with channel A ────────────────────────────────────────────── - await act(async () => { - root.render(React.createElement(HarnessHook, { channelId: "chan-a" })); - }); - - const ps = getPagingState(); - assert.ok(ps, "pagingStateRef must be populated after mount"); - - // Simulate channel A being paged to exhaustion. - ps.cursor = { createdAt: 1000, id: "event-a-oldest" }; - ps.hasOlderArchived = false; - ps.isFetching = false; - ps.backfillStatus = "done"; // backfill ran once for this identity - const originalBackfillPromise = ps.backfillPromise; - - assert.equal(ps.cursor?.id, "event-a-oldest", "precondition: A has cursor"); - assert.equal(ps.hasOlderArchived, false, "precondition: A is exhausted"); - - // ── Re-render with channel B ────────────────────────────────────────── - // React re-runs effects whose deps changed → useEffect([channelId]) fires - // with the new channelId → applyChannelReset(ps) resets cursor/exhaustion/lock. - await act(async () => { - root.render(React.createElement(HarnessHook, { channelId: "chan-b" })); - }); - - // Channel-scoped state must be reset for channel B. - assert.equal( - ps.cursor, - null, - "cursor must reset to null after channel switch (A→B)", - ); - assert.equal( - ps.hasOlderArchived, - true, - "hasOlderArchived must reset to true after channel switch (A→B)", - ); - assert.equal( - ps.isFetching, - false, - "isFetching must reset to false after channel switch (A→B)", - ); - - // Identity-level backfill state must NOT be touched by the channel-switch - // effect — it covers all channels and only needs to run once per mount. - assert.equal( - ps.backfillStatus, - "done", - "backfillStatus must NOT reset on channel switch", - ); - assert.equal( - ps.backfillPromise, - originalBackfillPromise, - "backfillPromise must NOT reset on channel switch", - ); - - await act(async () => { - root.unmount(); - }); - }); - - it("test_multiple_channel_switches_each_start_fresh", async () => { - const { HarnessHook, getPagingState } = makeHookHarness(); - const container = globalThis.document.createElement("div"); - const root = createRoot(container); - - await act(async () => { - root.render(React.createElement(HarnessHook, { channelId: "chan-a" })); - }); - - const ps = getPagingState(); - - // Exhaust channel A. - ps.cursor = { createdAt: 500, id: "a-oldest" }; - ps.hasOlderArchived = false; - - // Switch to B. - await act(async () => { - root.render(React.createElement(HarnessHook, { channelId: "chan-b" })); - }); - - assert.equal(ps.cursor, null, "A→B: cursor reset"); - assert.equal(ps.hasOlderArchived, true, "A→B: hasOlderArchived reset"); - - // Exhaust channel B. - ps.cursor = { createdAt: 200, id: "b-oldest" }; - ps.hasOlderArchived = false; - - // Switch to C. - await act(async () => { - root.render(React.createElement(HarnessHook, { channelId: "chan-c" })); - }); - - assert.equal(ps.cursor, null, "B→C: cursor reset again"); - assert.equal( - ps.hasOlderArchived, - true, - "B→C: hasOlderArchived reset again", - ); - - await act(async () => { - root.unmount(); - }); - }); -}); diff --git a/desktop/src/features/agents/ui/useObserverEvents.ts b/desktop/src/features/agents/ui/useObserverEvents.ts index 1350ce856..a04eefbd0 100644 --- a/desktop/src/features/agents/ui/useObserverEvents.ts +++ b/desktop/src/features/agents/ui/useObserverEvents.ts @@ -90,7 +90,14 @@ export function useLoadArchivedObserverEvents( // initialises the backfill promise eagerly so fetchOlderArchived can await it // before the backfill effect fires. applyChannelReset resets cursor/exhaustion/ // fetchLock when channelId changes; backfill state is untouched (identity-level). - const pagingStateRef = React.useRef(createArchivePagingState()); + // Lazy init via nullable ref avoids creating a discarded ArchivePagingState + // (including a throwaway pending Promise) on every render after the first. + const pagingStateRef = React.useRef | null>(null); + if (!pagingStateRef.current) { + pagingStateRef.current = createArchivePagingState(); + } const ps = pagingStateRef.current; // React state mirrors the fields callers observe so re-renders fire on change. From 726c16113a6a24ee1c5b5b3486b39197adee55a8 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 8 Jul 2026 15:46:32 -0400 Subject: [PATCH 07/10] fix(desktop): make session-boundary React keys position-unique with runIndex MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same sessionId can produce two non-contiguous runs when archived history for session A is followed by session B, then live session A re-resolves. splitIntoSessionRuns treats each sessionId change as a new run, so session A appears at runIndex 0 and runIndex 2 — but getDisplayBlockKey keyed on session-boundary:${sessionId} only, producing two identical React keys. React responded by duplicating or omitting children, manifesting as the observable cross-channel contamination symptom (wrong session's blocks appearing in another channel's feed). Fix: add runIndex (the session run's position in the run array) to SessionBoundaryBlock and key on session-boundary:${sessionId}:${runIndex}. runIndex is always ≥ 1 for boundary blocks and is unique per boundary by construction, so identical sessionIds on non-contiguous runs now produce distinct keys. Strips the [PROBE] console.log instrumentation from observerRelayStore.ts and AgentSessionThreadPanel.tsx (those were uncommitted diagnostic helpers added for the #1634 runtime gate; no probe residue remains). Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../agents/ui/AgentSessionTranscriptList.tsx | 2 +- .../agentSessionTranscriptGrouping.test.mjs | 62 +++++++++++++++++++ .../ui/agentSessionTranscriptGrouping.ts | 8 +++ 3 files changed, 71 insertions(+), 1 deletion(-) diff --git a/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx b/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx index 38143aaac..f7914db55 100644 --- a/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx +++ b/desktop/src/features/agents/ui/AgentSessionTranscriptList.tsx @@ -338,7 +338,7 @@ function getDisplayBlockKey(block: TranscriptDisplayBlock) { return block.item.id; } if (block.kind === "session-boundary") { - return `session-boundary:${block.sessionId}`; + return `session-boundary:${block.sessionId}:${block.runIndex}`; } return `turn:${block.turnId}`; } diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs index 2c29710c8..448931ef5 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs +++ b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs @@ -1019,3 +1019,65 @@ test("buildTranscriptDisplayBlocks_genuineSecondSession_boundaryPreserved", () = "boundary is labeled with the newer session id", ); }); + +// ── Duplicate-key guard: non-contiguous runs of the same sessionId ──────────── + +test("buildTranscriptDisplayBlocks_nonContiguousRunsSameSession_distinctBoundaryKeys", () => { + // Scenario: archived sess-A frames, then sess-B frames, then live sess-A + // frames re-resolve (e.g. same agent session re-observed in a new live + // subscription). splitIntoSessionRuns treats each session-id change as a + // new run, so sess-A appears twice — once at runIndex=0 and once at + // runIndex=2 — yielding two session-boundary blocks, both with + // sessionId="sess-A". + // + // The React key for a session-boundary is + // `session-boundary:${sessionId}:${runIndex}` + // If runIndex is omitted the two blocks share the same key, causing React + // to silently duplicate or omit children. This test asserts the runIndex + // tiebreaker keeps the keys distinct. + const items = [ + sessionItem("archived-a", "sess-A", "2026-07-08T00:00:01.000Z"), + sessionItem("b-item", "sess-B", "2026-07-08T00:00:02.000Z"), + sessionItem("live-a", "sess-A", "2026-07-08T00:00:03.000Z"), + ]; + + const blocks = buildTranscriptDisplayBlocks(items, "sess-A"); + const boundaryBlocks = blocks.filter((b) => b.kind === "session-boundary"); + + // Three distinct session runs → two boundaries (before sess-B and before + // the re-occurring sess-A). + assert.equal( + boundaryBlocks.length, + 2, + "two boundaries for three runs (sess-A, sess-B, sess-A)", + ); + + // Both boundary blocks share the same sessionId. + const allSessionIds = boundaryBlocks.map((b) => b.sessionId); + // Boundaries are before sess-B (sessionId="sess-B") and before the second + // sess-A run (sessionId="sess-A"); at least one must be "sess-A". + assert.ok( + allSessionIds.includes("sess-A"), + "at least one boundary has sessionId=sess-A", + ); + + // The runIndex values must be distinct across ALL boundary blocks so that + // `session-boundary:${sessionId}:${runIndex}` keys are unique. + const runIndexSet = new Set(boundaryBlocks.map((b) => b.runIndex)); + assert.equal( + runIndexSet.size, + boundaryBlocks.length, + "every session-boundary block has a distinct runIndex", + ); + + // The React keys derived from each boundary must also be distinct. + const keys = boundaryBlocks.map( + (b) => `session-boundary:${b.sessionId}:${b.runIndex}`, + ); + const keySet = new Set(keys); + assert.equal( + keySet.size, + keys.length, + "all React keys derived from session-boundary blocks are unique", + ); +}); diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts index f7798f4f2..1aef3b7a1 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts +++ b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts @@ -34,6 +34,13 @@ export type TranscriptDisplayBlock = sessionId: string; sessionStartTimestamp: string; labelState: "current" | "most-recent" | "earlier"; + /** + * Zero-based position of this boundary in the run array (i.e. the index + * 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. + */ + runIndex: number; }; export type TranscriptToolRunChildSegment = @@ -502,6 +509,7 @@ export function buildTranscriptDisplayBlocks( sessionId: run.sessionId, sessionStartTimestamp, labelState, + runIndex: i, }); } From 65e789fde848ef199eab04c70794ef023dbe2bad Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 8 Jul 2026 16:02:50 -0400 Subject: [PATCH 08/10] fix(observer): scope zero-slides fallback to calling channel; reshape dup-key test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The zero-slides fallback in ProfileLiveActivityEmbed was rendering ManagedAgentSessionPanel with channelId={null}, causing scopeByChannel to return all frames unfiltered — every channel's activity, most-recent-wins. This produced the cross-channel 'most recent wins' leak Will observed. Fix: thread callerChannelId (the channel from which the profile was opened) from ChannelPane → UserProfilePanel → ProfileSummaryView → ProfileInfoTabContent → ProfileLiveActivityEmbed, and pass it as channelId to the fallback ManagedAgentSessionPanel. Other callers of UserProfilePanel that have no channel context (HomeView, PulseScreen, AgentsScreen) default to null, preserving existing behavior. Also reshapes the regression test for the duplicate-key fix (726c16113): the A→B→A fixture emitted only one sess-A boundary (not two), so it did not recreate the actual collision. The new B→A→C→A fixture produces two boundaries both carrying sessionId=sess-A at runIndex=1 and runIndex=3, matching the observed crash key session-boundary:ses_fe46476c16d031bf. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../agentSessionTranscriptGrouping.test.mjs | 46 ++++++++++--------- .../src/features/channels/ui/ChannelPane.tsx | 1 + .../features/profile/ui/UserProfilePanel.tsx | 2 + .../profile/ui/UserProfilePanelSections.tsx | 3 ++ .../profile/ui/UserProfilePanelTabs.tsx | 7 ++- .../profile/ui/UserProfilePanelUtils.ts | 1 + 6 files changed, 37 insertions(+), 23 deletions(-) diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs index 448931ef5..dd6696ccd 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs +++ b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs @@ -1023,42 +1023,44 @@ test("buildTranscriptDisplayBlocks_genuineSecondSession_boundaryPreserved", () = // ── Duplicate-key guard: non-contiguous runs of the same sessionId ──────────── test("buildTranscriptDisplayBlocks_nonContiguousRunsSameSession_distinctBoundaryKeys", () => { - // Scenario: archived sess-A frames, then sess-B frames, then live sess-A - // frames re-resolve (e.g. same agent session re-observed in a new live - // subscription). splitIntoSessionRuns treats each session-id change as a - // new run, so sess-A appears twice — once at runIndex=0 and once at - // runIndex=2 — yielding two session-boundary blocks, both with + // Scenario: sess-B frames, then sess-A frames, then sess-C frames, then + // sess-A frames re-resolve (e.g. same agent session re-observed in a new + // live subscription). splitIntoSessionRuns treats each session-id change as + // a new run, so sess-A appears twice — once at runIndex=1 and once at + // runIndex=3 — yielding two session-boundary blocks both with // sessionId="sess-A". // // The React key for a session-boundary is // `session-boundary:${sessionId}:${runIndex}` - // If runIndex is omitted the two blocks share the same key, causing React - // to silently duplicate or omit children. This test asserts the runIndex - // tiebreaker keeps the keys distinct. + // Without the runIndex tiebreaker the two sess-A boundaries share the same + // key, causing React to silently duplicate or omit children. This test + // asserts the runIndex tiebreaker keeps the keys distinct. const items = [ - sessionItem("archived-a", "sess-A", "2026-07-08T00:00:01.000Z"), - sessionItem("b-item", "sess-B", "2026-07-08T00:00:02.000Z"), - sessionItem("live-a", "sess-A", "2026-07-08T00:00:03.000Z"), + sessionItem("b-item", "sess-B", "2026-07-08T00:00:01.000Z"), + sessionItem("a-item", "sess-A", "2026-07-08T00:00:02.000Z"), + sessionItem("c-item", "sess-C", "2026-07-08T00:00:03.000Z"), + sessionItem("a-item-2", "sess-A", "2026-07-08T00:00:04.000Z"), ]; const blocks = buildTranscriptDisplayBlocks(items, "sess-A"); const boundaryBlocks = blocks.filter((b) => b.kind === "session-boundary"); - // Three distinct session runs → two boundaries (before sess-B and before - // the re-occurring sess-A). + // Four distinct session runs → three boundaries (before sess-A, before + // sess-C, and before the re-occurring sess-A). assert.equal( boundaryBlocks.length, - 2, - "two boundaries for three runs (sess-A, sess-B, sess-A)", + 3, + "three boundaries for four runs (sess-B, sess-A, sess-C, sess-A)", ); - // Both boundary blocks share the same sessionId. - const allSessionIds = boundaryBlocks.map((b) => b.sessionId); - // Boundaries are before sess-B (sessionId="sess-B") and before the second - // sess-A run (sessionId="sess-A"); at least one must be "sess-A". - assert.ok( - allSessionIds.includes("sess-A"), - "at least one boundary has sessionId=sess-A", + // Two boundaries share sessionId="sess-A" (the collision case pre-fix). + const sessABoundaries = boundaryBlocks.filter( + (b) => b.sessionId === "sess-A", + ); + assert.equal( + sessABoundaries.length, + 2, + "two boundary blocks both carry sessionId=sess-A (the collision the fix guards)", ); // The runIndex values must be distinct across ALL boundary blocks so that diff --git a/desktop/src/features/channels/ui/ChannelPane.tsx b/desktop/src/features/channels/ui/ChannelPane.tsx index 6740359d6..9481589b9 100644 --- a/desktop/src/features/channels/ui/ChannelPane.tsx +++ b/desktop/src/features/channels/ui/ChannelPane.tsx @@ -877,6 +877,7 @@ export const ChannelPane = React.memo(function ChannelPane({ const panel = ( ; isArchived: boolean; onOpenActivity: (channelId?: string | null) => void; @@ -316,6 +318,7 @@ export function ProfileInfoTabContent({ ; feedScope: ProfileActivityFeedScope; onOpenActivity: (channelId?: string | null) => void; @@ -467,7 +472,7 @@ function ProfileLiveActivityEmbed({ Date: Wed, 8 Jul 2026 16:27:36 -0400 Subject: [PATCH 09/10] fix(observer): re-scope activity panel on channel mismatch; seed carousel to caller channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two remaining leak surfaces on #1634: 1. ChannelPane passed the raw openAgentSessionChannelId to AgentSessionThreadPanel even when activeChannel.id != openAgentSessionChannelId. The channel prop already nulled on mismatch (interrupt guard), but channelId was still stale, so sessionChannelId = channelId (test-3) leaked the wrong channel's content and badge. Fix: when mismatch, pass activeChannelId instead so the panel re-scopes to the current channel. 2. resolveActivityChannelId in ProfileLiveActivityEmbed used feedScope.preferredChannelId (globally most-recently-active) as the preferred resolver, so the carousel opened on whichever channel was most recently active across all sessions — not the channel the profile was opened from. Fix: prefer callerChannelId (already threaded from ChannelPane by the zero-slides fix) so the carousel defaults to the caller's channel when that channel has a slide. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- desktop/src/features/channels/ui/ChannelPane.tsx | 7 ++++++- desktop/src/features/profile/ui/UserProfilePanelTabs.tsx | 2 +- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/desktop/src/features/channels/ui/ChannelPane.tsx b/desktop/src/features/channels/ui/ChannelPane.tsx index 9481589b9..1ef321fb6 100644 --- a/desktop/src/features/channels/ui/ChannelPane.tsx +++ b/desktop/src/features/channels/ui/ChannelPane.tsx @@ -858,7 +858,12 @@ export const ChannelPane = React.memo(function ChannelPane({ ? activeChannel : null } - channelId={openAgentSessionChannelId} + channelId={ + openAgentSessionChannelId && + activeChannel?.id !== openAgentSessionChannelId + ? activeChannelId + : openAgentSessionChannelId + } isSinglePanelView={ useSplitAuxiliaryPane ? false : isSinglePanelView } diff --git a/desktop/src/features/profile/ui/UserProfilePanelTabs.tsx b/desktop/src/features/profile/ui/UserProfilePanelTabs.tsx index e920c1433..7659aa906 100644 --- a/desktop/src/features/profile/ui/UserProfilePanelTabs.tsx +++ b/desktop/src/features/profile/ui/UserProfilePanelTabs.tsx @@ -371,7 +371,7 @@ function ProfileLiveActivityEmbed({ const activeChannelId = resolveActivityChannelId( slides, selectedChannelId, - feedScope.preferredChannelId, + callerChannelId ?? feedScope.preferredChannelId, ); const selectedIndex = activeChannelId ? slides.indexOf(activeChannelId) : 0; From 6657ea353b4bd34fd4620af300f976361208c462 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 8 Jul 2026 16:35:16 -0400 Subject: [PATCH 10/10] fix(desktop): unify channel scope for Activity panel channel and channelId props Bug 3's fix rebased channelId to activeChannelId on a stale openAgentSessionChannelId mismatch, but the adjacent channel prop still passed null for the same case. This split the panel's effective channel across two values: sessionChannelId (from channelId) resolved to the current channel and correctly scoped the feed and header badge, but handleInterruptTurn early-returned on !channel so an agent actively working in the now-active channel could show an enabled Stop action that silently did nothing. Fix: derive effectiveAgentSessionChannelId once (the same mismatch expression previously inline on channelId) and feed both channelId and channel from it. In the mismatch case effectiveAgentSessionChannelId === activeChannel.id (guaranteed because this IIFE only runs inside the activeChannel && selectedAgent branch), so channel passes activeChannel instead of null. channel and channelId now always express a single consistent scope. Three-case walk: - mismatch: effectiveAgentSessionChannelId = activeChannelId, channel = activeChannel, Stop operates on the current channel - match: effectiveAgentSessionChannelId = openAgentSessionChannelId, channel = activeChannel (same as before, channel id matches) - null open: effectiveAgentSessionChannelId = null, channel falls through to the isAgentInActivityList guard (unchanged) Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../src/features/channels/ui/ChannelPane.tsx | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/desktop/src/features/channels/ui/ChannelPane.tsx b/desktop/src/features/channels/ui/ChannelPane.tsx index 1ef321fb6..4f49910e1 100644 --- a/desktop/src/features/channels/ui/ChannelPane.tsx +++ b/desktop/src/features/channels/ui/ChannelPane.tsx @@ -842,13 +842,22 @@ export const ChannelPane = React.memo(function ChannelPane({ })() ) : activeChannel && selectedAgent ? ( (() => { + // When the panel was opened from a different channel than the + // currently active one, re-scope it to the active channel so + // that both the content/header AND channel-backed actions (e.g. + // Stop current turn) operate on the same channel object. + const effectiveAgentSessionChannelId = + openAgentSessionChannelId && + activeChannel.id !== openAgentSessionChannelId + ? activeChannelId + : openAgentSessionChannelId; const panel = (