diff --git a/crates/buzz-core/src/desktop_observation.rs b/crates/buzz-core/src/desktop_observation.rs new file mode 100644 index 00000000000..21ba206280a --- /dev/null +++ b/crates/buzz-core/src/desktop_observation.rs @@ -0,0 +1,123 @@ +//! Owner-private, advisory Desktop observations; never agent readiness or placement. +use nostr::{nips::nip44, Event, EventBuilder, Keys, Kind, Tag}; +use serde::{Deserialize, Serialize}; + +use crate::{desktop_profile::DesktopProfile, kind::KIND_DESKTOP_OBSERVATION}; + +/// A pulse for one local profile. The signed event timestamp is the observed time. +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct DesktopObservation { + /// Format version. + pub v: u8, + /// Canonical community, encrypted along with the coordinate. + pub community: String, + /// Stable Desktop profile coordinate, not an execution credential. + pub id: String, +} + +/// Validate the bounded public envelope without decrypting it. +pub fn validate_envelope(event: &Event) -> Result<(), &'static str> { + crate::desktop_profile::validate_private_desktop_envelope(event, KIND_DESKTOP_OBSERVATION) +} + +impl DesktopObservation { + /// Observe a profile belonging to this local Desktop. + pub fn new(profile: DesktopProfile) -> Self { + Self { + v: 1, + community: profile.community, + id: profile.id, + } + } + + /// Encrypt and sign a fresh observation, without rewriting the durable profile. + pub fn sign(&self, keys: &Keys) -> Result { + let content = nip44::encrypt( + keys.secret_key(), + &keys.public_key(), + serde_json::to_string(self).map_err(|e| e.to_string())?, + nip44::Version::V2, + ) + .map_err(|e| e.to_string())?; + EventBuilder::new(Kind::Custom(KIND_DESKTOP_OBSERVATION as u16), content) + .tag(Tag::identifier(&self.id)) + .sign_with_keys(keys) + .map_err(|e| e.to_string()) + } + + /// Authenticate and decrypt an observation before displaying its timestamp. + pub fn read(event: &Event, keys: &Keys, community: &str) -> Result { + validate_envelope(event)?; + event + .verify() + .map_err(|_| "invalid Desktop observation signature")?; + if event.pubkey != keys.public_key() { + return Err("foreign Desktop observation".into()); + } + let plaintext = nip44::decrypt(keys.secret_key(), &keys.public_key(), &event.content) + .map_err(|_| "Desktop observation decryption failed")?; + let observation: Self = + serde_json::from_str(&plaintext).map_err(|_| "invalid Desktop observation")?; + let expected = Self::new(DesktopProfile::new( + community.to_owned(), + event.tags.identifier().unwrap_or_default().to_owned(), + )?); + if observation != expected { + return Err("Desktop observation scope mismatch".into()); + } + Ok(observation) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn observation_is_private_scoped_and_distinct_from_profile() { + let keys = Keys::generate(); + let profile = DesktopProfile::new("wss://one.example".into(), "a".repeat(32)).unwrap(); + let saved = profile.sign(&keys).unwrap(); + let observation = DesktopObservation::new(profile); + let event = observation.sign(&keys).unwrap(); + assert_eq!( + DesktopObservation::read(&event, &keys, &observation.community).unwrap(), + observation + ); + assert!(DesktopObservation::read(&event, &keys, "wss://two.example").is_err()); + assert!( + DesktopObservation::read(&event, &Keys::generate(), &observation.community).is_err() + ); + assert!(DesktopObservation::read(&saved, &keys, &observation.community).is_err()); + assert!(DesktopProfile::read(&event, &keys, &observation.community).is_err()); + let forged_author = EventBuilder::new(event.kind, &event.content) + .tags(event.tags.clone()) + .sign_with_keys(&Keys::generate()) + .unwrap(); + assert!(DesktopObservation::read(&forged_author, &keys, &observation.community).is_err()); + let mut tampered = event.clone(); + tampered.created_at = nostr::Timestamp::from(1); + assert!(DesktopObservation::read(&tampered, &keys, &observation.community).is_err()); + assert!(crate::kind::AUTHOR_ONLY_KINDS.contains(&KIND_DESKTOP_OBSERVATION)); + for field in ["v", "community", "id", "extra"] { + let mut payload = serde_json::to_value(&observation).unwrap(); + payload[field] = serde_json::json!("invalid"); + let ciphertext = nip44::encrypt( + keys.secret_key(), + &keys.public_key(), + payload.to_string(), + nip44::Version::V2, + ) + .unwrap(); + let invalid = EventBuilder::new(event.kind, ciphertext) + .tags(event.tags.clone()) + .sign_with_keys(&keys) + .unwrap(); + assert!( + DesktopObservation::read(&invalid, &keys, &observation.community).is_err(), + "{field}" + ); + } + } +} diff --git a/crates/buzz-core/src/desktop_profile.rs b/crates/buzz-core/src/desktop_profile.rs index 697808c4677..5b08a438b3a 100644 --- a/crates/buzz-core/src/desktop_profile.rs +++ b/crates/buzz-core/src/desktop_profile.rs @@ -20,8 +20,15 @@ pub struct DesktopProfile { /// Validate the public envelope without decrypting private content. pub fn validate_envelope(event: &Event) -> Result<(), &'static str> { + validate_private_desktop_envelope(event, KIND_DESKTOP_PROFILE) +} + +pub(crate) fn validate_private_desktop_envelope( + event: &Event, + kind: u32, +) -> Result<(), &'static str> { let tags: Vec<_> = event.tags.iter().map(|tag| tag.as_slice()).collect(); - if event.kind.as_u16() as u32 != KIND_DESKTOP_PROFILE + if event.kind.as_u16() as u32 != kind || event.created_at.as_secs() > 253_402_300_799 || !(132..=2048).contains(&event.content.len()) || tags.len() != 1 diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index bb8be1a2819..1e5bfd43d6e 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -120,6 +120,9 @@ pub const KIND_PRIVATE_MANAGED_AGENT: u32 = 30179; /// Owner-private encrypted Desktop identity/name, keyed by installation coordinate. pub const KIND_DESKTOP_PROFILE: u32 = 30180; +/// Owner-private, per-Desktop last-heard observation; not online or readiness. +pub const KIND_DESKTOP_OBSERVATION: u32 = 30181; + /// Kinds whose stored events are readable only by their author. /// /// The relay must never reveal the existence, count, tags, content, schedule, @@ -134,6 +137,7 @@ pub const AUTHOR_ONLY_KINDS: &[u32] = &[ KIND_PUSH_LEASE, KIND_PRIVATE_MANAGED_AGENT, KIND_DESKTOP_PROFILE, + KIND_DESKTOP_OBSERVATION, ]; /// Kinds that require a result-level read gate beyond the filter-layer @@ -664,6 +668,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_TEAM_CATALOG, KIND_PRIVATE_MANAGED_AGENT, KIND_DESKTOP_PROFILE, + KIND_DESKTOP_OBSERVATION, KIND_REPORT, KIND_PRODUCT_FEEDBACK, KIND_NIP29_PUT_USER, diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs index fe94c0f356a..2105e4a15a7 100644 --- a/crates/buzz-core/src/lib.rs +++ b/crates/buzz-core/src/lib.rs @@ -9,6 +9,7 @@ pub mod agent_turn_metric; /// Channel and membership enums shared across crates. pub mod channel; +pub mod desktop_observation; /// Owner-private Desktop display profiles. pub mod desktop_profile; /// NIP-AE Agent Engrams — slug grammar, conversation key, d-tag derivation, diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 8cc2b3cd08d..5a0d2a64150 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -702,7 +702,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 45); + assert_eq!(migrations.len(), 46); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -910,8 +910,9 @@ mod postgres_tests { assert!(migrations[32].sql.as_str().contains("kind = 30179")); assert!(migrations[32].sql.as_str().contains("search_tsv")); assert!(!migrations[0].sql.as_str().contains("30179")); - assert!(include_str!("../../../../schema/schema.sql") - .contains("kind IN (1059, 30179, 30180, 30300, 30350, 30622, 44100, 44101, 44200)")); + assert!(include_str!("../../../../schema/schema.sql").contains( + "kind IN (1059, 30179, 30180, 30181, 30300, 30350, 30622, 44100, 44101, 44200)" + )); // Public push-gateway authority is intentionally deployment-global and // durable: immediate revocation and hostile-relay admission cannot be @@ -2391,6 +2392,7 @@ mod postgres_tests { (2_u8, 30_350_i32), (3_u8, 30_179_i32), (4_u8, 30_180_i32), + (5_u8, 30_181_i32), ] { sqlx::query( "INSERT INTO events \ @@ -2420,7 +2422,13 @@ mod postgres_tests { .expect("read pre-push search behavior"); assert_eq!( before, - vec![(1, true), (30_179, true), (30_180, true), (30_350, true)] + vec![ + (1, true), + (30_179, true), + (30_180, true), + (30_181, true), + (30_350, true) + ] ); // 0014 fixes 30350 only. A brownfield database that stopped here still @@ -2442,6 +2450,7 @@ mod postgres_tests { (1, Some(true)), (30_179, Some(true)), (30_180, Some(true)), + (30_181, Some(true)), (30_350, None) ] ); @@ -2457,6 +2466,18 @@ mod postgres_tests { .await .expect("read pre-0045 Desktop search behavior"); assert!(desktop_indexed, "upgrade fixture must exercise legacy FTS"); + run_migrations_through(&pool, 45) + .await + .expect("apply through 45"); + let observation_indexed: bool = + sqlx::query_scalar("SELECT search_tsv IS NOT NULL FROM events WHERE kind = 30181") + .fetch_one(&pool) + .await + .unwrap(); + assert!( + observation_indexed, + "0046 must change brownfield observation FTS" + ); run_migrations(&pool) .await @@ -2474,6 +2495,7 @@ mod postgres_tests { (1, Some(true)), (30_179, None), (30_180, None), + (30_181, None), (30_350, None) ] ); diff --git a/crates/buzz-relay/src/api/desktop_profile_postgres_tests.rs b/crates/buzz-relay/src/api/desktop_profile_postgres_tests.rs index 922c4bb0226..5d863666832 100644 --- a/crates/buzz-relay/src/api/desktop_profile_postgres_tests.rs +++ b/crates/buzz-relay/src/api/desktop_profile_postgres_tests.rs @@ -2,7 +2,7 @@ use super::postgres_tests::bridge_handler_test_state; use super::*; use axum::{body::Body, http::Request}; -use buzz_core::kind::KIND_DESKTOP_PROFILE; +use buzz_core::kind::{KIND_DESKTOP_OBSERVATION, KIND_DESKTOP_PROFILE}; use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use serde_json::json; use tower::ServiceExt; @@ -67,6 +67,16 @@ fn drain(rx: &mut tokio::sync::mpsc::Receiver) -> Ve #[tokio::test] #[ignore = "requires Postgres"] async fn desktop_profile_authenticated_owner_query_and_private_storage() { + assert_private_desktop(KIND_DESKTOP_PROFILE).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn desktop_observation_authenticated_owner_query_and_private_storage() { + assert_private_desktop(KIND_DESKTOP_OBSERVATION).await; +} + +async fn assert_private_desktop(kind: u32) { let mut state = bridge_handler_test_state() .await .expect("test infrastructure"); @@ -86,13 +96,21 @@ async fn desktop_profile_authenticated_owner_query_and_private_storage() { uuid::Uuid::new_v4().simple().to_string(), ) .unwrap(); - let event = profile.sign(&owner).unwrap(); + let id = profile.id.clone(); + let event = if kind == KIND_DESKTOP_PROFILE { + profile.sign(&owner).unwrap() + } else { + buzz_core::desktop_observation::DesktopObservation::new(profile) + .sign(&owner) + .unwrap() + }; let (status, result) = post(&state, &host, "/events", &owner, json!(event), true).await; assert_eq!(status, StatusCode::OK, "{result}"); assert_eq!(result["accepted"], true, "{result}"); // Match Desktop's actual bounded owner+kind inventory and exact-coordinate probe. - let own = json!([{"kinds":[KIND_DESKTOP_PROFILE], "authors":[owner.public_key().to_hex()], "limit":100}]); - let exact = json!([{"kinds":[KIND_DESKTOP_PROFILE], "authors":[owner.public_key().to_hex()], "#d":[profile.id], "limit":1}]); + let own = json!([{"kinds":[kind], "authors":[owner.public_key().to_hex()], "limit":100}]); + let exact = + json!([{"kinds":[kind], "authors":[owner.public_key().to_hex()], "#d":[id], "limit":1}]); for filters in [&own, &exact] { let (status, rows) = post(&state, &host, "/query", &owner, filters.clone(), true).await; assert_eq!(status, StatusCode::OK, "{rows}"); @@ -105,7 +123,7 @@ async fn desktop_profile_authenticated_owner_query_and_private_storage() { assert_eq!(status, StatusCode::UNAUTHORIZED, "{result}"); } // Known IDs cannot grant an authenticated outsider read access either. - let known = json!([{"ids":[event.id.to_hex()], "kinds":[KIND_DESKTOP_PROFILE,1]}]); + let known = json!([{"ids":[event.id.to_hex()], "kinds":[kind,1]}]); let (status, rows) = post(&state, &host, "/query", &outsider, known, true).await; assert_eq!(status, StatusCode::OK, "{rows}"); assert_eq!(rows, json!([])); diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 5ed5f1037e2..b2f86ec357c 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -2215,6 +2215,11 @@ mod tests { assert_author_only_fanout(buzz_core::kind::KIND_DESKTOP_PROFILE).await; } + #[tokio::test] + async fn desktop_observation_delivers_to_author_only() { + assert_author_only_fanout(buzz_core::kind::KIND_DESKTOP_OBSERVATION).await; + } + async fn assert_author_only_fanout(kind: u32) { let state = test_state().await; diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 13c781613fb..f60a76fd4a8 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -10,7 +10,6 @@ use tracing::{debug, error, info, warn}; use uuid::Uuid; use buzz_auth::Scope; -use buzz_core::kind::KIND_DESKTOP_PROFILE; use buzz_core::kind::{ event_kind_u32, is_identity_archive_request_kind, is_parameterized_replaceable, is_relay_admin_kind, KIND_AGENT_ENGRAM, KIND_AGENT_PROFILE, KIND_AGENT_TURN_METRIC, @@ -37,6 +36,7 @@ use buzz_core::kind::{ RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, RELAY_ADMIN_REMOVE_MEMBER, RELAY_ADMIN_SET_WORKSPACE_PROFILE, }; +use buzz_core::kind::{KIND_DESKTOP_OBSERVATION, KIND_DESKTOP_PROFILE}; use buzz_core::tenant::TenantContext; use buzz_core::verification::verify_event; use buzz_core::CommunityId; @@ -437,7 +437,7 @@ fn map_push_accept_error(error: super::push_lease::AcceptError) -> IngestError { /// Returns `Err` for unknown kinds — the relay rejects them. fn required_scope_for_kind(kind: u32, event: &Event) -> Result { match kind { - KIND_PROFILE | KIND_DESKTOP_PROFILE => Ok(Scope::UsersWrite), + KIND_PROFILE | KIND_DESKTOP_PROFILE | KIND_DESKTOP_OBSERVATION => Ok(Scope::UsersWrite), KIND_TEXT_NOTE | KIND_LONG_FORM => Ok(Scope::MessagesWrite), KIND_CONTACT_LIST | KIND_READ_STATE | KIND_USER_STATUS | KIND_AGENT_ENGRAM | KIND_EVENT_REMINDER | KIND_PERSONA | KIND_TEAM | KIND_MANAGED_AGENT @@ -659,6 +659,7 @@ pub(crate) fn is_global_only_kind(kind: u32) -> bool { | KIND_MANAGED_AGENT | KIND_PRIVATE_MANAGED_AGENT | KIND_DESKTOP_PROFILE + | KIND_DESKTOP_OBSERVATION | KIND_TEAM_CATALOG // NIP-34: git events use `a` tags (repo reference), not `h` tags (channel scope). // Parameterized replaceable kinds are keyed by (pubkey, kind, d_tag). @@ -2792,6 +2793,11 @@ async fn ingest_event_inner( } } + if kind_u32 == KIND_DESKTOP_OBSERVATION { + buzz_core::desktop_observation::validate_envelope(&event) + .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; + } + if kind_u32 == KIND_DESKTOP_PROFILE { buzz_core::desktop_profile::validate_envelope(&event) .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; diff --git a/desktop/src-tauri/src/commands/desktop_profiles.rs b/desktop/src-tauri/src/commands/desktop_profiles.rs index 3b3017da71b..d3438d71c51 100644 --- a/desktop/src-tauri/src/commands/desktop_profiles.rs +++ b/desktop/src-tauri/src/commands/desktop_profiles.rs @@ -1,5 +1,5 @@ //! Durable read-only Desktop profiles, separate from persona publication queues. -use buzz_core_pkg::desktop_profile::DesktopProfile; +use buzz_core_pkg::{desktop_observation::DesktopObservation, desktop_profile::DesktopProfile}; use nostr::{Event, JsonUtil}; use rusqlite::{Connection, OptionalExtension, TransactionBehavior}; use serde_json::{json, Value}; @@ -97,6 +97,50 @@ pub fn read_desktop_profiles( Ok(json!(rows?)) } +/// Sign a fresh pulse for the persisted local profile, never an arbitrary host ID. +#[tauri::command] +pub fn prepare_desktop_observation( + app: AppHandle, + state: State<'_, AppState>, + owner: String, + community: String, +) -> Result { + let scope = scope(&app, &state, &owner, &community)?; + Ok(json!({ "event": prepare_observation(&mut open_retention_db(&scope.db_path)?, &scope)? })) +} + +fn prepare_observation(conn: &mut Connection, scope: &RetentionScope) -> Result { + let saved = prepare(conn, scope)?; + let event: Event = serde_json::from_value(saved["event"].clone()).map_err(|e| e.to_string())?; + let profile = DesktopProfile::read( + &event, + &scope.owner_keys, + scope.relay_url.trim_end_matches('/'), + )?; + DesktopObservation::new(profile).sign(&scope.owner_keys) +} + +/// Verify bounded owner-private observations; the UI treats their timestamps as advisory. +#[tauri::command] +pub fn read_desktop_observations( + app: AppHandle, + state: State<'_, AppState>, + owner: String, + community: String, + events: Vec, +) -> Result { + let scope = scope(&app, &state, &owner, &community)?; + if events.len() > 100 { + return Err("too many Desktop observations".into()); + } + let mut rows = Vec::with_capacity(events.len()); + for event in &events { + let observation = DesktopObservation::read(event, &scope.owner_keys, &community)?; + rows.push(json!({ "id": observation.id, "heard": event.created_at.as_secs() })); + } + Ok(json!(rows)) +} + #[cfg(test)] mod tests { use super::*; @@ -119,6 +163,16 @@ mod tests { let accepted = prepare(&mut reopened, &scope).unwrap(); assert_eq!(accepted, first); assert_eq!(reopened.total_changes(), 0, "no repeated native writes"); + let pulse = prepare_observation(&mut reopened, &scope).unwrap(); + let observation = + DesktopObservation::read(&pulse, &scope.owner_keys, &scope.relay_url).unwrap(); + assert_eq!(first["event"]["tags"][0][1], observation.id); + assert_eq!( + prepare(&mut reopened, &scope).unwrap(), + first, + "pulses never rewrite profiles" + ); + assert_eq!(reopened.total_changes(), 0); let other = RetentionScope { db_path: dir.path().join("two.db"), relay_url: scope.relay_url.clone(), diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 65fe1aaa4d7..edffed5e462 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -554,6 +554,8 @@ pub fn run() { get_identity, prepare_desktop_profile, read_desktop_profiles, + prepare_desktop_observation, + read_desktop_observations, get_nsec, generate_backup_passphrase, create_ncryptsec_backup, diff --git a/desktop/src/features/agents/desktopList.test.mjs b/desktop/src/features/agents/desktopList.test.mjs index 4ffa141e0d2..6f7330d978c 100644 --- a/desktop/src/features/agents/desktopList.test.mjs +++ b/desktop/src/features/agents/desktopList.test.mjs @@ -155,10 +155,11 @@ test("rendered list distinguishes current, partial, unavailable and empty withou render({ ...list, rows: [], partial: false }), /No Desktop profiles found/, ); - assert.doesNotMatch(html, /Last heard|Online|Offline/); + assert.match(html, /Last heard: Unknown/); + assert.doesNotMatch(html, /Online|Offline/); }); -test("mounted cache clears both scopes, fences late reads and retains rows on failure", async () => { +test("mounted cache clears both scopes, fences late reads and retains rows on failure", async (t) => { const { JSDOM } = await import("jsdom"); const dom = new JSDOM("
", { url: "https://desktop.test", @@ -198,8 +199,26 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa let current; const originalFetch = relayClient.fetchEvents; const originalPublish = relayClient.publishEvent; + const originalReconnect = relayClient.subscribeToReconnects; + let reconnect; + let pulses = 0; + relayClient.subscribeToReconnects = (callback) => { + reconnect = callback; + return () => { + reconnect = undefined; + }; + }; window.__TAURI_INTERNALS__ = { invoke: async (command, args) => { + if (command === "prepare_desktop_observation") + return { event: { ...first, kind: 30181 } }; + if (command === "read_desktop_observations") + return [ + { + id: `${args.owner}-${args.community}`, + heard: Math.floor(Date.now() / 1000), + }, + ]; if (command === "prepare_desktop_profile") { current = { ...first, @@ -218,6 +237,7 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa }; relayClient.fetchEvents = async (filter) => { if (fail) throw Error("unavailable"); + if (filter.kinds[0] === 30181) return []; if (filter["#d"]) return [current]; const rows = [current]; if (hold) { @@ -228,8 +248,9 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa } return rows; }; - relayClient.publishEvent = async () => { - throw Error("unexpected rewrite"); + relayClient.publishEvent = async (event) => { + assert.equal(event.kind, 30181, "no profile heartbeat rewrite"); + pulses++; }; function Screen() { controls = useCommunities(); @@ -246,6 +267,7 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa await new Promise((resolve) => setTimeout(resolve, 20)); }); const text = () => document.body.textContent; + t.mock.timers.enable({ apis: ["setInterval"] }); try { await React.act(async () => root.render( @@ -262,6 +284,15 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa ); await settle(); assert.match(text(), /owner-a-wss:\/\/a.example/); + assert.match(text(), /Last heard: Recent/); + const beforeReconnect = pulses; + await React.act(async () => reconnect()); + await settle(); + assert.ok(pulses > beforeReconnect, "reconnect reports a fresh pulse"); + const beforeTimer = pulses; + await React.act(async () => t.mock.timers.tick(60_000)); + await settle(); + assert.ok(pulses > beforeTimer, "bounded periodic publisher runs"); hold = true; await React.act(async () => { void client.refetchQueries({ queryKey: ["desktop-profiles"] }); @@ -283,8 +314,11 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa fail = true; await React.act(async () => { await client.refetchQueries({ queryKey: ["desktop-profiles"] }); + await client.refetchQueries({ queryKey: ["desktop-observations"] }); }); await settle(); + assert.match(text(), /Last-heard refresh unavailable/); + assert.match(text(), /Last heard: Recent/); assert.match(text(), /unavailable/); assert.match(text(), /owner-b-wss:\/\/a.example/); await React.act(async () => controls.switchCommunity("b")); @@ -293,6 +327,7 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa fail = false; await React.act(async () => { await client.refetchQueries({ queryKey: ["desktop-profiles"] }); + await client.refetchQueries({ queryKey: ["desktop-observations"] }); }); await settle(); assert.match(text(), /owner-b-wss:\/\/b.example/); @@ -301,6 +336,13 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa client.clear(); relayClient.fetchEvents = originalFetch; relayClient.publishEvent = originalPublish; + relayClient.subscribeToReconnects = originalReconnect; + assert.equal( + reconnect, + undefined, + "reconnect producer unsubscribed on unmount", + ); + t.mock.timers.reset(); dom.window.close(); } }); diff --git a/desktop/src/features/agents/desktopObservations.test.mjs b/desktop/src/features/agents/desktopObservations.test.mjs new file mode 100644 index 00000000000..b3166ba25d8 --- /dev/null +++ b/desktop/src/features/agents/desktopObservations.test.mjs @@ -0,0 +1,125 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { + refreshDesktopObservations, + desktopFreshness, +} from "./desktopObservations.ts"; + +const scope = { owner: "owner-a", community: "wss://a.example" }; +function fixture(boundary) { + let epoch = 0; + const calls = []; + const event = { id: "pulse", kind: 30181 }; + const rows = [ + { id: "a", heard: 100 }, + { id: "b", heard: 10 }, + ]; + const finish = (name, value) => { + if (name === boundary) epoch++; + return value; + }; + const f = { + calls, + rows, + ipc: async (command, args) => { + assert.equal(args.owner, scope.owner); + assert.equal(args.community, scope.community); + calls.push(command); + if (command === "prepare_desktop_observation") + return finish(command, { event }); + assert.equal(command, "read_desktop_observations"); + return finish(command, rows); + }, + relay: { + getSessionEpoch: () => epoch, + publishEvent: async (value) => { + assert.equal(value, event); + calls.push("publish"); + finish("publishEvent"); + }, + fetchEvents: async (filter) => { + assert.deepEqual(filter, { + kinds: [30181], + authors: [scope.owner], + limit: 100, + }); + calls.push("fetch"); + return finish("fetchEvents", [event]); + }, + }, + }; + f.refresh = (active = () => true) => + refreshDesktopObservations(scope, active, f.ipc, f.relay); + return f; +} + +test("every async completion is fenced, including decrypt and late ACK", async () => { + for (const boundary of [ + "prepare_desktop_observation", + "publishEvent", + "fetchEvents", + "read_desktop_observations", + ]) { + const f = fixture(boundary); + await assert.rejects(f.refresh(), /scope changed/); + if (boundary === "prepare_desktop_observation") + assert.ok(!f.calls.includes("publish")); + } + const f = fixture(); + await assert.rejects(f.refresh(() => false)); + assert.deepEqual(f.calls, []); +}); + +test("bounded/invalid reads are not silently authoritative, clocks and staleness stay advisory", async () => { + const f = fixture(); + const good = await f.refresh(); + assert.deepEqual(good.rows, f.rows); + assert.deepEqual(f.calls, [ + "prepare_desktop_observation", + "publish", + "fetch", + "read_desktop_observations", + ]); + f.relay.publishEvent = async () => { + throw Error("offline"); + }; + assert.ok((await f.refresh()).warning); + f.relay.fetchEvents = async () => Array(100).fill({}); + assert.equal((await f.refresh()).partial, true); + const read = f.ipc; + f.ipc = async (command, args) => { + if (command === "read_desktop_observations") + throw Error("invalid signature"); + return read(command, args); + }; + await assert.rejects(f.refresh(), /invalid signature/); + for (const [heard, now, expected] of [ + [undefined, 100, "Unknown"], + [100, 280, "Recent"], + [100, 281, "Stale"], + [101, 100, "Unknown (Desktop clock ahead)"], + ]) + assert.ok(desktopFreshness(heard, now).startsWith(expected)); +}); + +test("history/live replacement races select newest signed time and lower ID, not arrival", async () => { + const f = fixture(); + const events = [ + { id: "old", created_at: 10 }, + { id: "b", created_at: 20 }, + { id: "a", created_at: 20 }, + ]; + const ipc = f.ipc; + f.ipc = async (command, args) => { + if (command === "read_desktop_observations") + assert.deepEqual( + args.events.map((e) => e.id), + ["a", "b", "old"], + ); + return ipc(command, args); + }; + for (const batch of [events, [...events].reverse()]) { + f.relay.fetchEvents = async () => batch; + await f.refresh(); + } +}); diff --git a/desktop/src/features/agents/desktopObservations.ts b/desktop/src/features/agents/desktopObservations.ts new file mode 100644 index 00000000000..d548fa42433 --- /dev/null +++ b/desktop/src/features/agents/desktopObservations.ts @@ -0,0 +1,89 @@ +import { invoke } from "@tauri-apps/api/core"; +import { useQuery } from "@tanstack/react-query"; +import { relayClient } from "@/shared/api/relayClient"; +import type { RelayEvent } from "@/shared/api/types"; +import type { DesktopScope } from "./desktopList"; + +export type DesktopObservation = { id: string; heard: number }; +export const DESKTOP_PULSE_MS = 60_000; + +/** A failed pulse does not hide other Desktops; failed reads retain cached observations. */ +export async function refreshDesktopObservations( + scope: DesktopScope, + active: () => boolean, + ipc = invoke, + relay = relayClient, +) { + const epoch = relay.getSessionEpoch(); + const check = () => { + if (!active() || epoch !== relay.getSessionEpoch()) + throw new Error("Desktop observation scope changed"); + }; + const wait = async (work: Promise) => { + const result = await work; + check(); + return result; + }; + check(); + let warning = ""; + try { + const { event } = await wait( + ipc<{ event: RelayEvent }>("prepare_desktop_observation", scope), + ); + await wait( + relay.publishEvent( + event, + "Desktop pulse timed out", + "Desktop pulse failed", + check, + ), + ); + } catch { + check(); + // The next bounded interval retries with a new observation, not an old heartbeat. + warning = "This Desktop could not report its last-heard time. Will retry."; + } + const events = await wait( + relay.fetchEvents({ kinds: [30181], authors: [scope.owner], limit: 100 }), + ); + const rows = await wait( + ipc("read_desktop_observations", { + ...scope, + // History is chronological and may include a live replacement before EOSE. + // First matching host wins in the view: newest signed time, then lower ID. + events: [...events].sort( + (a, b) => + b.created_at - a.created_at || + (a.id < b.id ? -1 : a.id > b.id ? 1 : 0), + ), + }), + ); + return { rows, warning, partial: events.length === 100 }; +} + +export function useDesktopObservations(scope: DesktopScope | null) { + return useQuery({ + queryKey: ["desktop-observations", scope?.owner, scope?.community], + enabled: !!scope, + queryFn: ({ signal }) => { + if (!scope) throw new Error("Desktop scope unavailable"); + return refreshDesktopObservations(scope, () => !signal.aborted); + }, + gcTime: 0, + staleTime: DESKTOP_PULSE_MS / 2, + retry: false, + refetchOnWindowFocus: false, + }); +} + +/** Signed sender time is advisory, including clock skew; it never establishes agent death. */ +export function desktopFreshness(heard: number | undefined, now: number) { + if (heard === undefined) return "Unknown"; + const state = + heard > now + ? "Unknown (Desktop clock ahead)" + : now - heard <= 180 + ? "Recent" + : "Stale"; + return `${state} · ${new Date(heard * 1000).toLocaleString()}`; +} diff --git a/desktop/src/features/agents/ui/KnownDesktops.tsx b/desktop/src/features/agents/ui/KnownDesktops.tsx index 1d986d7bfd2..68eeacc1bcb 100644 --- a/desktop/src/features/agents/ui/KnownDesktops.tsx +++ b/desktop/src/features/agents/ui/KnownDesktops.tsx @@ -1,25 +1,40 @@ -import { useEffect } from "react"; +import { useEffect, useState } from "react"; import { useQuery } from "@tanstack/react-query"; import { useIdentityQuery } from "@/shared/api/hooks"; import { useCommunities } from "@/features/communities/useCommunities"; import { relayClient } from "@/shared/api/relayClient"; import { Button } from "@/shared/ui/button"; import { refreshDesktopList, type DesktopList } from "../desktopList"; +import { + DESKTOP_PULSE_MS, + desktopFreshness, + useDesktopObservations, + type DesktopObservation, +} from "../desktopObservations"; type View = { list: DesktopList | null; error: boolean; loading: boolean; refresh: () => void; + observations?: DesktopObservation[]; + observationWarning?: string; + now?: number; }; -function useDesktopList() { +function useDesktopScope() { const owner = useIdentityQuery().data?.pubkey; const { activeCommunity } = useCommunities(); const community = activeCommunity?.relayUrl .trim() .replace(/^http/, "ws") .replace(/\/+$/, ""); + return owner && community ? { owner, community } : null; +} + +function useDesktopList() { + const scope = useDesktopScope(); + const { owner, community } = scope ?? {}; return useQuery({ queryKey: ["desktop-profiles", owner, community], enabled: !!owner && !!community, @@ -38,31 +53,62 @@ function useDesktopList() { /** Startup and Agents share the existing owner/community query cache. */ export function DesktopListStartup() { const { refetch } = useDesktopList(); - useEffect( - () => - relayClient.subscribeToReconnects(() => { - void refetch(); - }), - [refetch], - ); + const { refetch: pulse } = useDesktopObservations(useDesktopScope()); + useEffect(() => { + const timer = setInterval(() => { + void pulse(); + }, DESKTOP_PULSE_MS); + const unsubscribe = relayClient.subscribeToReconnects(() => { + void refetch(); + void pulse(); + }); + return () => { + clearInterval(timer); + unsubscribe(); + }; + }, [refetch, pulse]); return null; } export function KnownDesktops() { const query = useDesktopList(); + const observations = useDesktopObservations(useDesktopScope()); + const [now, setNow] = useState(() => Date.now() / 1000); + useEffect(() => { + const timer = setInterval(() => setNow(Date.now() / 1000), 30_000); + return () => clearInterval(timer); + }, []); return ( { void query.refetch(); + void observations.refetch(); }} /> ); } -export function DesktopListView({ list, loading, error, refresh }: View) { +export function DesktopListView({ + list, + loading, + error, + refresh, + observations, + observationWarning, + now = Date.now() / 1000, +}: View) { return (

Private to you. Saved profiles do not indicate whether a Desktop is - online or ready to run agents. + online or ready to run agents. Last heard is a Desktop observation, not + proof that its agents are running or stopped.

{loading &&

Loading Desktop profiles…

} {error && ( @@ -89,6 +136,7 @@ export function DesktopListView({ list, loading, error, refresh }: View) { Desktop profiles unavailable. Previously loaded profiles are retained.

)} + {observationWarning &&

{observationWarning}

} {list?.warning &&

{list.warning}

} {list?.partial && (

Partial list: showing up to 100 profiles.

@@ -105,6 +153,13 @@ export function DesktopListView({ list, loading, error, refresh }: View) { {new Date(row.updated * 1000).toLocaleString()} +
+ Last heard:{" "} + {desktopFreshness( + observations?.find((item) => item.id === row.id)?.heard, + now, + )} +
))} diff --git a/desktop/src/shared/api/relayClientPublishRejection.test.mjs b/desktop/src/shared/api/relayClientPublishRejection.test.mjs index 7875b8679ab..f323623ef6f 100644 --- a/desktop/src/shared/api/relayClientPublishRejection.test.mjs +++ b/desktop/src/shared/api/relayClientPublishRejection.test.mjs @@ -293,3 +293,52 @@ test("a community switch after send failure cannot retry through its replacement ); assert.equal(eventFrames().length, 0); }); + +test("Desktop pulse cancellation fences rate-limit and reconnect continuations", async () => { + const { refreshDesktopObservations } = await import( + "../../features/agents/desktopObservations.ts" + ); + for (const boundary of ["rate-limit", "reconnect"]) { + reset(); + const client = connectedClient(); + const scope = { owner: "owner", community: "wss://a.example" }; + let active = true; + const reconnect = deferred(); + let reconnecting = false; + client.ensureConnected = async () => { + reconnecting = true; + await reconnect.promise; + client.wsId = 8; + return client.connectionGeneration; + }; + if (boundary === "rate-limit") activateRateLimit(4); + else + sendTransport = async () => { + throw Error("socket failed"); + }; + const result = refreshDesktopObservations( + scope, + () => active, + async (command) => { + assert.equal(command, "prepare_desktop_observation"); + return { event: { id: "pulse", kind: 30181 } }; + }, + client, + ); + const rejected = assert.rejects(result, /scope changed/); + if (boundary === "reconnect") await flushUntil(() => reconnecting); + else await Promise.resolve(); + active = false; + const generation = client.connectionGeneration; + resetRateLimitGate(); + reconnect.resolve(); + await rejected; + assert.equal(eventFrames().length, 0, boundary); + assert.equal(client.pendingEvents.size, 0); + assert.equal( + client.connectionGeneration, + generation, + "cancellation is not socket failure", + ); + } +}); diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 93384a40741..3839d01c7e5 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -725,6 +725,7 @@ export class RelayClient { event: RelayEvent, timeoutMessage: string, sendErrorMessage: string, + assertActive?: () => void, ) { return publishSessionEvent( { @@ -742,6 +743,7 @@ export class RelayClient { event, timeoutMessage, sendErrorMessage, + assertActive, ); } diff --git a/desktop/src/shared/api/relayEventPublisher.ts b/desktop/src/shared/api/relayEventPublisher.ts index ff719926700..6f6b6b7c824 100644 --- a/desktop/src/shared/api/relayEventPublisher.ts +++ b/desktop/src/shared/api/relayEventPublisher.ts @@ -19,9 +19,11 @@ export async function publishSessionEvent( event: RelayEvent, timeoutMessage: string, sendErrorMessage: string, + assertActive: () => void = () => {}, ): Promise { const publishOwnership = session.ownership(); await waitForRateLimit(); + assertActive(); if (publishOwnership !== session.ownership()) { throw new Error("Relay disconnected for community switch."); } @@ -48,6 +50,14 @@ export async function publishSessionEvent( return; } + try { + assertActive(); + } catch (error) { + window.clearTimeout(timeout); + session.pendingEvents.delete(event.id); + reject(error); + return; + } // Expected socket recovery must not reject the operation being retried. session.pendingEvents.delete(event.id); const sendError = session.recoverSocketFailure(error, sendErrorMessage); @@ -55,7 +65,9 @@ export async function publishSessionEvent( let retryGeneration: number | null = null; try { - retryGeneration = await session.reconnect(); + const reconnected = await session.reconnect(); + assertActive(); + retryGeneration = reconnected; if ( publishOwnership !== session.ownership() || session.generation() !== retryGeneration || diff --git a/migrations/0046_desktop_observation_fts.sql b/migrations/0046_desktop_observation_fts.sql new file mode 100644 index 00000000000..ba4776fd040 --- /dev/null +++ b/migrations/0046_desktop_observation_fts.sql @@ -0,0 +1,26 @@ +-- Owner-private Desktop observations must not enter legacy ciphertext search indexes. +-- Like 0033, this rewrites events under ACCESS EXCLUSIVE; schedule accordingly. +DO $$ +DECLARE + existing_expression TEXT; +BEGIN + SELECT pg_get_expr(d.adbin, d.adrelid) + INTO existing_expression + FROM pg_attrdef d + JOIN pg_attribute a + ON a.attrelid = d.adrelid + AND a.attnum = d.adnum + WHERE d.adrelid = 'events'::regclass + AND a.attname = 'search_tsv'; + + IF existing_expression IS NULL THEN + RAISE EXCEPTION 'events.search_tsv generated expression not found'; + END IF; + + ALTER TABLE events DROP COLUMN search_tsv; + EXECUTE format( + 'ALTER TABLE events ADD COLUMN search_tsv TSVECTOR GENERATED ALWAYS AS (CASE WHEN kind = 30181 THEN NULL::tsvector ELSE (%s) END) STORED', + existing_expression + ); + CREATE INDEX idx_events_search_tsv ON events USING GIN (search_tsv); +END $$; diff --git a/schema/schema.sql b/schema/schema.sql index 59147e4204e..16613480710 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -221,7 +221,7 @@ CREATE TABLE events ( -- never matches `@@`. -- Keep in sync with migrations (final state: 0001 + 0005 + 0014 + 0033). search_tsv TSVECTOR GENERATED ALWAYS AS ( - CASE WHEN kind IN (1059, 30179, 30180, 30300, 30350, 30622, 44100, 44101, 44200) THEN NULL::tsvector + CASE WHEN kind IN (1059, 30179, 30180, 30181, 30300, 30350, 30622, 44100, 44101, 44200) THEN NULL::tsvector ELSE to_tsvector('simple', content) END ) STORED,