perf(desktop): publish mention sends before waking agents

crates/buzz-acp/src/config.rs

@@ -516,6 +516,15 @@ pub struct CliArgs {
     /// Requires `--lazy-pool`; ignored otherwise. 0 disables idle re-sleep.
     #[arg(long, env = "BUZZ_ACP_IDLE_POOL_SLEEP", default_value_t = 0)]
     pub idle_pool_sleep: u64,
+
+    /// Unix-seconds replay floor for the startup watermark. A publish-first
+    /// mention send publishes the triggering message and then spawns this
+    /// harness, passing the send timestamp here so the first REQ replays past
+    /// that message however long the spawn takes. Floors older than 15 minutes
+    /// are clamped to 15 minutes before startup; floors in the future are
+    /// ignored (the watermark stays at startup time).
+    #[arg(long, env = "BUZZ_ACP_REPLAY_FLOOR")]
+    pub replay_floor: Option<u64>,
 }
 
 /// Merged NIP-01 subscription filter for a single channel.
@@ -605,6 +614,12 @@ pub struct Config {
     /// woken lazy pool is torn back down to the empty-slot state. 0 = disabled.
     /// Only meaningful when `lazy_pool` is true.
     pub idle_pool_sleep_secs: u64,
+    /// Optional unix-seconds replay floor for the startup watermark
+    /// (`--replay-floor` / `BUZZ_ACP_REPLAY_FLOOR`), set by a publish-first
+    /// mention send so the first REQ replays past the already-published
+    /// triggering message. Clamped where consumed — see
+    /// `startup_watermark_with_floor`.
+    pub replay_floor_unix: Option<u64>,
     /// Agent owner pubkey (hex). Used for `--respond-to=owner-only` gate.
     /// Replaces the old REST-based owner lookup.
     pub agent_owner: Option<String>,
@@ -1185,6 +1200,7 @@ impl Config {
             exit_after_inactivity_secs: args.exit_after_inactivity,
             lazy_pool: args.lazy_pool,
             idle_pool_sleep_secs: args.idle_pool_sleep,
+            replay_floor_unix: args.replay_floor,
             agent_owner: args.agent_owner.map(|s| s.trim().to_ascii_lowercase()),
             no_base_prompt: args.no_base_prompt,
             base_prompt_content,
@@ -1560,6 +1576,7 @@ mod tests {
             exit_after_inactivity_secs: 0,
             lazy_pool: false,
             idle_pool_sleep_secs: 0,
+            replay_floor_unix: None,
             agent_owner: None,
             no_base_prompt: false,
             base_prompt_content: None,

crates/buzz-acp/src/lib.rs

@@ -2387,6 +2387,63 @@ mod idle_pool_sleep_tests {
     }
 }
 
+/// Oldest a caller-supplied replay floor may reach back from startup. Bounds
+/// the stale-event burst when a spawn request sat around (e.g. the desktop
+/// slept between the send and this spawn actually running).
+const REPLAY_FLOOR_MAX_AGE_SECS: u64 = 15 * 60;
+
+/// Resolve the startup watermark from process-start time and an optional
+/// replay floor (`--replay-floor` / `BUZZ_ACP_REPLAY_FLOOR`).
+///
+/// A publish-first mention send publishes the triggering message BEFORE this
+/// harness spawns, so the watermark must reach back to the send timestamp for
+/// the first REQ (`since = watermark − 5s`) to replay that message. Floors
+/// older than [`REPLAY_FLOOR_MAX_AGE_SECS`] clamp to that bound; floors in
+/// the future clamp to `now` (a skewed sender must not push the watermark
+/// forward past startup and re-open the blind spot the watermark closes).
+fn startup_watermark_with_floor(now_unix: u64, replay_floor: Option<u64>) -> u64 {
+    match replay_floor {
+        Some(floor) => floor.clamp(now_unix.saturating_sub(REPLAY_FLOOR_MAX_AGE_SECS), now_unix),
+        None => now_unix,
+    }
+}
+
+#[cfg(test)]
+mod replay_floor_tests {
+    use super::{startup_watermark_with_floor, REPLAY_FLOOR_MAX_AGE_SECS};
+
+    const NOW: u64 = 1_700_000_000;
+
+    #[test]
+    fn no_floor_keeps_startup_time() {
+        assert_eq!(startup_watermark_with_floor(NOW, None), NOW);
+    }
+
+    #[test]
+    fn recent_floor_moves_watermark_back_to_the_send_timestamp() {
+        // The publish-first case: message sent 4s before the harness booted.
+        assert_eq!(startup_watermark_with_floor(NOW, Some(NOW - 4)), NOW - 4);
+    }
+
+    #[test]
+    fn stale_floor_clamps_to_the_max_age_bound() {
+        assert_eq!(
+            startup_watermark_with_floor(NOW, Some(NOW - REPLAY_FLOOR_MAX_AGE_SECS - 1)),
+            NOW - REPLAY_FLOOR_MAX_AGE_SECS
+        );
+    }
+
+    #[test]
+    fn future_floor_is_ignored() {
+        assert_eq!(startup_watermark_with_floor(NOW, Some(NOW + 60)), NOW);
+    }
+
+    #[test]
+    fn early_epoch_now_does_not_underflow() {
+        assert_eq!(startup_watermark_with_floor(10, Some(0)), 0);
+    }
+}
+
 pub fn run() -> Result<()> {
     config::propagate_legacy_env_vars();
     tokio_main()
@@ -2484,10 +2541,24 @@ async fn tokio_main() -> Result<()> {
     // the initial subscribe_since for channels discovered at startup. The Subscribe
     // handler falls back to subscribe_since when last_seen is None, closing the
     // blind spot between "agents ready" and "first REQ sent".
-    let startup_watermark: u64 = std::time::SystemTime::now()
+    //
+    // A publish-first mention send passes the triggering message's send
+    // timestamp as a replay floor (`--replay-floor` / `BUZZ_ACP_REPLAY_FLOOR`):
+    // the message is already on the relay when this process spawns, so the
+    // watermark must reach back to it for the first REQ to replay it — however
+    // long the spawn took.
+    let now_unix: u64 = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .unwrap_or_default()
         .as_secs();
+    let startup_watermark = startup_watermark_with_floor(now_unix, config.replay_floor_unix);
+    if let Some(floor) = config.replay_floor_unix {
+        tracing::info!(
+            floor,
+            startup_watermark,
+            "applying replay floor to startup watermark"
+        );
+    }
 
     let pubkey_hex = config.keys.public_key().to_hex();
 
@@ -8862,6 +8933,7 @@ mod build_mcp_servers_tests {
             exit_after_inactivity_secs: 0,
             lazy_pool: false,
             idle_pool_sleep_secs: 0,
+            replay_floor_unix: None,
             agent_owner: None,
             no_base_prompt: false,
             base_prompt_content: None,
@@ -9087,6 +9159,7 @@ mod error_outcome_emission_tests {
             exit_after_inactivity_secs: 0,
             lazy_pool: false,
             idle_pool_sleep_secs: 0,
+            replay_floor_unix: None,
             agent_owner: None,
             no_base_prompt: false,
             base_prompt_content: None,

desktop/src-tauri/src/app_state.rs

@@ -129,6 +129,15 @@ pub struct AppState {
     /// bounded and letting a later leave correctly flip the channel back to
     /// `is_member=false`.
     pub pending_owned_channels: Mutex<std::collections::HashSet<(String, String)>>,
+    /// NIP-11 `self` pubkeys keyed by relay WS URL, each with its fetch
+    /// instant. A relay's signing identity is effectively static, yet every
+    /// send-time agent revalidation used to re-GET the document — one of the
+    /// dominant costs of agent-mention send latency. Entries expire after
+    /// `identity_archive::RELAY_SELF_CACHE_TTL` so a relay-side key rotation
+    /// still converges. Keyed by URL, so switching communities can never serve
+    /// another relay's identity; only verified `Some` values are stored (an
+    /// outage or a document without `self` must stay retryable).
+    pub relay_self_cache: Mutex<HashMap<String, (std::time::Instant, String)>>,
     pub archive_db: crate::archive::ArchiveDb,
 }
 
@@ -231,86 +240,13 @@ pub fn build_app_state() -> AppState {
         #[cfg(feature = "mesh-llm")]
         mesh_coordinator: AsyncMutex::new(None),
         pending_owned_channels: Mutex::new(std::collections::HashSet::new()),
+        relay_self_cache: Mutex::new(HashMap::new()),
         archive_db: crate::archive::ArchiveDb::default(),
     }
 }
 
-impl AppState {
-    /// Lock the huddle state mutex, converting a poisoned-lock error to a String.
-    ///
-    /// Convenience wrapper — replaces 15+ instances of
-    /// `state.huddle_state.lock().map_err(|e| e.to_string())?` throughout the
-    /// huddle module.
-    pub fn huddle(&self) -> Result<std::sync::MutexGuard<'_, crate::huddle::HuddleState>, String> {
-        self.huddle_state.lock().map_err(|e| e.to_string())
-    }
-
-    pub fn get_session_cache(&self, key: &ManagedAgentRuntimeKey) -> Option<SessionConfigCache> {
-        self.session_config_cache.lock().ok()?.get(key).cloned()
-    }
-
-    pub fn put_session_cache(&self, key: ManagedAgentRuntimeKey, cache: SessionConfigCache) {
-        if let Ok(mut map) = self.session_config_cache.lock() {
-            map.insert(key, cache);
-        }
-    }
-
-    pub fn clear_agent_session_cache(&self, key: &ManagedAgentRuntimeKey) {
-        if let Ok(mut map) = self.session_config_cache.lock() {
-            map.remove(key);
-        }
-    }
-
-    pub fn clear_agent_session_caches(&self, pubkey: &str) {
-        if let Ok(mut map) = self.session_config_cache.lock() {
-            map.retain(|key, _| key.pubkey != pubkey);
-        }
-    }
-
-    /// Return the active identity keys if they are in a signable state.
-    ///
-    /// Returns `Err` when the identity is in a lost state (`identity_lost`
-    /// — ephemeral key, user must re-import their nsec) or when the keyring
-    /// is locked (`keyring_locked` — key is held in a keyring that is
-    /// unavailable this boot). All signing and publish commands must call
-    /// this instead of locking `state.keys` directly, so that recovery mode
-    /// blocks publishing under an invalid or inaccessible identity.
-    pub fn signing_keys(&self) -> Result<Keys, String> {
-        if self
-            .identity_lost
-            .load(std::sync::atomic::Ordering::Acquire)
-            || self
-                .keyring_locked
-                .load(std::sync::atomic::Ordering::Acquire)
-        {
-            return Err("identity is in recovery mode; event signing is disabled \
-                 until the identity is restored and Buzz is relaunched"
-                .to_string());
-        }
-        self.keys
-            .lock()
-            .map_err(|e| e.to_string())
-            .map(|k| k.clone())
-    }
-
-    /// Emit the current huddle state to the frontend via Tauri event.
-    ///
-    /// Acquires both locks (app_handle + huddle_state), clones a snapshot,
-    /// releases both, then emits. Best-effort — no-op if either lock is
-    /// poisoned or the app_handle hasn't been set yet.
-    pub fn emit_huddle_state_changed(&self) {
-        let app = match self.app_handle.lock() {
-            Ok(guard) => guard.clone(),
-            Err(_) => return,
-        };
-        let Some(app) = app else { return };
-        let snapshot = match self.huddle_state.lock() {
-            Ok(hs) => hs.clone(),
-            Err(_) => return,
-        };
-        crate::huddle::state::emit_huddle_state(&app, &snapshot);
-    }
-}
+#[path = "app_state_accessors.rs"]
+mod accessors;
 
 /// Resolve the user's identity key from the app data directory and wire
 /// the resulting [`RecoveryState`] into `AppState`.

desktop/src-tauri/src/app_state_accessors.rs

@@ -0,0 +1,87 @@
+//! Convenience accessors over [`AppState`]'s lock-guarded fields.
+//!
+//! Kept apart from `app_state.rs`, which owns the struct, its builder, and the
+//! identity-key resolution that populates it.
+
+use nostr::Keys;
+
+use crate::app_state::AppState;
+use crate::managed_agents::config_bridge::SessionConfigCache;
+use crate::managed_agents::ManagedAgentRuntimeKey;
+
+impl AppState {
+    /// Lock the huddle state mutex, converting a poisoned-lock error to a String.
+    ///
+    /// Convenience wrapper — replaces 15+ instances of
+    /// `state.huddle_state.lock().map_err(|e| e.to_string())?` throughout the
+    /// huddle module.
+    pub fn huddle(&self) -> Result<std::sync::MutexGuard<'_, crate::huddle::HuddleState>, String> {
+        self.huddle_state.lock().map_err(|e| e.to_string())
+    }
+
+    pub fn get_session_cache(&self, key: &ManagedAgentRuntimeKey) -> Option<SessionConfigCache> {
+        self.session_config_cache.lock().ok()?.get(key).cloned()
+    }
+
+    pub fn put_session_cache(&self, key: ManagedAgentRuntimeKey, cache: SessionConfigCache) {
+        if let Ok(mut map) = self.session_config_cache.lock() {
+            map.insert(key, cache);
+        }
+    }
+
+    pub fn clear_agent_session_cache(&self, key: &ManagedAgentRuntimeKey) {
+        if let Ok(mut map) = self.session_config_cache.lock() {
+            map.remove(key);
+        }
+    }
+
+    pub fn clear_agent_session_caches(&self, pubkey: &str) {
+        if let Ok(mut map) = self.session_config_cache.lock() {
+            map.retain(|key, _| key.pubkey != pubkey);
+        }
+    }
+
+    /// Return the active identity keys if they are in a signable state.
+    ///
+    /// Returns `Err` when the identity is in a lost state (`identity_lost`
+    /// — ephemeral key, user must re-import their nsec) or when the keyring
+    /// is locked (`keyring_locked` — key is held in a keyring that is
+    /// unavailable this boot). All signing and publish commands must call
+    /// this instead of locking `state.keys` directly, so that recovery mode
+    /// blocks publishing under an invalid or inaccessible identity.
+    pub fn signing_keys(&self) -> Result<Keys, String> {
+        if self
+            .identity_lost
+            .load(std::sync::atomic::Ordering::Acquire)
+            || self
+                .keyring_locked
+                .load(std::sync::atomic::Ordering::Acquire)
+        {
+            return Err("identity is in recovery mode; event signing is disabled \
+                 until the identity is restored and Buzz is relaunched"
+                .to_string());
+        }
+        self.keys
+            .lock()
+            .map_err(|e| e.to_string())
+            .map(|k| k.clone())
+    }
+
+    /// Emit the current huddle state to the frontend via Tauri event.
+    ///
+    /// Acquires both locks (app_handle + huddle_state), clones a snapshot,
+    /// releases both, then emits. Best-effort — no-op if either lock is
+    /// poisoned or the app_handle hasn't been set yet.
+    pub fn emit_huddle_state_changed(&self) {
+        let app = match self.app_handle.lock() {
+            Ok(guard) => guard.clone(),
+            Err(_) => return,
+        };
+        let Some(app) = app else { return };
+        let snapshot = match self.huddle_state.lock() {
+            Ok(hs) => hs.clone(),
+            Err(_) => return,
+        };
+        crate::huddle::state::emit_huddle_state(&app, &snapshot);
+    }
+}