feat(buzz-acp): gate prompting on client-side leader check

When multiple Buzz dev instances share an agent keypair, the relay fans
each mention out to every connected buzz-acp (NIP-01) and all of them
respond. Add a per-agent-key leader lock so only the leader instance
promotes queued events to prompts; non-leaders keep their queue
populated for UI rendering but stay silent.

Reads ~/.buzz/leader-locks/<pubkey>.lock: absent or self-owned means
leader, foreign-owned means observer, malformed fails safe to leader so
a corrupt lock never silences the only responder. Solo dev has no lock
file, so behavior is unchanged. The 👀 reaction fires at queue-push time
before dispatch, so it is gated separately from dispatch_pending.

The election identity is sourced from BUZZ_INSTANCE_ELECTION_ID behind a
single named constant. It is deliberately NOT the Tauri bundle
identifier (BUZZ_MANAGED_AGENT): that collides across same-class windows
(DMG + dev, or worktrees whose icon-gen fell back to the shared dev id),
which would let two windows both lead. Phase 2 writes a process-unique
value; Phase 1 reads whatever that env carries and defaults to leader
when it is unset.

Phase 1 of 3: read side only. Lock acquire/steal/failover (Phase 2) and
claim UI (Phase 3) follow.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

crates/buzz-acp/src/leader.rs

@@ -0,0 +1,254 @@
+//! Client-side leader election (read side).
+//!
+//! When multiple Buzz instances share the same agent keypair, the relay fans
+//! every matching event out to all of them (NIP-01). Without coordination,
+//! each instance would prompt its agent and respond — duplicate replies for a
+//! single mention. A per-agent-key *leader lock* designates exactly one
+//! instance as the active responder; non-leaders still receive and render
+//! events but suppress the prompt path (and the pre-dispatch `👀` side-effect).
+//!
+//! This module owns only the **read** half: given a lock file written by some
+//! instance, decide whether *this* process is the leader for a given agent
+//! pubkey. Lock acquisition, stealing, and failover live elsewhere (Phase 2).
+//!
+//! # Lock contract
+//!
+//! - Lock dir: `~/.buzz/leader-locks/`, one file per agent pubkey:
+//!   `<pubkey-hex>.lock`, JSON `{"instance_id","pid","claimed_at"}`.
+//! - **Absent** lock file → this process is leader. Single-instance dev is
+//!   thereby unaffected: no lock, no suppression.
+//! - **Present** → leader iff the lock's `instance_id` equals this process's
+//!   own election id ([`ELECTION_ID_ENV`]).
+//! - **No instance id** (env unset, e.g. solo CLI use or pre-Phase-2 where
+//!   nothing writes the election id) → leader. There is no coordinating
+//!   desktop, so there is nothing to defer to.
+//! - **Malformed** lock file → fail safe to leader. A corrupt lock must never
+//!   silence the only responder.
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Mutex;
+
+use serde::Deserialize;
+
+/// Environment variable carrying this window's leader-election identity.
+///
+/// Per-window election identity; Phase 2 supplies a process-unique value —
+/// the Tauri bundle identifier is NOT sufficient (it collides across
+/// same-class windows like DMG + dev, or worktrees whose icon-gen fell back
+/// to the shared dev id). Kept behind this constant so Phase 2 can swap the
+/// source without touching the gate. Distinct from `BUZZ_MANAGED_AGENT`
+/// (reaper identity): reaper identity partitions by app-class, election
+/// identity must be unique per window — opposite uniqueness requirements.
+const ELECTION_ID_ENV: &str = "BUZZ_INSTANCE_ELECTION_ID";
+
+/// Decides whether this process should act on events for a given agent key.
+pub trait LeaderCheck: Send + Sync {
+    /// Whether this process is the leader for `agent_pubkey_hex`.
+    ///
+    /// Reads through the cache on first sight of a key so the very first
+    /// dispatch is correct without waiting for a refresh tick; subsequent
+    /// calls are served from cache and updated by [`LeaderCheck::refresh`].
+    fn is_leader(&self, agent_pubkey_hex: &str) -> bool;
+
+    /// Re-read all known lock files and update cached status. Called on a
+    /// fixed cadence by the event loop so leadership changes take effect
+    /// without a restart.
+    fn refresh(&self);
+}
+
+/// On-disk lock file shape. Only `instance_id` is load-bearing for the read
+/// side; `pid` and `claimed_at` are written by the (Phase 2) acquire path and
+/// ignored here.
+#[derive(Deserialize)]
+struct LockFile {
+    instance_id: String,
+}
+
+/// Filesystem-backed [`LeaderCheck`] over `~/.buzz/leader-locks/`.
+pub struct FileLeaderCheck {
+    /// This process's election id ([`ELECTION_ID_ENV`]). `None` when unset —
+    /// solo CLI use or the pre-Phase-2 regime, where this process is always
+    /// leader.
+    instance_id: Option<String>,
+    lock_dir: PathBuf,
+    /// Cached leader status per agent pubkey hex. Seeded read-through on first
+    /// `is_leader`, refreshed in place by `refresh`.
+    cache: Mutex<HashMap<String, bool>>,
+}
+
+impl FileLeaderCheck {
+    /// Build from the ambient environment: election id from
+    /// [`ELECTION_ID_ENV`], lock dir under `$HOME/.buzz/leader-locks/`.
+    pub fn from_env() -> Self {
+        let lock_dir = std::env::var_os("HOME")
+            .map(PathBuf::from)
+            .unwrap_or_default()
+            .join(".buzz")
+            .join("leader-locks");
+        Self::new(std::env::var(ELECTION_ID_ENV).ok(), lock_dir)
+    }
+
+    fn new(instance_id: Option<String>, lock_dir: PathBuf) -> Self {
+        Self {
+            instance_id,
+            lock_dir,
+            cache: Mutex::new(HashMap::new()),
+        }
+    }
+
+    /// Read the lock for `pubkey_hex` and compute leadership per the contract.
+    fn read_status(&self, pubkey_hex: &str) -> bool {
+        // No coordinating instance id → nothing to defer to.
+        let Some(self_id) = self.instance_id.as_deref() else {
+            return true;
+        };
+        let path = self.lock_dir.join(format!("{pubkey_hex}.lock"));
+        let contents = match std::fs::read_to_string(&path) {
+            Ok(c) => c,
+            // Absent (or unreadable) lock → leader.
+            Err(_) => return true,
+        };
+        match serde_json::from_str::<LockFile>(&contents) {
+            Ok(lock) => lock.instance_id == self_id,
+            // Malformed lock → fail safe to leader.
+            Err(_) => true,
+        }
+    }
+}
+
+impl LeaderCheck for FileLeaderCheck {
+    fn is_leader(&self, agent_pubkey_hex: &str) -> bool {
+        if let Some(&cached) = self.cache.lock().unwrap().get(agent_pubkey_hex) {
+            return cached;
+        }
+        let status = self.read_status(agent_pubkey_hex);
+        self.cache
+            .lock()
+            .unwrap()
+            .insert(agent_pubkey_hex.to_string(), status);
+        status
+    }
+
+    fn refresh(&self) {
+        // Re-read only keys we've already been asked about — those are the
+        // agent pubkeys this process actually dispatches for.
+        let keys: Vec<String> = self.cache.lock().unwrap().keys().cloned().collect();
+        for key in keys {
+            let status = self.read_status(&key);
+            self.cache.lock().unwrap().insert(key, status);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicU32, Ordering};
+
+    const PUBKEY: &str = "abc123";
+    // Opaque per-window election ids — deliberately NOT bundle-class strings.
+    // The leader-election identity must be unique per window; baking in
+    // `bundle-id == election-id` would mask the same-class collision Phase 2
+    // must avoid.
+    const SELF_ID: &str = "window-a";
+    const OTHER_ID: &str = "window-b";
+
+    /// Unique scratch dir per test, removed on drop. Avoids a dev-dep on
+    /// `tempfile` for four file reads.
+    struct TmpDir(PathBuf);
+
+    impl TmpDir {
+        fn new() -> Self {
+            static N: AtomicU32 = AtomicU32::new(0);
+            let dir = std::env::temp_dir().join(format!(
+                "buzz-acp-leader-{}-{}",
+                std::process::id(),
+                N.fetch_add(1, Ordering::Relaxed),
+            ));
+            std::fs::create_dir_all(&dir).unwrap();
+            Self(dir)
+        }
+
+        fn lock_path(&self) -> PathBuf {
+            self.0.join(format!("{PUBKEY}.lock"))
+        }
+
+        fn write_lock(&self, contents: &str) {
+            std::fs::write(self.lock_path(), contents).unwrap();
+        }
+    }
+
+    impl Drop for TmpDir {
+        fn drop(&mut self) {
+            let _ = std::fs::remove_dir_all(&self.0);
+        }
+    }
+
+    #[test]
+    fn test_absent_lock_file_is_leader() {
+        let dir = TmpDir::new();
+        let lc = FileLeaderCheck::new(Some(SELF_ID.into()), dir.0.clone());
+        assert!(lc.is_leader(PUBKEY));
+    }
+
+    #[test]
+    fn test_lock_matching_own_instance_is_leader() {
+        let dir = TmpDir::new();
+        dir.write_lock(&format!(
+            r#"{{"instance_id":"{SELF_ID}","pid":123,"claimed_at":"2026-06-15T00:00:00Z"}}"#
+        ));
+        let lc = FileLeaderCheck::new(Some(SELF_ID.into()), dir.0.clone());
+        assert!(lc.is_leader(PUBKEY));
+    }
+
+    #[test]
+    fn test_lock_naming_other_instance_is_observer() {
+        let dir = TmpDir::new();
+        dir.write_lock(&format!(
+            r#"{{"instance_id":"{OTHER_ID}","pid":123,"claimed_at":"2026-06-15T00:00:00Z"}}"#
+        ));
+        let lc = FileLeaderCheck::new(Some(SELF_ID.into()), dir.0.clone());
+        assert!(!lc.is_leader(PUBKEY));
+    }
+
+    #[test]
+    fn test_malformed_lock_fails_safe_to_leader() {
+        let dir = TmpDir::new();
+        dir.write_lock("{ this is not json ");
+        let lc = FileLeaderCheck::new(Some(SELF_ID.into()), dir.0.clone());
+        assert!(lc.is_leader(PUBKEY));
+    }
+
+    #[test]
+    fn test_no_instance_id_is_leader_even_with_foreign_lock() {
+        let dir = TmpDir::new();
+        dir.write_lock(&format!(
+            r#"{{"instance_id":"{OTHER_ID}","pid":123,"claimed_at":"2026-06-15T00:00:00Z"}}"#
+        ));
+        let lc = FileLeaderCheck::new(None, dir.0.clone());
+        assert!(lc.is_leader(PUBKEY));
+    }
+
+    #[test]
+    fn test_refresh_flips_status_when_lock_changes() {
+        let dir = TmpDir::new();
+        let lc = FileLeaderCheck::new(Some(SELF_ID.into()), dir.0.clone());
+
+        // No lock yet → leader, and the key is now cached.
+        assert!(lc.is_leader(PUBKEY));
+
+        // A foreign instance claims the lock.
+        dir.write_lock(&format!(r#"{{"instance_id":"{OTHER_ID}"}}"#));
+        // Cache still says leader until refresh.
+        assert!(lc.is_leader(PUBKEY));
+
+        lc.refresh();
+        assert!(!lc.is_leader(PUBKEY));
+
+        // Lock removed (leader stepped down) → back to leader after refresh.
+        std::fs::remove_file(dir.lock_path()).unwrap();
+        lc.refresh();
+        assert!(lc.is_leader(PUBKEY));
+    }
+}

crates/buzz-acp/src/lib.rs

@@ -4,6 +4,7 @@ mod acp;
 mod config;
 mod engram_fetch;
 mod filter;
+mod leader;
 mod observer;
 mod pool;
 mod queue;
@@ -1295,6 +1296,7 @@ async fn tokio_main() -> Result<()> {
             .as_deref()
             .and_then(|hex| nostr::PublicKey::from_hex(hex).ok()),
         memory_enabled: config.memory_enabled,
+        leader: Arc::new(leader::FileLeaderCheck::from_env()),
     });
 
     if !config.memory_enabled {
@@ -1340,6 +1342,13 @@ async fn tokio_main() -> Result<()> {
     let mut typing_channels: HashMap<Uuid, ThreadTags> = HashMap::new();
     let mut presence_task: Option<tokio::task::JoinHandle<()>> = None;
 
+    // ── Step 6c2: Leader-status refresh timer ────────────────────────────────
+    // Re-reads the per-key leader lock every 5s so a leadership change (another
+    // instance claiming/releasing this agent) takes effect without a restart.
+    // Faster cadence than the 30s maintenance tick because failover latency is
+    // user-visible. Cheap: one file read per known agent key.
+    let mut leader_refresh = tokio::time::interval(Duration::from_secs(5));
+
     // ── Step 6d: Maintenance (slot refill + queue compaction) ────────────────
     // Runs at the TOP of every loop iteration via Instant check — cannot be
     // starved by the biased select. Slot refill spawns background tasks so
@@ -1828,11 +1837,15 @@ async fn tokio_main() -> Result<()> {
                                 prompt_tag,
                             });
                             // 👀 — immediate "seen" reaction, only if the event
-                            // was actually queued (not dropped by DedupMode::Drop).
+                            // was actually queued (not dropped by DedupMode::Drop)
+                            // AND this instance is the leader for this agent key.
+                            // The reaction fires at queue-push time, BEFORE
+                            // dispatch, so the dispatch gate alone would let
+                            // every sharing instance emit a redundant 👀.
                             // Fire-and-forget: on rare fast-failure paths the
                             // guard's cleanup may race with this add, leaving a
                             // cosmetic stale 👀. Acceptable — see ReactionGuard docs.
-                            if accepted {
+                            if accepted && ctx.leader.is_leader(&pubkey_hex) {
                                 let rc = ctx.rest_client.clone();
                                 tokio::spawn(async move {
                                     pool::reaction_add(&rc, &event_id_hex, "👀").await;
@@ -1942,6 +1955,14 @@ async fn tokio_main() -> Result<()> {
                     }
                     None
                 }
+                _ = leader_refresh.tick() => {
+                    let _ = result_rx;
+                    // Re-read leader lock(s) so claim/release/failover by
+                    // another instance flips this process between active and
+                    // observer without a restart.
+                    ctx.leader.refresh();
+                    None
+                }
                 _ = shutdown_rx.changed() => {
                     tracing::info!("shutting down");
                     break;
@@ -2177,6 +2198,14 @@ fn dispatch_pending(
     queue: &mut EventQueue,
     ctx: &Arc<PromptContext>,
 ) -> Vec<(Uuid, ThreadTags)> {
+    // Multi-instance leader gate: when this agent's keypair is shared across
+    // several Buzz instances, only the leader promotes queued events to
+    // prompts. Non-leaders keep the queue populated (so the UI still renders
+    // the conversation) but never dispatch. Solo dev has no lock file, so this
+    // is unconditionally `true` and behavior is unchanged.
+    if !ctx.leader.is_leader(&ctx.agent_keys.public_key().to_hex()) {
+        return Vec::new();
+    }
     let mut dispatched_channels = Vec::new();
     loop {
         let batch = match queue.flush_next() {

crates/buzz-acp/src/pool.rs

@@ -33,6 +33,7 @@ use crate::acp::{
     AcpError, McpServer, ModelSwitchMethod, StopReason,
 };
 use crate::config::{DedupMode, PermissionMode};
+use crate::leader::LeaderCheck;
 use crate::observer;
 use crate::queue::{
     ContextMessage, ConversationContext, FlushBatch, PromptChannelInfo, PromptProfile,
@@ -253,6 +254,11 @@ pub struct PromptContext {
     /// `[Agent Memory — core]` section. On by default; disabled via
     /// `--no-memory` / `BUZZ_ACP_NO_MEMORY`.
     pub memory_enabled: bool,
+    /// Multi-instance leader gate. When several Buzz instances share this
+    /// agent's keypair, only the leader promotes queued events to prompts and
+    /// emits the pre-dispatch `👀` reaction; non-leaders observe silently.
+    /// Defaults to "always leader" when no lock file exists (solo dev).
+    pub leader: Arc<dyn LeaderCheck>,
 }
 
 // ── AgentPool impl ────────────────────────────────────────────────────────────