fix: chart auto-migrate image, AppImage dmabuf, silent reply notice

.env.example

@@ -204,6 +204,13 @@ RUST_LOG=buzz_relay=debug,buzz_db=debug,buzz_auth=debug,buzz_pubsub=debug,tower_
 # Set to true to process the agent's own messages (default: ignore self).
 # BUZZ_ACP_NO_IGNORE_SELF=false
 
+# Seconds to wait after an Ok mention turn before posting a "couldn't publish"
+# notice, so a late ws echo (including post-reconnect replay) can clear it.
+# Default 8 covers ≈6s relay resubscribe bursts; set higher for many-channel
+# harnesses. Values ≤0 or unparsable fall back to 8 (grace cannot be disabled).
+# (#2459)
+# BUZZ_SILENT_REPLY_GRACE_SECS=8
+
 # ── Context ──────────────────────────────────────────────────────────────────
 # Max context messages fetched for thread replies and DMs (0–100). 0 = disabled.
 # BUZZ_ACP_CONTEXT_MESSAGE_LIMIT=12

crates/buzz-acp/src/lib.rs

@@ -10,15 +10,38 @@ mod pool_lifecycle;
 mod queue;
 mod relay;
 mod setup_mode;
+mod silent_reply;
 mod usage;
 
+/// Default grace before posting a silent-reply notice so a late ws echo can
+/// land (#2459). Relay resubscribe bursts can spread ≈6s under REQ pacing, so
+/// the default sits above that; override with `BUZZ_SILENT_REPLY_GRACE_SECS`.
+const DEFAULT_SILENT_REPLY_GRACE_SECS: u64 = 8;
+
+fn silent_reply_grace() -> Duration {
+    std::env::var("BUZZ_SILENT_REPLY_GRACE_SECS")
+        .ok()
+        .and_then(|raw| raw.parse::<u64>().ok())
+        .filter(|&secs| secs > 0)
+        .map(Duration::from_secs)
+        .unwrap_or(Duration::from_secs(DEFAULT_SILENT_REPLY_GRACE_SECS))
+}
+
+/// Deferred silent-reply check payload: batch plus the watch generation armed
+/// when the Ok turn completed.
+#[derive(Clone)]
+struct SilentReplyCheck {
+    batch: FlushBatch,
+    generation: u64,
+}
+
 pub use usage::TurnUsage;
 
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::sync::Arc;
 use std::time::Duration;
 
-use acp::{AcpClient, EnvVar, McpServer};
+use acp::{AcpClient, EnvVar, McpServer, StopReason};
 use anyhow::Result;
 use buzz_core::kind::{
     KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_STREAM_MESSAGE,
@@ -1627,6 +1650,9 @@ async fn tokio_main() -> Result<()> {
     //      withheld event in `EventQueue::withheld_native_steer` until
     //      `IN_FLIGHT_DEADLINE_SECS` expires.
     let (steer_ack_tx, mut steer_ack_rx) = mpsc::unbounded_channel::<SteerAckEvent>();
+    // Deferred silent-reply checks: Ok mention turns wait a grace window so a
+    // late ws echo can clear the "couldn't publish" notice (#2459).
+    let (silent_reply_tx, mut silent_reply_rx) = mpsc::unbounded_channel::<SilentReplyCheck>();
 
     // ── Step 7: Shutdown signal ───────────────────────────────────────────────
     let (shutdown_tx, mut shutdown_rx) = watch::channel(());
@@ -1701,6 +1727,7 @@ async fn tokio_main() -> Result<()> {
         Result(Box<PromptResult>),
         Panic(tokio::task::JoinError),
         SteerAck(SteerAckEvent),
+        SilentReplyDue(SilentReplyCheck),
         Wake(u32, Result<AgentPool, String>),
     }
 
@@ -1844,6 +1871,9 @@ async fn tokio_main() -> Result<()> {
                 Some(ack_event) = steer_ack_rx.recv() => {
                     Some(PoolEvent::SteerAck(ack_event))
                 }
+                Some(check) = silent_reply_rx.recv() => {
+                    Some(PoolEvent::SilentReplyDue(check))
+                }
                 Some((attempt, result)) = wake_rx.recv(), if config.lazy_pool && !pool_ready => {
                     Some(PoolEvent::Wake(attempt, result))
                 }
@@ -2024,6 +2054,14 @@ async fn tokio_main() -> Result<()> {
                                 continue;
                             }
 
+                            // Count self-authored stream replies for silent-reply
+                            // detection regardless of ignore_self (#2459): under
+                            // --no-ignore-self the counter must still increment.
+                            if buzz_event.event.pubkey.to_hex() == pubkey_hex
+                                && silent_reply::is_agent_reply_kind(kind_u32)
+                            {
+                                queue.note_self_publish(buzz_event.channel_id);
+                            }
                             if config.ignore_self && buzz_event.event.pubkey.to_hex() == pubkey_hex {
                                 tracing::debug!(channel_id = %buzz_event.channel_id, "dropping self-authored event");
                                 continue;
@@ -2367,6 +2405,7 @@ async fn tokio_main() -> Result<()> {
                     &mut respawn_tasks,
                     observer.clone(),
                     Some(&ctx.rest_client),
+                    Some(&silent_reply_tx),
                 ) == LoopAction::Exit
                 {
                     break;
@@ -2533,6 +2572,26 @@ async fn tokio_main() -> Result<()> {
                     typing_channels.insert(channel_id, thread_tags);
                 }
             }
+            Some(PoolEvent::SilentReplyDue(check)) => {
+                let Some(publishes) =
+                    queue.take_silent_reply_watch(check.batch.channel_id, check.generation)
+                else {
+                    // Already consumed or superseded by a later Ok on the same
+                    // channel; nothing to do.
+                    continue;
+                };
+                if let Some(content) = silent_reply::silent_reply_loss_notice(
+                    &PromptOutcome::Ok(StopReason::EndTurn),
+                    Some(&check.batch),
+                    publishes,
+                ) {
+                    tracing::warn!(
+                        channel_id = %check.batch.channel_id,
+                        "mention turn completed Ok with no agent channel publish — posting failure notice"
+                    );
+                    spawn_failure_notice(Some(&ctx.rest_client), &check.batch, content);
+                }
+            }
             Some(PoolEvent::Wake(attempt, result)) => {
                 let completion = result.as_ref().map(|_| ()).map_err(|error| error.clone());
                 if let Err(error) =
@@ -3043,6 +3102,7 @@ fn handle_prompt_result(
     respawn_tasks: &mut tokio::task::JoinSet<()>,
     observer: Option<observer::ObserverHandle>,
     rest_client: Option<&relay::RestClient>,
+    silent_reply_tx: Option<&mpsc::UnboundedSender<SilentReplyCheck>>,
 ) -> LoopAction {
     let before = pool.task_map().len();
     let agent_index = result.agent.index;
@@ -3068,7 +3128,47 @@ fn handle_prompt_result(
         // Don't requeue batches for channels the agent was removed from —
         // those events are stale and should be silently dropped.
         if !removed_channels.contains(&batch.channel_id) {
-            if matches!(
+            if matches!(result.outcome, PromptOutcome::Ok(_)) {
+                // Ok turns keep the batch only for silent-reply detection —
+                // never requeue a successful turn (#2459).
+                if silent_reply::batch_expects_channel_reply(&batch) {
+                    // Defer the notice: the reply echo arrives on the ws path
+                    // and is not ordered vs pool completion. Arm a watch so
+                    // counts survive mark_complete, then re-check after a
+                    // short grace window.
+                    let generation = queue.arm_silent_reply_watch(batch.channel_id);
+                    if let Some(tx) = silent_reply_tx {
+                        let tx = tx.clone();
+                        let check = SilentReplyCheck {
+                            batch: batch.clone(),
+                            generation,
+                        };
+                        let grace = silent_reply_grace();
+                        tokio::spawn(async move {
+                            tokio::time::sleep(grace).await;
+                            let _ = tx.send(check);
+                        });
+                    } else {
+                        // Tests / callers without a deferred channel: decide now.
+                        let publishes = queue
+                            .take_silent_reply_watch(batch.channel_id, generation)
+                            .unwrap_or(0);
+                        if let Some(content) = silent_reply::silent_reply_loss_notice(
+                            &result.outcome,
+                            Some(&batch),
+                            publishes,
+                        ) {
+                            tracing::warn!(
+                                channel_id = %batch.channel_id,
+                                "mention turn completed Ok with no agent channel publish — posting failure notice"
+                            );
+                            spawn_failure_notice(rest_client, &batch, content);
+                        }
+                    }
+                } else {
+                    let _ = queue.take_self_publishes(batch.channel_id);
+                }
+            } else if matches!(
                 result.outcome,
                 PromptOutcome::Cancelled | PromptOutcome::CancelDrainTimeout(_)
             ) {
@@ -5246,6 +5346,7 @@ mod error_outcome_emission_tests {
             &mut respawn_tasks,
             Some(observer.clone()),
             None,
+            None,
         );
 
         let turn_errors: Vec<_> = observer
@@ -5411,6 +5512,7 @@ mod error_outcome_emission_tests {
                 &mut respawn_tasks,
                 Some(observer.clone()),
                 None,
+                None,
             );
             let events = observer.snapshot();
             let turn_error = events.iter().find(|e| e.kind == "turn_error").unwrap();
@@ -5501,6 +5603,7 @@ mod error_outcome_emission_tests {
                 &mut respawn_tasks,
                 None,
                 None,
+                None,
             );
             (
                 queue.pending_channels(),
@@ -5606,6 +5709,7 @@ mod error_outcome_emission_tests {
                 &mut respawn_tasks,
                 None,
                 None,
+                None,
             );
             (
                 queue.pending_channels(),
@@ -5697,6 +5801,7 @@ mod error_outcome_emission_tests {
             &mut respawn_tasks,
             Some(observer.clone()),
             None,
+            None,
         );
 
         let events = observer.snapshot();
@@ -5790,6 +5895,7 @@ mod error_outcome_emission_tests {
             &mut respawn_tasks,
             Some(observer.clone()),
             None,
+            None,
         );
 
         let events = observer.snapshot();
@@ -5905,6 +6011,7 @@ mod error_outcome_emission_tests {
             &mut respawn_tasks,
             Some(observer.clone()),
             None,
+            None,
         );
 
         // Batch preserved as a cancelled merge, not dead-lettered — same
@@ -6037,6 +6144,7 @@ mod error_outcome_emission_tests {
             &mut respawn_tasks,
             Some(observer.clone()),
             None,
+            None,
         );
 
         // No batch to merge — the queue has nothing pending for any channel.
@@ -6219,6 +6327,7 @@ mod error_outcome_emission_tests {
             &mut respawn_tasks,
             None,
             None,
+            None,
         );
 
         // The batch must not be requeued: pending_channels returns 0.
@@ -6304,6 +6413,7 @@ mod error_outcome_emission_tests {
             &mut respawn_tasks,
             None,
             None,
+            None,
         );
 
         // Non-auth application error: batch IS requeued (first attempt, retry budget > 0).

crates/buzz-acp/src/pool.rs

@@ -1978,7 +1978,8 @@ pub async fn run_prompt_task(
                             agent,
                             source,
                             PromptOutcome::Ok(StopReason::EndTurn),
-                            None, // turn succeeded — batch was processed, no requeue
+                            // Keep batch for silent-reply detection; Ok must not requeue.
+                            batch,
                         );
                         return;
                     }
@@ -2041,7 +2042,8 @@ pub async fn run_prompt_task(
                 agent,
                 source,
                 PromptOutcome::Ok(stop_reason),
-                None,
+                // Keep batch for silent-reply detection; Ok must not requeue.
+                batch,
             );
         }
         Err(AcpError::AgentExited) => {