fix(acp): reduce runaway subscription usage

crates/buzz-acp/src/acp.rs

@@ -10,12 +10,14 @@
 
 use futures_util::StreamExt;
 use tokio::io::AsyncWriteExt;
-use tokio::process::{Child, ChildStdin, ChildStdout};
+use tokio::process::{Child, ChildStdout};
 use tokio_util::codec::{FramedRead, LinesCodec, LinesCodecError};
 
 use crate::observer::{ObserverContext, ObserverHandle};
 use crate::usage::{TurnUsage, UsageTracker};
 
+type StdinWriteRequest = (Vec<u8>, tokio::sync::oneshot::Sender<std::io::Result<()>>);
+
 /// Maximum allowed size of a single NDJSON line from the agent's stdout.
 /// Lines exceeding this limit are rejected to prevent OOM from rogue agents.
 const MAX_LINE_SIZE: usize = 10_000_000; // 10 MB
@@ -139,8 +141,13 @@ fn build_initialize_params() -> serde_json::Value {
 pub struct AcpClient {
     /// The agent child process (kept alive to prevent zombie).
     child: Child,
-    /// Write end of the agent's stdin pipe.
-    stdin: ChildStdin,
+    /// Cancellation-safe writer queue for the agent's stdin pipe.
+    ///
+    /// A dedicated task owns `ChildStdin` and finishes each complete NDJSON
+    /// frame after accepting it. This prevents a cancelled prompt future from
+    /// leaving a partial JSON payload that the next control frame would append
+    /// to on the same stream.
+    stdin_tx: tokio::sync::mpsc::Sender<StdinWriteRequest>,
     /// Framed reader over the agent's stdout pipe (line-oriented, bounded).
     /// Uses `LinesCodec::new_with_max_length` to enforce MAX_LINE_SIZE at the
     /// read level — prevents OOM from rogue agents writing infinite non-newline bytes.
@@ -481,9 +488,26 @@ impl AcpClient {
             .take()
             .ok_or_else(|| AcpError::Protocol("failed to open agent stdout".into()))?;
 
+        let (stdin_tx, mut stdin_rx) = tokio::sync::mpsc::channel::<StdinWriteRequest>(8);
+        tokio::spawn(async move {
+            let mut stdin = stdin;
+            while let Some((frame, completion_tx)) = stdin_rx.recv().await {
+                let result = async {
+                    stdin.write_all(&frame).await?;
+                    stdin.flush().await
+                }
+                .await;
+                let write_failed = result.is_err();
+                let _ = completion_tx.send(result);
+                if write_failed {
+                    break;
+                }
+            }
+        });
+
         Ok(Self {
             child,
-            stdin,
+            stdin_tx,
             reader: FramedRead::new(stdout, LinesCodec::new_with_max_length(MAX_LINE_SIZE)),
             next_id: 0,
             pending_permission_id: None,
@@ -944,18 +968,32 @@ impl AcpClient {
         self.parse_stop_reason(&result)
     }
 
-    /// Serialize `value` as a single NDJSON line and flush to the agent's stdin.
+    /// Serialize `value` as a single NDJSON frame and enqueue it for the
+    /// dedicated stdin writer.
     ///
     /// Bounded by a 30-second write timeout. If the agent stops reading stdin
     /// (e.g., it's stuck or dead), the write would otherwise block forever.
     async fn write_ndjson(&mut self, value: &serde_json::Value) -> Result<(), AcpError> {
         const WRITE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
-        let line = serde_json::to_string(value)?;
+        let mut frame = serde_json::to_vec(value)?;
+        frame.push(b'\n');
+        let (completion_tx, completion_rx) = tokio::sync::oneshot::channel();
         tokio::time::timeout(WRITE_TIMEOUT, async {
-            self.stdin.write_all(line.as_bytes()).await?;
-            self.stdin.write_all(b"\n").await?;
-            self.stdin.flush().await?;
-            Ok::<(), std::io::Error>(())
+            self.stdin_tx
+                .send((frame, completion_tx))
+                .await
+                .map_err(|_| {
+                    std::io::Error::new(
+                        std::io::ErrorKind::BrokenPipe,
+                        "agent stdin writer stopped",
+                    )
+                })?;
+            completion_rx.await.map_err(|_| {
+                std::io::Error::new(
+                    std::io::ErrorKind::BrokenPipe,
+                    "agent stdin writer dropped completion",
+                )
+            })?
         })
         .await
         .map_err(|_| AcpError::WriteTimeout(WRITE_TIMEOUT))?

crates/buzz-acp/src/filter.rs

@@ -371,6 +371,21 @@ pub async fn match_event(
     rules: &[SubscriptionRule],
     agent_pubkey_hex: &str,
 ) -> Option<MatchedRule> {
+    // UI activity signals are never actionable agent input. Keep this guard
+    // ahead of operator-authored rules so an accidental wildcard subscription
+    // cannot turn high-frequency typing/presence traffic into LLM prompts or
+    // mid-turn steer signals. Huddle reaction bursts are similarly visual-only.
+    // Durable message reactions (kind 7) are intentionally not included here:
+    // callers may explicitly subscribe to those for an agent workflow.
+    if matches!(
+        event.kind.as_u16() as u32,
+        buzz_core::kind::KIND_PRESENCE_UPDATE
+            | buzz_core::kind::KIND_TYPING_INDICATOR
+            | buzz_core::kind::KIND_HUDDLE_REACTION
+    ) {
+        return None;
+    }
+
     let filter_ctx = FilterContext::from_event(event, channel_id);
 
     for (index, rule) in rules.iter().enumerate() {
@@ -635,6 +650,39 @@ mod tests {
         assert_eq!(matched.prompt_tag, "matched");
     }
 
+    #[tokio::test]
+    async fn test_match_event_rejects_ui_activity_signals_under_wildcard_rule() {
+        let channel_id = any_channel();
+        let rules = vec![make_rule(
+            "wildcard",
+            ChannelScope::All("all".into()),
+            vec![],
+            false,
+            None,
+            Some("all"),
+        )];
+
+        for kind in [
+            buzz_core::kind::KIND_PRESENCE_UPDATE,
+            buzz_core::kind::KIND_TYPING_INDICATOR,
+            buzz_core::kind::KIND_HUDDLE_REACTION,
+        ] {
+            let event = make_event(kind, "");
+            assert!(
+                match_event(&event, channel_id, &rules, "").await.is_none(),
+                "UI activity kind {kind} must never become an agent prompt"
+            );
+        }
+
+        let message = make_event(buzz_core::kind::KIND_STREAM_MESSAGE, "hello");
+        assert!(
+            match_event(&message, channel_id, &rules, "")
+                .await
+                .is_some(),
+            "the wildcard rule must still accept actionable messages"
+        );
+    }
+
     #[tokio::test]
     async fn test_match_event_require_mention() {
         let agent_pubkey = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef";

crates/buzz-acp/src/lib.rs

@@ -3007,6 +3007,26 @@ fn is_auth_error(error: &acp::AcpError) -> bool {
     message.contains("Re-authenticate") || message.contains("API Error: 401")
 }
 
+/// Returns `true` when the provider has rejected work until the user's quota
+/// or subscription window resets.
+///
+/// These failures cannot recover through immediate retry. Requeueing them can
+/// repeatedly submit the same large prompt, consuming retry budget and making
+/// an already-exhausted subscription worse. Match only provider-originated
+/// `AgentError` messages and narrow, observed quota phrases to avoid treating
+/// transient transport/rate errors as terminal.
+fn is_usage_limit_error(error: &acp::AcpError) -> bool {
+    let acp::AcpError::AgentError { message, .. } = error else {
+        return false;
+    };
+    let message = message.to_ascii_lowercase();
+    message.contains("session limit")
+        || message.contains("usage credits required")
+        || message.contains("weekly limit")
+        || message.contains("monthly limit")
+        || message.contains("spending limit")
+}
+
 /// Spawn a task that posts a user-visible failure notice to the relay.
 ///
 /// Shared by the hard-cap immediate dead-letter path and the retries-exhausted
@@ -3142,6 +3162,22 @@ fn handle_prompt_result(
                     and then re-send."
                     .to_string();
                 spawn_failure_notice(rest_client, &batch, content);
+            } else if matches!(&result.outcome, PromptOutcome::Error(e) if is_usage_limit_error(e))
+            {
+                // Subscription/quota failures are also non-retryable until an
+                // external reset or account change. Do not resubmit the same
+                // prompt in a retry storm.
+                tracing::warn!(
+                    channel_id = %batch.channel_id,
+                    events = batch.events.len(),
+                    "dead-lettering batch immediately — provider usage limit reached"
+                );
+                let content = "⚠️ I couldn't process the last request because the AI provider's \
+                    usage or subscription limit was reached. Wait for the provider reset (or \
+                    change the configured model/account), then re-send. Buzz will not retry \
+                    automatically."
+                    .to_string();
+                spawn_failure_notice(rest_client, &batch, content);
             } else if let Some(dead) = queue.requeue(batch) {
                 let reason = match &result.outcome {
                     PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(),
@@ -6146,6 +6182,38 @@ mod error_outcome_emission_tests {
         );
     }
 
+    // ── is_usage_limit_error classification ───────────────────────────────
+
+    #[test]
+    fn is_usage_limit_error_matches_claude_session_limit() {
+        let e = acp::AcpError::AgentError {
+            code: -32000,
+            message: "You've hit your session limit · resets 9:50pm".to_string(),
+        };
+        assert!(is_usage_limit_error(&e));
+    }
+
+    #[test]
+    fn is_usage_limit_error_matches_usage_credits_requirement() {
+        let e = acp::AcpError::AgentError {
+            code: -32000,
+            message: "Usage credits required for 1M context".to_string(),
+        };
+        assert!(is_usage_limit_error(&e));
+    }
+
+    #[test]
+    fn is_usage_limit_error_rejects_transient_and_transport_errors() {
+        let transient = acp::AcpError::AgentError {
+            code: -32000,
+            message: "Service temporarily unavailable".to_string(),
+        };
+        assert!(!is_usage_limit_error(&transient));
+        assert!(!is_usage_limit_error(&acp::AcpError::Io(
+            std::io::Error::other("pipe broke")
+        )));
+    }
+
     // ── auth error dead-letter behavior ────────────────────────────────────
 
     /// An auth-class `PromptOutcome::Error` must dead-letter immediately
@@ -6234,10 +6302,10 @@ mod error_outcome_emission_tests {
         );
     }
 
-    /// A non-auth application error (e.g. usage credits) must still follow the
-    /// standard requeue path so today's behavior is unchanged.
+    /// A provider quota error must dead-letter immediately instead of
+    /// repeatedly resubmitting the same prompt before the external reset.
     #[tokio::test]
-    async fn non_auth_application_error_is_requeued() {
+    async fn usage_limit_error_dead_letters_immediately_without_requeueing() {
         let keys = nostr::Keys::generate();
         let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "test")
             .sign_with_keys(&keys)
@@ -6306,17 +6374,88 @@ mod error_outcome_emission_tests {
             None,
         );
 
-        // Non-auth application error: batch IS requeued (first attempt, retry budget > 0).
+        // Provider usage error: batch is not requeued.
         assert_eq!(
             queue.pending_channels(),
-            1,
-            "non-auth application error must requeue the batch for retry"
+            0,
+            "usage-limit error must not requeue the batch"
         );
         assert_eq!(
             queue.queued_event_count(&channel_id),
-            1,
-            "non-auth application error must preserve the event for retry"
+            0,
+            "usage-limit error must not leave events pending"
+        );
+    }
+
+    /// Other application failures retain the bounded retry behavior.
+    #[tokio::test]
+    async fn transient_application_error_is_requeued() {
+        let keys = nostr::Keys::generate();
+        let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "test")
+            .sign_with_keys(&keys)
+            .unwrap();
+        let channel_id = uuid::Uuid::new_v4();
+        let batch = FlushBatch {
+            channel_id,
+            events: vec![BatchEvent {
+                event,
+                prompt_tag: "test".into(),
+                received_at: std::time::Instant::now(),
+            }],
+            cancelled_events: vec![],
+            cancel_reason: None,
+        };
+        let error = acp::AcpError::AgentError {
+            code: -32000,
+            message: "Service temporarily unavailable".to_string(),
+        };
+        let agent = dummy_agent(0).await;
+        let mut pool = AgentPool::from_slots(vec![None]);
+        let task_id = pool.join_set.spawn(async {}).id();
+        pool.task_map_mut().insert(
+            task_id,
+            crate::pool::TaskMeta {
+                agent_index: 0,
+                channel_id: None,
+                turn_id: "test-turn-id".to_string(),
+                recoverable_batch: None,
+                control_tx: None,
+                steer_tx: None,
+            },
+        );
+        let mut queue = EventQueue::new(config::DedupMode::Queue);
+        let config = test_config();
+        let mut heartbeat_in_flight = false;
+        let removed_channels = std::collections::HashSet::new();
+        let mut crash_history = vec![SlotCircuit {
+            crash_times: Vec::new(),
+            open_until: None,
+            respawn_in_flight: false,
+        }];
+        let (respawn_tx, _respawn_rx) = mpsc::channel(8);
+        let mut respawn_tasks = tokio::task::JoinSet::new();
+        let result = PromptResult {
+            agent,
+            source: PromptSource::Channel(channel_id),
+            turn_id: "test-turn-id".to_string(),
+            outcome: PromptOutcome::Error(error),
+            batch: Some(batch),
+        };
+        handle_prompt_result(
+            &mut pool,
+            &mut queue,
+            &config,
+            result,
+            &mut heartbeat_in_flight,
+            &removed_channels,
+            &mut crash_history,
+            &respawn_tx,
+            &mut respawn_tasks,
+            None,
+            None,
         );
+        assert_eq!(queue.pending_channels(), 1);
+        assert_eq!(queue.queued_event_count(&channel_id), 1);
     }
 }