fix(desktop): defer capability changes until real time advances

Signed-off-by: Logan Johnson <loganj@squareup.com>

Author: loganj

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<Event, String> {
+        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<Event, String> {
         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)?;

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<RuntimeFact>,
+    clock: impl FnOnce() -> nostr::Timestamp,
 ) -> Result<Event, String> {
     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());
     }
 }

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);

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