feat(gateway): /stream cosmetic tool progress for Pi

docs/prd.md

@@ -169,6 +169,9 @@ user prompt.
 ## Control Commands
 
 - `/clear`, `/new`, `/reset`: rotate the current backend session.
+- `/stream`, `/stream on`, `/stream off`: toggle cosmetic tool-progress
+  messages for the current conversation (in-memory; off after restart). Progress
+  is never stored in canonical history and does not change agent prompts.
 - `/help`: show available commands.
 
 ## Acceptance Criteria

docs/reference/cli.md

@@ -323,6 +323,7 @@ These messages are handled by the gateway before backend dispatch:
 | --- | --- |
 | `/clear`, `/new`, `/reset` | Start a fresh backend session for that conversation |
 | `/stop` | Stop the active request; already queued messages continue in order |
+| `/stream`, `/stream on`, `/stream off` | Toggle cosmetic tool-progress messages for that conversation (Pi). Does not change LLM prompts or canonical history. |
 | `/help` | Return the available chat commands |
 
 Starting a fresh session preserves canonical history. Push can seed the new

src/agent.rs

@@ -5,7 +5,10 @@ use std::time::Duration;
 
 use uuid::Uuid;
 
+use tokio::sync::mpsc;
+
 use crate::config::{AgentBackend, Config};
+use crate::progress::ProgressEvent;
 use crate::{claude, codex, pi};
 
 /// One headless agent turn.
@@ -55,10 +58,10 @@ pub enum Runner {
 /// replaced. Retry only that transient spawn error, within the caller's
 /// overall timeout, and preserve every other error unchanged. `spawn` must
 /// build a fresh child process attempt on every call.
-pub(crate) async fn output_with_retry<F, Fut>(mut spawn: F) -> std::io::Result<std::process::Output>
+pub(crate) async fn output_with_retry<F, Fut, T>(mut spawn: F) -> std::io::Result<T>
 where
     F: FnMut() -> Fut,
-    Fut: std::future::Future<Output = std::io::Result<std::process::Output>>,
+    Fut: std::future::Future<Output = std::io::Result<T>>,
 {
     let mut attempts = 0;
     loop {
@@ -122,13 +125,28 @@ impl Runner {
         matches!(self, Runner::Claude(_))
     }
 
+    /// Like [`Self::run_with_progress`] with no progress channel.
+    #[allow(dead_code)] // jobs and other callers; gateway uses run_with_progress
     pub async fn run(&self, req: Request<'_>, timeout: Duration) -> Result<RunOutput, RunError> {
+        self.run_with_progress(req, timeout, None).await
+    }
+
+    /// Like [`Self::run`], optionally forwarding cosmetic tool progress (Pi only).
+    pub async fn run_with_progress(
+        &self,
+        req: Request<'_>,
+        timeout: Duration,
+        progress: Option<mpsc::UnboundedSender<ProgressEvent>>,
+    ) -> Result<RunOutput, RunError> {
         match self {
             Runner::Claude(r) => r.run(req, timeout).await,
             Runner::Codex(r) => r.run(req, timeout).await,
-            Runner::Pi(r) => r.run(req, timeout).await,
+            Runner::Pi(r) => r.run_with_progress(req, timeout, progress).await,
             #[cfg(test)]
-            Runner::Fake(r) => r.run(req, timeout).await,
+            Runner::Fake(r) => {
+                let _ = progress;
+                r.run(req, timeout).await
+            }
         }
     }
 

src/channel.rs

@@ -289,6 +289,29 @@ impl Channel {
         }
     }
 
+    /// Send a cosmetic progress bubble. Telegram returns a message id for edits.
+    pub async fn send_progress(&self, target: &str, text: &str) -> Result<Option<i64>> {
+        match self {
+            Self::Telegram(channel) => Ok(Some(channel.send_progress(target, text).await?)),
+            // Progress edits are Telegram-first; other channels skip delivery.
+            Self::IMessage(_) | Self::Slack(_) => {
+                let _ = (target, text);
+                Ok(Some(1))
+            }
+        }
+    }
+
+    /// Edit a Telegram progress bubble in place. No-op on other channels.
+    pub async fn edit_progress(&self, target: &str, message_id: i64, text: &str) -> Result<()> {
+        match self {
+            Self::Telegram(channel) => channel.edit_progress(target, message_id, text).await,
+            Self::IMessage(_) | Self::Slack(_) => {
+                let _ = (target, message_id, text);
+                Ok(())
+            }
+        }
+    }
+
     pub async fn download_voice(&self, voice: &InboundVoice) -> Result<AudioClip> {
         match self {
             Self::IMessage(channel) => ChannelContract::download_voice(channel, voice).await,

src/gateway/mod.rs

@@ -23,6 +23,7 @@ use crate::channel::{Channel, InboundVoice, RawMessage};
 use crate::config::{AgentBackend, ChannelKind, Config, PrimaryDeliveryConfig};
 use crate::history::{History, OutboundOrigin};
 use crate::jobs;
+use crate::progress::StreamPrefs;
 use crate::store::Store;
 use crate::util::now_ms;
 use crate::voice::Voice;
@@ -60,11 +61,14 @@ struct Ctx {
     audit: Arc<AuditLog>,
     voice: Option<Voice>,
     schedule_destination: Option<PrimaryDestination>,
+    stream_prefs: StreamPrefs,
     #[cfg(test)]
     setup_failure_replies: Arc<Mutex<Vec<String>>>,
     #[cfg(test)]
     sent_replies: Arc<Mutex<Vec<(String, String)>>>,
     #[cfg(test)]
+    sent_progress: Arc<Mutex<Vec<(String, String)>>>,
+    #[cfg(test)]
     sent_voice_replies: SentVoiceReplies,
     #[cfg(test)]
     send_failures_remaining: Arc<Mutex<usize>>,
@@ -412,6 +416,7 @@ impl Gateway {
             assistant_dir: cfg.assistant_dir.clone(),
             audit,
             schedule_destination: None,
+            stream_prefs: StreamPrefs::default(),
             #[cfg(not(test))]
             voice: Voice::from_config(&cfg),
             #[cfg(test)]
@@ -421,6 +426,8 @@ impl Gateway {
             #[cfg(test)]
             sent_replies: Arc::new(Mutex::new(Vec::new())),
             #[cfg(test)]
+            sent_progress: Arc::new(Mutex::new(Vec::new())),
+            #[cfg(test)]
             sent_voice_replies: Arc::new(Mutex::new(Vec::new())),
             #[cfg(test)]
             send_failures_remaining: Arc::new(Mutex::new(0)),
@@ -801,7 +808,7 @@ impl Gateway {
                     voice_attachment: m.voice.clone(),
                     approval_origin,
                 };
-                if job.text.trim().eq_ignore_ascii_case("/stop") {
+                if worker::normalize_slash_command(&job.text) == "/stop" {
                     if !self.stop(job).await {
                         return;
                     }

src/gateway/tests.rs

@@ -226,8 +226,10 @@ fn setup_failure_ctx(
         )),
         schedule_destination: None,
         voice: None,
+        stream_prefs: crate::progress::StreamPrefs::default(),
         setup_failure_replies: Arc::new(Mutex::new(Vec::new())),
         sent_replies: Arc::new(Mutex::new(Vec::new())),
+        sent_progress: Arc::new(Mutex::new(Vec::new())),
         sent_voice_replies: Arc::new(Mutex::new(Vec::new())),
         send_failures_remaining: Arc::new(Mutex::new(0)),
         send_failure_after: Arc::new(Mutex::new(None)),
@@ -516,6 +518,69 @@ async fn cursor_save_failure_retries_without_rerunning_or_redelivering() {
     let _ = std::fs::remove_dir_all(assistant_dir);
 }
 
+#[tokio::test(flavor = "current_thread")]
+async fn stream_command_toggles_without_backend_run() {
+    let state_path = temp_state_path();
+    let sessions_dir = temp_path("stream-toggle-sessions");
+    let assistant_dir = temp_path("stream-toggle-assistant");
+    std::fs::create_dir_all(&assistant_dir).unwrap();
+    let calls = Arc::new(Mutex::new(Vec::new()));
+    let mut gateway = Gateway::new(test_config(
+        &state_path,
+        sessions_dir.to_str().unwrap(),
+        assistant_dir.to_str().unwrap(),
+    ))
+    .unwrap();
+    gateway.ctx.runners = Arc::new(fake_runners(calls.clone()));
+
+    run_messages(
+        &mut gateway,
+        vec![message(1, "+15551234567", "+15551234567", false, "/stream")],
+    )
+    .await;
+    assert!(gateway
+        .ctx
+        .stream_prefs
+        .is_enabled("imessage:dm:+15551234567"));
+    assert!(gateway
+        .ctx
+        .sent_replies
+        .lock()
+        .unwrap()
+        .last()
+        .is_some_and(|(_, text)| text.starts_with("Stream progress: on.")));
+
+    run_messages(
+        &mut gateway,
+        vec![message(
+            2,
+            "+15551234567",
+            "+15551234567",
+            false,
+            "/stream off",
+        )],
+    )
+    .await;
+    assert!(!gateway
+        .ctx
+        .stream_prefs
+        .is_enabled("imessage:dm:+15551234567"));
+    assert!(gateway
+        .ctx
+        .sent_replies
+        .lock()
+        .unwrap()
+        .last()
+        .is_some_and(|(_, text)| text.starts_with("Stream progress: off.")));
+    assert!(calls.lock().unwrap().is_empty());
+
+    let _ = std::fs::remove_file(&state_path);
+    let _ = std::fs::remove_file(format!("{state_path}.db"));
+    let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl"));
+    let _ = std::fs::remove_dir_all(sessions_dir);
+    let _ = std::fs::remove_dir_all(assistant_dir);
+}
+
 #[test]
 fn setup_failure_completion_unblocks_later_completed_rows() {
     let path = temp_state_path();