fix(buzz-acp): tell a refused author instead of dropping them silently

README.md

@@ -1,3 +1,19 @@
+> **This is Dreamforge**, a fork of [`block/buzz`](https://github.com/block/buzz).
+>
+> The product ships as Dreamforge and every screen says so. Everything below the
+> surface deliberately keeps upstream's names — crates stay `buzz-*`, environment
+> variables stay `BUZZ_*`, and on-the-wire tags (`buzz-channel`, `buzz-protect`,
+> `buzz-visibility`) and event kinds are untouched.
+>
+> That split is intentional. Renaming an environment variable would make upstream
+> code read one nobody sets — it compiles, runs, and silently takes a default.
+> Renaming a wire tag would diverge the protocol with no error at all. Both fail
+> quietly, and quiet failure is the thing this fork exists to remove. Renaming
+> what people read costs nothing, because no code depends on it.
+>
+> The product name lives in `desktop/src/shared/constants/brand.ts`.
+> Upstream documentation below is unchanged and still refers to Buzz.
+
 <h1 align="center">Buzz 🐝</h1>
 
 <p align="center">

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-channel-delivery-lifecycle-0114e547-fcbc-4964-ac73-cc39825f5cba.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-channel-delivery-lifecycle-09cc5e54-d9c1-45ea-a422-1b0967912ee8.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-channel-delivery-lifecycle-5d24be64-44bc-4b53-8970-461f683422bf.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-channel-delivery-lifecycle-7d3d29d8-606f-4ed4-be7c-ab61079e7d3e.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-channel-delivery-lifecycle-7e9e13b6-620e-4dc4-8da8-1c3b5f55e01c.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-late-steer-wire-031482d2-7183-459d-b959-ab311f2e4eaa.ndjson

@@ -0,0 +1 @@
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-late-steer-wire-548c3a05-fed4-4329-ad4e-0a64c0a768e9.ndjson

@@ -0,0 +1 @@
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-late-steer-wire-e9aa4770-3a63-47f3-814b-d2828f07ac70.ndjson

@@ -0,0 +1 @@
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-late-steer-wire-efa7efad-ce54-4a49-84df-e5ff197ecc5c.ndjson

@@ -0,0 +1 @@
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-late-steer-wire-fca9651f-8084-4073-8951-079e542d6704.ndjson

@@ -0,0 +1 @@
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-merged-delivery-wire-0126fe48-4d4c-4792-97a0-d184a35ab949.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-merged-delivery-wire-0433391f-3465-4c74-ad9b-c918582a5a86.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-merged-delivery-wire-1fc05edd-9f73-4c25-85a9-322737290eb7.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-merged-delivery-wire-900aab02-2f7f-4bad-9f47-e0d1d433cc67.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-merged-delivery-wire-e3f49afc-0711-441b-99c2-ee72ee55bcb5.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-standing-lifecycle-3735dd72-0cd8-4264-8a28-baebd390ba19.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-standing-lifecycle-600fdd4b-cfba-40ca-ae02-642c750d3282.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-standing-lifecycle-61b95968-5472-4e75-b12c-6780029c5a51.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-standing-lifecycle-97123208-44a0-4dd0-8fcd-11add8e7cd8f.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/CUsersPCOWNE~1AppDataLocalTempbuzz-acp-standing-lifecycle-a5e64bcb-9bd1-47f5-b77d-0761d9044b6a.ndjson

@@ -0,0 +1,2 @@
+
+

crates/buzz-acp/src/pool.rs

@@ -4752,15 +4752,27 @@ pub(crate) async fn reaction_add(rest: &crate::relay::RestClient, event_id: &str
     }
 }
 
-/// Best-effort: post a visible failure notice (kind:9) to a channel after a
-/// batch is dead-lettered. Replies into the thread of `thread_tags` when the
-/// triggering event was threaded. Errors are logged and swallowed — the
-/// notice must never take down the main loop.
-pub(crate) async fn post_failure_notice(
+/// Post a notice from the agent into a channel, optionally p-tagging people who
+/// need to know.
+///
+/// Replies into the thread of `thread_tags` when the triggering event was
+/// threaded, so the notice lands beside the request it concerns. Errors are
+/// logged and swallowed — a notice must never take down the main loop.
+///
+/// `mentions` is how a notice reaches the person who can act on it: the owner
+/// learns through the normal mention path rather than by watching a channel.
+/// Notices stay in the channel the work arrived in — keeping the exchange with
+/// its project preserves the context instead of stranding it in a private
+/// inbox, and the agent never opens a DM of its own.
+///
+/// Best-effort by design: the caller must not be blocked waiting on the relay,
+/// and a notice that cannot be delivered must not take the turn down with it.
+pub(crate) async fn post_notice(
     rest: &crate::relay::RestClient,
     channel_id: Uuid,
     thread_tags: &ThreadTags,
     content: &str,
+    mentions: &[String],
 ) {
     let thread_ref = thread_tags.root_event_id.as_deref().and_then(|root| {
         let root_id = nostr::EventId::from_hex(root).ok()?;
@@ -4774,25 +4786,32 @@ pub(crate) async fn post_failure_notice(
             parent_event_id: parent_id,
         })
     });
-    let builder =
-        match buzz_sdk::build_message(channel_id, content, thread_ref.as_ref(), &[], false, &[]) {
-            Ok(b) => b,
-            Err(e) => {
-                tracing::warn!(channel = %channel_id, "failure notice: build failed: {e}");
-                return;
-            }
-        };
+    let mention_refs: Vec<&str> = mentions.iter().map(String::as_str).collect();
+    let builder = match buzz_sdk::build_message(
+        channel_id,
+        content,
+        thread_ref.as_ref(),
+        &mention_refs,
+        false,
+        &[],
+    ) {
+        Ok(b) => b,
+        Err(e) => {
+            tracing::warn!(channel = %channel_id, "notice: build failed: {e}");
+            return;
+        }
+    };
     let event = match builder.sign_with_keys(&rest.keys) {
         Ok(e) => e,
         Err(e) => {
-            tracing::warn!(channel = %channel_id, "failure notice: sign failed: {e}");
+            tracing::warn!(channel = %channel_id, "notice: sign failed: {e}");
             return;
         }
     };
     match tokio::time::timeout(Duration::from_secs(5), rest.submit_event(&event)).await {
         Ok(Ok(_)) => {}
-        Ok(Err(e)) => tracing::warn!(channel = %channel_id, "failure notice failed: {e}"),
-        Err(_) => tracing::warn!(channel = %channel_id, "failure notice timed out"),
+        Ok(Err(e)) => tracing::warn!(channel = %channel_id, "notice failed: {e}"),
+        Err(_) => tracing::warn!(channel = %channel_id, "notice timed out"),
     }
 }
 

crates/buzz-acp/src/queue.rs

@@ -136,9 +136,35 @@ pub struct FlushBatch {
 ///     if retry_counts[channel] > MAX_RETRIES: dead-letter (log ERROR, return batch to caller)
 ///     else: push_front with original received_at, set exponential backoff retry_after with jitter
 /// ```
+/// Events discarded before ever reaching an agent, by cause.
+///
+/// Both paths lose work somebody is waiting on, so they are counted and not
+/// merely logged. A running total is what makes the loss detectable: a
+/// watchdog can alert on any non-zero delta without a metrics pipeline, and
+/// "this counter moved" is a question anyone can answer from the logs.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct DropCounts {
+    /// Discarded because that channel already had a turn in flight, under
+    /// [`DedupMode::Drop`].
+    pub in_flight_dedup: u64,
+    /// Discarded because the channel queue was at its depth cap. The *oldest*
+    /// event is evicted, so the request that has waited longest is the one
+    /// lost — the reverse of what someone waiting would expect.
+    pub queue_depth_cap: u64,
+}
+
+impl DropCounts {
+    /// Total events lost across every cause.
+    pub fn total(&self) -> u64 {
+        self.in_flight_dedup + self.queue_depth_cap
+    }
+}
+
 pub struct EventQueue {
     queues: HashMap<Uuid, VecDeque<QueuedEvent>>,
     in_flight_channels: HashSet<Uuid>,
+    /// Running totals of work discarded before delivery. See [`DropCounts`].
+    drops: DropCounts,
     /// Per-channel deadline for auto-expiring stuck in-flight entries.
     in_flight_deadlines: HashMap<Uuid, Instant>,
     /// Number of events in each in-flight batch (for expiry logging).
@@ -182,6 +208,7 @@ impl EventQueue {
         Self {
             queues: HashMap::new(),
             in_flight_channels: HashSet::new(),
+            drops: DropCounts::default(),
             in_flight_deadlines: HashMap::new(),
             in_flight_batch_sizes: HashMap::new(),
             retry_after: HashMap::new(),
@@ -233,19 +260,27 @@ impl EventQueue {
         if matches!(self.dedup_mode, DedupMode::Drop)
             && self.in_flight_channels.contains(&event.channel_id)
         {
-            tracing::debug!(
+            self.drops.in_flight_dedup += 1;
+            // warn, not debug: this discards a request someone sent. At debug
+            // level the loss is invisible in practice, which is how work goes
+            // missing without anyone being able to say so afterwards.
+            tracing::warn!(
                 channel_id = %event.channel_id,
+                dropped_total = self.drops.in_flight_dedup,
                 "dropping event for in-flight channel (drop mode)"
             );
             return false;
         }
+        let drops = &mut self.drops;
         let queue = self.queues.entry(event.channel_id).or_default();
         // Enforce per-channel depth cap: drop oldest to make room.
         if queue.len() >= MAX_PENDING_PER_CHANNEL {
             queue.pop_front();
+            drops.queue_depth_cap += 1;
             tracing::warn!(
                 channel_id = %event.channel_id,
                 limit = MAX_PENDING_PER_CHANNEL,
+                dropped_total = drops.queue_depth_cap,
                 "queue depth cap reached — dropped oldest event"
             );
         }
@@ -626,6 +661,15 @@ impl EventQueue {
     }
 
     /// Number of channels with pending events.
+    /// Running totals of work discarded before delivery.
+    ///
+    /// Exposed so a supervisor outside this queue can observe the loss. A
+    /// counter only the queue can see is not observability — the process that
+    /// drops the work is the least reliable place to report it from.
+    pub fn drop_counts(&self) -> DropCounts {
+        self.drops
+    }
+
     pub fn pending_channels(&self) -> usize {
         self.queues.len()
     }
@@ -2005,6 +2049,78 @@ mod tests {
     use std::time::Duration;
 
     /// Build a test event with the given content and kind.
+    // ── Discarded-work accounting ─────────────────────────────────────
+    //
+    // Both drop paths lose a request someone sent. They were previously
+    // observable only as scattered log lines — one of them at debug level —
+    // so work could go missing with no way to establish afterwards that it
+    // had. These lock the counting down.
+
+    #[test]
+    fn test_dedup_drop_is_counted_not_just_logged() {
+        let mut q = EventQueue::new(DedupMode::Drop);
+        let chan = Uuid::new_v4();
+        // flush_next() is what puts a channel in flight.
+        assert!(q.push(make_queued(chan, "first")));
+        let _in_flight = q.flush_next().expect("first push is flushable");
+
+        assert_eq!(q.drop_counts().total(), 0, "nothing lost yet");
+        assert!(
+            !q.push(make_queued(chan, "dropped one")),
+            "drop mode discards"
+        );
+        assert!(
+            !q.push(make_queued(chan, "dropped two")),
+            "drop mode discards"
+        );
+
+        let drops = q.drop_counts();
+        assert_eq!(
+            drops.in_flight_dedup, 2,
+            "every discarded event must be counted"
+        );
+        assert_eq!(drops.queue_depth_cap, 0, "wrong cause attributed");
+        assert_eq!(drops.total(), 2);
+    }
+
+    #[test]
+    fn test_queue_mode_does_not_count_drops_because_it_does_not_drop() {
+        // Queue mode is the CLI default. It must not inflate the loss counter,
+        // or a real drop becomes indistinguishable from normal operation.
+        let mut q = EventQueue::new(DedupMode::Queue);
+        let chan = Uuid::new_v4();
+        assert!(q.push(make_queued(chan, "first")));
+        let _in_flight = q.flush_next().expect("first push is flushable");
+
+        assert!(q.push(make_queued(chan, "queued behind in-flight")));
+        assert_eq!(q.drop_counts().total(), 0, "queue mode loses nothing");
+    }
+
+    #[test]
+    fn test_depth_cap_drop_is_counted_and_evicts_the_oldest() {
+        let mut q = EventQueue::new(DedupMode::Queue);
+        let chan = Uuid::new_v4();
+        for i in 0..MAX_PENDING_PER_CHANNEL {
+            assert!(q.push(make_queued(chan, &format!("event {i}"))));
+        }
+        assert_eq!(q.drop_counts().total(), 0, "at the cap, nothing lost yet");
+
+        // One past the cap evicts the OLDEST — the request that has waited
+        // longest is the one lost, which is worth asserting because it is the
+        // opposite of what someone waiting would expect.
+        assert!(q.push(make_queued(chan, "one too many")));
+        let drops = q.drop_counts();
+        assert_eq!(drops.queue_depth_cap, 1);
+        assert_eq!(drops.in_flight_dedup, 0, "wrong cause attributed");
+
+        let batch = q.flush_next().expect("a batch is pending");
+        let first = batch.events.first().expect("batch is non-empty");
+        assert!(
+            !first.event.content.contains("event 0"),
+            "the oldest event should have been evicted, but it is still at the front"
+        );
+    }
+
     fn make_event(content: &str) -> Event {
         let keys = Keys::generate();
         EventBuilder::new(Kind::Custom(9), content)