From 08fbf9c95bbd9107fcf196d14d5e677368879db6 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 4 Sep 2026 12:24:55 -0400 Subject: [PATCH] fix(desktop): defer capability changes until real time advances Signed-off-by: Logan Johnson --- crates/buzz-core/src/desktop_capabilities.rs | 6 ++ .../src/commands/desktop_capabilities.rs | 83 +++++++++++++++++-- .../agents/desktopCapabilities.test.mjs | 28 +++++++ .../src/features/agents/desktopList.test.mjs | 34 +++++++- 4 files changed, 141 insertions(+), 10 deletions(-) diff --git a/crates/buzz-core/src/desktop_capabilities.rs b/crates/buzz-core/src/desktop_capabilities.rs index c621cca977..ef1c4d8f63 100644 --- a/crates/buzz-core/src/desktop_capabilities.rs +++ b/crates/buzz-core/src/desktop_capabilities.rs @@ -75,6 +75,11 @@ impl DesktopCapabilities { /// Encrypt/sign once, then persist these exact bytes for retries. pub fn sign(&self, keys: &Keys) -> Result { + self.sign_at(keys, nostr::Timestamp::now()) + } + + /// Sign at an observed wall-clock second, never a synthesized logical time. + pub fn sign_at(&self, keys: &Keys, observed: nostr::Timestamp) -> Result { self.validate()?; let content = nip44::encrypt( keys.secret_key(), @@ -85,6 +90,7 @@ impl DesktopCapabilities { .map_err(|e| e.to_string())?; let event = EventBuilder::new(Kind::Custom(KIND_DESKTOP_CAPABILITIES as u16), content) .tag(Tag::identifier(&self.id)) + .custom_created_at(observed) .sign_with_keys(keys) .map_err(|e| e.to_string())?; validate_envelope(&event)?; diff --git a/desktop/src-tauri/src/commands/desktop_capabilities.rs b/desktop/src-tauri/src/commands/desktop_capabilities.rs index bdab02274e..f5efd5c953 100644 --- a/desktop/src-tauri/src/commands/desktop_capabilities.rs +++ b/desktop/src-tauri/src/commands/desktop_capabilities.rs @@ -50,13 +50,16 @@ pub async fn prepare_desktop_capabilities( let facts = project(super::agent_discovery::discover_acp_providers(app.clone(), Some(false)).await?)?; let scope = scope(&app, &state, &owner, &community)?; - Ok(json!({ "event": prepare_report(&mut open_retention_db(&scope.db_path)?, &scope, facts)? })) + Ok( + json!({ "event": prepare_report(&mut open_retention_db(&scope.db_path)?, &scope, facts, nostr::Timestamp::now)? }), + ) } fn prepare_report( conn: &mut Connection, scope: &RetentionScope, facts: Vec, + clock: impl FnOnce() -> nostr::Timestamp, ) -> Result { let saved = prepare(conn, scope)?; let profile: Event = @@ -90,8 +93,16 @@ fn prepare_report( .unwrap_or(false); let event = match previous { Some(event) if unchanged => event, - _ => { - let event = report.sign(&scope.owner_keys)?; + previous => { + let now = clock(); + // Keep the prior retry record until real time advances. Signing tied + // ciphertext can lose NIP-33's lower-ID tie; never cache that loss or + // future-date a replacement. The existing pulse/reconnect/Refresh + // retries discovery, not a captured projection, without waiting here. + if previous.as_ref().is_some_and(|e| now <= e.created_at) { + return Err("Desktop capability facts deferred until the clock advances".into()); + } + let event = report.sign_at(&scope.owner_keys, now)?; tx.execute( "INSERT OR REPLACE INTO desktop_capabilities VALUES (1, ?1)", [event.as_json()], @@ -123,7 +134,7 @@ pub fn read_desktop_capabilities( mod tests { use super::*; #[test] - fn unchanged_facts_reopen_exact_bytes_changed_facts_replace_atomically() { + fn changed_facts_defer_until_real_clock_advances_then_win_signed_order() { let dir = tempfile::tempdir().unwrap(); let scope = RetentionScope { db_path: dir.path().join("report.db"), @@ -134,26 +145,80 @@ mod tests { &mut open_retention_db(&scope.db_path).unwrap(), &scope, vec![], + || nostr::Timestamp::from(1000), ) .unwrap(); let mut reopened = open_retention_db(&scope.db_path).unwrap(); assert_eq!( - prepare_report(&mut reopened, &scope, vec![]).unwrap(), + prepare_report(&mut reopened, &scope, vec![], || panic!( + "unchanged must not sign" + )) + .unwrap(), first ); assert_eq!(reopened.total_changes(), 0); - let facts = vec![RuntimeFact { + let mut facts = vec![RuntimeFact { id: "goose".into(), availability: "available".into(), requires_external_cli: true, max_parallelism: None, }]; - let changed = prepare_report(&mut reopened, &scope, facts).unwrap(); - assert_ne!(changed.id, first.id); + for now in [1000, 990, 999, 1000] { + let error = prepare_report(&mut reopened, &scope, facts.clone(), || { + nostr::Timestamp::from(now) + }) + .unwrap_err(); + assert!(error.contains("clock advances")); + assert_eq!(reopened.total_changes(), 0, "deferral must not persist"); + // Returning to old facts cancels the proposed change, even after a + // restart/rollback: no deferred payload or timestamp renewal survives. + assert_eq!( + prepare_report(&mut reopened, &scope, vec![], || panic!("exact retry")).unwrap(), + first + ); + reopened = open_retention_db(&scope.db_path).unwrap(); + } + // The retry observes today's facts, not the projection first deferred. + facts[0].availability = "cli_missing".into(); + let changed = prepare_report(&mut reopened, &scope, facts.clone(), || { + nostr::Timestamp::from(1001) + }) + .unwrap(); + first.verify().unwrap(); + changed.verify().unwrap(); + assert_eq!(changed.created_at.as_secs(), 1001, "no future timestamp"); + assert!(changed.created_at > first.created_at); assert_eq!(changed.tags, first.tags); + for events in [ + vec![first.clone(), changed.clone()], + vec![changed.clone(), first], + ] { + let rows = + DesktopCapabilities::read_latest(events, &scope.owner_keys, &scope.relay_url) + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].0.runtimes, facts); + assert_eq!(rows[0].1, 1001); + } + let mut reopened = open_retention_db(&scope.db_path).unwrap(); + let mut invalid = facts.clone(); + invalid[0].max_parallelism = Some(0); + assert!(prepare_report(&mut reopened, &scope, invalid, || { + nostr::Timestamp::from(1002) + }) + .is_err()); + assert_eq!( + prepare_report(&mut reopened, &scope, facts, || panic!("exact retry")).unwrap(), + changed + ); + assert_eq!( + reopened.total_changes(), + 0, + "failed signing must not persist" + ); reopened .execute("UPDATE desktop_capabilities SET raw = 'corrupt'", []) .unwrap(); - assert!(prepare_report(&mut reopened, &scope, vec![]).is_err()); + assert!(prepare_report(&mut reopened, &scope, vec![], nostr::Timestamp::now).is_err()); } } diff --git a/desktop/src/features/agents/desktopCapabilities.test.mjs b/desktop/src/features/agents/desktopCapabilities.test.mjs index 9def2ec43d..da09f6bb4a 100644 --- a/desktop/src/features/agents/desktopCapabilities.test.mjs +++ b/desktop/src/features/agents/desktopCapabilities.test.mjs @@ -80,6 +80,34 @@ test("unchanged accepted report does not republish; failed publish retries exact await assert.rejects(f.refresh(), /unavailable/); }); +test("deferred preparation settles with prior relay facts, never publishes, and honors cancellation", async () => { + const f = fixture(); + const ipc = f.ipc; + let active = true; + let cancel = false; + f.ipc = async (command, args) => { + if (command === "prepare_desktop_capabilities") { + if (cancel) active = false; + throw Error("Desktop capability facts deferred until the clock advances"); + } + return ipc(command, args); + }; + const deferred = await f.refresh(() => active); + assert.deepEqual(deferred.rows, [row]); + assert.match(deferred.warning, /Will retry/); + assert.ok(!f.calls.includes(event)); + cancel = true; + await assert.rejects( + f.refresh(() => active), + /scope changed/, + ); + assert.ok(!f.calls.includes(event)); + // A later, active attempt prepares afresh; no held promise or queued event. + f.ipc = ipc; + assert.equal((await f.refresh()).warning, ""); + assert.equal(f.calls.filter((c) => c === event).length, 1); +}); + test("all async boundaries fence cancellation, account/community switches and late ACK", async () => { for (const boundary of ["prepare", "read", "fetch", "transport", "ack"]) { const f = fixture(boundary); diff --git a/desktop/src/features/agents/desktopList.test.mjs b/desktop/src/features/agents/desktopList.test.mjs index 6f7330d978..5e21adbb31 100644 --- a/desktop/src/features/agents/desktopList.test.mjs +++ b/desktop/src/features/agents/desktopList.test.mjs @@ -202,6 +202,9 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa const originalReconnect = relayClient.subscribeToReconnects; let reconnect; let pulses = 0; + let reports = 0; + let publishedReports = 0; + let deferReport = true; relayClient.subscribeToReconnects = (callback) => { reconnect = callback; return () => { @@ -210,6 +213,17 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa }; window.__TAURI_INTERNALS__ = { invoke: async (command, args) => { + if (command === "prepare_desktop_capabilities") { + reports++; + if (deferReport) throw Error("clock has not advanced"); + return { event: { ...first, kind: 30182 } }; + } + if (command === "read_desktop_capabilities") + return args.events.map(() => ({ + id: "desktop-a", + reported: 100, + runtimes: [], + })); if (command === "prepare_desktop_observation") return { event: { ...first, kind: 30181 } }; if (command === "read_desktop_observations") @@ -237,7 +251,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 ([30181, 30182].includes(filter.kinds[0])) return []; if (filter["#d"]) return [current]; const rows = [current]; if (hold) { @@ -249,6 +263,10 @@ test("mounted cache clears both scopes, fences late reads and retains rows on fa return rows; }; relayClient.publishEvent = async (event) => { + if (event.kind === 30182) { + publishedReports++; + return; + } assert.equal(event.kind, 30181, "no profile heartbeat rewrite"); pulses++; }; @@ -285,14 +303,28 @@ 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/); + assert.match(text(), /could not synchronize capability facts/); const beforeReconnect = pulses; + const reportsBeforeReconnect = reports; await React.act(async () => reconnect()); await settle(); assert.ok(pulses > beforeReconnect, "reconnect reports a fresh pulse"); + assert.ok( + reports > reportsBeforeReconnect, + "reconnect retries deferred facts", + ); const beforeTimer = pulses; + const reportsBeforeTimer = reports; await React.act(async () => t.mock.timers.tick(60_000)); await settle(); assert.ok(pulses > beforeTimer, "bounded periodic publisher runs"); + assert.ok(reports > reportsBeforeTimer, "periodic retry survives deferral"); + assert.equal(publishedReports, 0, "deferred facts are not published"); + deferReport = false; + await React.act(async () => t.mock.timers.tick(60_000)); + await settle(); + assert.equal(publishedReports, 1, "later preparation is published"); + assert.doesNotMatch(text(), /could not synchronize capability facts/); hold = true; await React.act(async () => { void client.refetchQueries({ queryKey: ["desktop-profiles"] });