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/ingestArchivedObserverEvents.test.mjs b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs index 49add6c6a..4a1978a28 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,108 @@ 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 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. + +import { + createArchivePagingState, + applyChannelReset, +} from "@/features/agents/ui/archivePagingState.ts"; + +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.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 + + // 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( + ps.hasOlderArchived, + true, + "hasOlderArchived resets to true on channel switch", + ); + assert.equal( + ps.isFetching, + false, + "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_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); + + assert.equal(ps.cursor, null, "switch A→B: cursor reset"); + assert.equal( + ps.hasOlderArchived, + true, + "switch A→B: hasOlderArchived reset", + ); + + // 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/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index adf6ea92d..e2de2e2bc 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); @@ -343,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); @@ -368,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); @@ -513,6 +598,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..f7914db55 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}:${block.runIndex}`; + } 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/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/agentSessionTranscriptGrouping.test.mjs b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.test.mjs index 599cc9244..dd6696ccd 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,446 @@ 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)); +}); + +// ── 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", + ); +}); + +// ── Duplicate-key guard: non-contiguous runs of the same sessionId ──────────── + +test("buildTranscriptDisplayBlocks_nonContiguousRunsSameSession_distinctBoundaryKeys", () => { + // 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}` + // 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("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"); + + // Four distinct session runs → three boundaries (before sess-A, before + // sess-C, and before the re-occurring sess-A). + assert.equal( + boundaryBlocks.length, + 3, + "three boundaries for four runs (sess-B, sess-A, sess-C, 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 + // `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 efbf1c5c2..1aef3b7a1 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts +++ b/desktop/src/features/agents/ui/agentSessionTranscriptGrouping.ts @@ -15,7 +15,33 @@ 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"; + /** + * 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 = | { kind: "item"; item: TranscriptItem } @@ -356,6 +382,72 @@ function getRenderClass(item: TranscriptItem) { return item.renderClass ?? descriptor.renderClass; } +/** + * Split a flat, time-ordered array of TranscriptItems into contiguous session + * 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. + */ +function splitIntoSessionRuns( + items: TranscriptItem[], +): 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) { + 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; +} + /** * Build presentation-only display blocks from normalized transcript items. * Raw observer order is preserved in the source items; this only reorders @@ -365,9 +457,86 @@ 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, + runIndex: i, + }); + } + + 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 +638,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/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 73603a029..a04eefbd0 100644 --- a/desktop/src/features/agents/ui/useObserverEvents.ts +++ b/desktop/src/features/agents/ui/useObserverEvents.ts @@ -9,10 +9,19 @@ 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"; +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 @@ -55,33 +64,62 @@ 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. + * + * 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). * - * 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. + * 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 subscription exists (returns `hasOlderArchived: - * false` without making any archive calls). + * 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; - // 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). + // 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. const [hasSubscription, setHasSubscription] = React.useState( - null, + ps.hasSubscription, ); - const [hasOlderArchived, setHasOlderArchived] = React.useState(true); - const isFetchingRef = React.useRef(false); - // 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; ps is a stable ref excluded from deps by convention; setHasOlderArchived is a stable React state setter + React.useEffect(() => { + 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; @@ -96,14 +134,24 @@ export function useLoadArchivedObserverEvents(enabled: boolean) { (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. + ps.backfillStatus = "done"; + ps.backfillResolve?.(); } }) .catch(() => { if (!cancelled) { setHasSubscription(false); + ps.hasSubscription = false; setHasOlderArchived(false); + ps.hasOlderArchived = false; + ps.backfillStatus = "done"; + ps.backfillResolve?.(); } }); return () => { @@ -111,22 +159,110 @@ 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 + // 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 || ps.backfillStatus !== "pending") { + return; + } + ps.backfillStatus = "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 { + ps.backfillStatus = "done"; + ps.backfillResolve?.(); + } + })(); + 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 || - isFetchingRef.current || + !channelId || + ps.isFetching || !hasOlderArchived ) { return; } - isFetchingRef.current = true; + // 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 (ps.backfillPromise) { + await ps.backfillPromise; + } + + // Re-check after awaiting: hasOlderArchived might have been set false + // while we were waiting (e.g. subscription check failed). + if (!hasOlderArchived) { + return; + } + + ps.isFetching = true; try { - const before = cursorRef.current ?? undefined; - const events = await readArchivedEvents("owner_p", identityPubkey, { - kinds: [24200], + const before = ps.cursor ?? undefined; + const events = await readArchivedObserverEventsForChannel(channelId, { before: before ?? null, limit: ARCHIVED_EVENTS_PAGE_SIZE, }); @@ -136,23 +272,24 @@ export function useLoadArchivedObserverEvents(enabled: boolean) { // 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, }; 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); + ps.hasOlderArchived = false; } } catch (error) { console.error("[useLoadArchivedObserverEvents] fetch failed:", error); } finally { - isFetchingRef.current = false; + ps.isFetching = 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..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); + 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/ChannelPane.tsx b/desktop/src/features/channels/ui/ChannelPane.tsx index 6740359d6..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 = ( { 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( diff --git a/desktop/src/features/profile/ui/UserProfilePanel.tsx b/desktop/src/features/profile/ui/UserProfilePanel.tsx index 24eb799bf..319352f01 100644 --- a/desktop/src/features/profile/ui/UserProfilePanel.tsx +++ b/desktop/src/features/profile/ui/UserProfilePanel.tsx @@ -101,6 +101,7 @@ import { getUserProfilePanelHeaderContent } from "@/features/profile/ui/UserProf export type { ProfilePanelTab, ProfilePanelView }; export function UserProfilePanel({ + callerChannelId = null, canResetWidth, currentPubkey, isSinglePanelView = false, @@ -819,6 +820,7 @@ export function UserProfilePanel({ canInstantiateAgent={canInstantiateAgent} canOpenAgentLogs={canOpenAgentLogs} canViewActivity={canViewActivity} + callerChannelId={callerChannelId} channelCount={profileChannels.length} channelIdToName={channelIdToName} channels={profileChannels} diff --git a/desktop/src/features/profile/ui/UserProfilePanelSections.tsx b/desktop/src/features/profile/ui/UserProfilePanelSections.tsx index ebde518c7..3e11b8185 100644 --- a/desktop/src/features/profile/ui/UserProfilePanelSections.tsx +++ b/desktop/src/features/profile/ui/UserProfilePanelSections.tsx @@ -63,6 +63,7 @@ export { AgentInstructionsFocusedView } from "@/features/profile/ui/UserProfileP export type ProfileSummaryViewProps = { activityAgent: ProfileActivityAgent | null; + callerChannelId: string | null; canAddToChannel: boolean; canEditAgent: boolean; canOpenAgentLogs: boolean; @@ -172,6 +173,7 @@ function RuntimeTabStatusDot({ status }: { status: RuntimeTabStatus }) { export function ProfileSummaryView({ activityAgent, + callerChannelId, canAddToChannel, canEditAgent, canOpenAgentLogs, @@ -384,6 +386,7 @@ export function ProfileSummaryView({ activeTurns={activeTurns} activityAgent={activityAgent} agentInfoFields={agentInfoFields} + callerChannelId={callerChannelId} channelIdToName={channelIdToName} isArchived={isArchived} onOpenActivity={onOpenActivity} diff --git a/desktop/src/features/profile/ui/UserProfilePanelTabs.tsx b/desktop/src/features/profile/ui/UserProfilePanelTabs.tsx index 25b0fccb9..7659aa906 100644 --- a/desktop/src/features/profile/ui/UserProfilePanelTabs.tsx +++ b/desktop/src/features/profile/ui/UserProfilePanelTabs.tsx @@ -273,6 +273,7 @@ export function ProfileInfoTabContent({ activeTurns, activityAgent, agentInfoFields, + callerChannelId, channelIdToName, isArchived, onOpenActivity, @@ -282,6 +283,7 @@ export function ProfileInfoTabContent({ activeTurns: ActiveTurnSummary[]; activityAgent: ProfileActivityAgent | null; agentInfoFields: ProfileField[]; + callerChannelId: string | null; channelIdToName: Record; isArchived: boolean; onOpenActivity: (channelId?: string | null) => void; @@ -316,6 +318,7 @@ export function ProfileInfoTabContent({ ; feedScope: ProfileActivityFeedScope; onOpenActivity: (channelId?: string | null) => void; @@ -366,7 +371,7 @@ function ProfileLiveActivityEmbed({ const activeChannelId = resolveActivityChannelId( slides, selectedChannelId, - feedScope.preferredChannelId, + callerChannelId ?? feedScope.preferredChannelId, ); const selectedIndex = activeChannelId ? slides.indexOf(activeChannelId) : 0; @@ -467,7 +472,7 @@ function ProfileLiveActivityEmbed({ { + 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. *