From a59771ef9915d68425c7b0ce53856457ac077901 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Sun, 2 Aug 2026 19:47:39 +0200 Subject: [PATCH 1/4] feat(gateway): add /stream cosmetic tool progress for Pi Toggle per conversation sends Hermes-style tool breadcrumbs from Pi's JSONL events without recording them in canonical history or changing LLM prompts. Co-authored-by: Cursor --- docs/prd.md | 3 + docs/reference/cli.md | 1 + src/agent.rs | 26 ++++- src/gateway/mod.rs | 7 ++ src/gateway/tests.rs | 65 +++++++++++ src/gateway/worker.rs | 101 ++++++++++++++++- src/main.rs | 1 + src/pi.rs | 257 +++++++++++++++++++++++++++++++----------- src/progress.rs | 161 ++++++++++++++++++++++++++ 9 files changed, 548 insertions(+), 74 deletions(-) create mode 100644 src/progress.rs diff --git a/docs/prd.md b/docs/prd.md index f503a57..b186ffe 100644 --- a/docs/prd.md +++ b/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 diff --git a/docs/reference/cli.md b/docs/reference/cli.md index cdc1b80..3240344 100644 --- a/docs/reference/cli.md +++ b/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 diff --git a/src/agent.rs b/src/agent.rs index 083d223..4caa680 100644 --- a/src/agent.rs +++ b/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(mut spawn: F) -> std::io::Result +pub(crate) async fn output_with_retry(mut spawn: F) -> std::io::Result where F: FnMut() -> Fut, - Fut: std::future::Future>, + Fut: std::future::Future>, { 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 { + 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>, + ) -> Result { 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 + } } } diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 33a3bcc..1879300 100644 --- a/src/gateway/mod.rs +++ b/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, voice: Option, schedule_destination: Option, + stream_prefs: StreamPrefs, #[cfg(test)] setup_failure_replies: Arc>>, #[cfg(test)] sent_replies: Arc>>, #[cfg(test)] + sent_progress: Arc>>, + #[cfg(test)] sent_voice_replies: SentVoiceReplies, #[cfg(test)] send_failures_remaining: Arc>, @@ -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)), diff --git a/src/gateway/tests.rs b/src/gateway/tests.rs index 21218bf..061b538 100644 --- a/src/gateway/tests.rs +++ b/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(); diff --git a/src/gateway/worker.rs b/src/gateway/worker.rs index 458184f..8f92251 100644 --- a/src/gateway/worker.rs +++ b/src/gateway/worker.rs @@ -3,7 +3,7 @@ use std::future::{pending, Future}; use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, Instant}; use anyhow::Result; use tokio::sync::{mpsc, watch}; @@ -11,6 +11,7 @@ use tracing::{error, info, warn}; use crate::agent::{Request, RunError}; use crate::history::{DeliveryStatus, OutboundMessage, OutboundOrigin}; +use crate::progress::{format_progress_line, ProgressEvent, ProgressPhase}; use crate::prompt::{ComposedPrompt, Composer}; use crate::voice::MAX_AUDIO_BYTES; @@ -220,6 +221,20 @@ where } }; + let progress_bundle = if ctx.stream_prefs.is_enabled(&job.thread) { + let (tx, rx) = mpsc::unbounded_channel(); + let pump_ctx = ctx.clone(); + let target = job.target.clone(); + let thread = job.thread.clone(); + let handle = tokio::spawn(async move { + pump_progress(pump_ctx, target, thread, rx).await; + }); + Some((tx, handle)) + } else { + None + }; + let progress = progress_bundle.as_ref().map(|(tx, _)| tx.clone()); + let run = async { let mut session_id = session_id; let mut prompt = match conversation_prompt(ctx, &job, &composer, is_new) { @@ -253,9 +268,10 @@ where prompt.rehydrated_messages ); let mut result = runner - .run( + .run_with_progress( backend_request(&session_id, is_new, &work_dir, &prompt), ctx.run_timeout, + progress.clone(), ) .await; // If the session id already exists (e.g. left over from a previous run @@ -268,9 +284,10 @@ where RunError::Failed(format!("compose resumed prompt: {error}")) })?; result = runner - .run( + .run_with_progress( backend_request(&session_id, false, &work_dir, &prompt), ctx.run_timeout, + progress.clone(), ) .await; } @@ -322,9 +339,10 @@ where ), ); result = runner - .run( + .run_with_progress( backend_request(&session_id, true, &work_dir, &prompt), ctx.run_timeout, + progress.clone(), ) .await; } @@ -357,6 +375,10 @@ where result = &mut run => Some(result), _ = &mut interrupt => None, }; + if let Some((tx, handle)) = progress_bundle { + drop(tx); + let _ = handle.await; + } match result { Some(Ok(out)) => { @@ -746,7 +768,8 @@ fn complete_job(ctx: &Ctx, job: &Job, reason: &str) { /// Handles gateway-level slash commands before anything reaches the agent. fn command(ctx: &Ctx, job: &Job) -> Option { - match job.text.trim().to_lowercase().as_str() { + let text = job.text.trim().to_lowercase(); + match text.as_str() { "/clear" | "/new" | "/reset" => match ctx.store.lock().unwrap().rotate( &job.thread, job.backend.as_str(), @@ -758,8 +781,25 @@ fn command(ctx: &Ctx, job: &Job) -> Option { Ok(()) => Some("Started a fresh conversation.".to_string()), Err(_) => Some("Couldn't reset the conversation.".to_string()), }, + "/stream" | "/stream on" | "/stream off" => { + let enabled = match text.as_str() { + "/stream on" => { + ctx.stream_prefs.set(&job.thread, true); + true + } + "/stream off" => { + ctx.stream_prefs.set(&job.thread, false); + false + } + _ => ctx.stream_prefs.toggle(&job.thread), + }; + Some(format!( + "Stream progress: {}.", + if enabled { "on" } else { "off" } + )) + } "/help" => Some( - "Commands:\n/clear - start a fresh conversation\n/stop - stop the active request\n/help - this message" + "Commands:\n/clear - start a fresh conversation\n/stop - stop the active request\n/stream - toggle cosmetic tool progress\n/help - this message" .to_string(), ), _ => None, @@ -1012,3 +1052,52 @@ fn backend_request<'a>( prompt: &prompt.content, } } + +const PROGRESS_MIN_INTERVAL: Duration = Duration::from_millis(300); + +async fn pump_progress( + ctx: Ctx, + target: String, + thread: String, + mut rx: mpsc::UnboundedReceiver, +) { + let mut last = Instant::now() + .checked_sub(PROGRESS_MIN_INTERVAL) + .unwrap_or_else(Instant::now); + while let Some(event) = rx.recv().await { + match &event.phase { + ProgressPhase::Start => {} + ProgressPhase::End { is_error: true } => {} + ProgressPhase::End { is_error: false } => continue, + } + let elapsed = last.elapsed(); + if elapsed < PROGRESS_MIN_INTERVAL { + tokio::time::sleep(PROGRESS_MIN_INTERVAL - elapsed).await; + } + let text = format_progress_line(&event); + if let Err(error) = send_progress_ephemeral(&ctx, &target, &text).await { + warn!("[{thread}] progress send failed: {error}"); + } + last = Instant::now(); + } +} + +/// Cosmetic progress delivery: never recorded in canonical history. +async fn send_progress_ephemeral(ctx: &Ctx, target: &str, text: &str) -> Result<()> { + #[cfg(test)] + { + ctx.sent_progress + .lock() + .unwrap() + .push((target.to_string(), text.to_string())); + Ok(()) + } + #[cfg(not(test))] + { + let chunks = ctx.channel.outbound_chunks(text, ""); + for chunk in &chunks { + super::send_reply_chunk(ctx, target, chunk).await?; + } + Ok(()) + } +} diff --git a/src/main.rs b/src/main.rs index 2b0a975..a45d29b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -19,6 +19,7 @@ mod jobs; mod markdown; mod paths; mod pi; +mod progress; mod prompt; mod restart; mod slack; diff --git a/src/pi.rs b/src/pi.rs index 4ec2f3a..f33757a 100644 --- a/src/pi.rs +++ b/src/pi.rs @@ -4,10 +4,12 @@ use std::time::Duration; use std::{io, process::Stdio}; use serde_json::Value; -use tokio::io::AsyncWriteExt; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::process::Command; +use tokio::sync::mpsc; use crate::agent::{final_reply, Request, RunError, RunOutput}; +use crate::progress::{preview_from_args, ProgressEvent, ProgressPhase}; /// Runner invokes `pi` in non-interactive JSON event mode. pub struct Runner { @@ -23,8 +25,19 @@ enum RunMode { impl Runner { /// Executes one turn and returns Pi's final reply plus its stable session id. + #[cfg_attr(not(test), allow(dead_code))] pub async fn run(&self, req: Request<'_>, timeout: Duration) -> Result { - self.run_with_mode(req, timeout, RunMode::Configured).await + self.run_with_progress(req, timeout, None).await + } + + pub async fn run_with_progress( + &self, + req: Request<'_>, + timeout: Duration, + progress: Option>, + ) -> Result { + self.run_with_mode(req, timeout, RunMode::Configured, progress) + .await } pub async fn run_unattended( @@ -39,6 +52,7 @@ impl Runner { RunMode::Unattended { trust_project_resources, }, + None, ) .await } @@ -48,7 +62,8 @@ impl Runner { req: Request<'_>, timeout: Duration, ) -> Result { - self.run_with_mode(req, timeout, RunMode::Evaluator).await + self.run_with_mode(req, timeout, RunMode::Evaluator, None) + .await } async fn run_with_mode( @@ -56,23 +71,13 @@ impl Runner { req: Request<'_>, timeout: Duration, mode: RunMode, + progress: Option>, ) -> Result { let attempt = crate::agent::output_with_retry(|| { let mut cmd = self.command(&req, mode); let prompt = req.prompt.as_bytes().to_vec(); - async move { - let mut child = cmd.spawn()?; - let mut stdin = child.stdin.take().ok_or_else(|| { - io::Error::new(io::ErrorKind::BrokenPipe, "pi stdin unavailable") - })?; - let write_result = stdin.write_all(&prompt).await; - drop(stdin); - let output = child.wait_with_output().await?; - if output.status.success() { - write_result?; - } - Ok(output) - } + let progress = progress.clone(); + async move { run_child(&mut cmd, &prompt, progress).await } }); let out = match tokio::time::timeout(timeout, attempt).await { Err(_) => return Err(RunError::Timeout), @@ -80,7 +85,6 @@ impl Runner { Ok(Ok(output)) => output, }; - let parsed = parse_jsonl(&out.stdout); if !out.status.success() { let stderr = String::from_utf8_lossy(&out.stderr); if !req.is_new && missing_resume_error(&stderr) { @@ -91,27 +95,29 @@ impl Runner { } return Err(RunError::Failed(exit_diagnostic( out.status.code(), - parsed.as_ref().err().map(String::as_str), + out.parse_error.as_deref(), ))); } - let parsed = parsed.map_err(RunError::Failed)?; - if req.is_new && parsed.session_id.is_none() { + if let Some(error) = out.parse_error { + return Err(RunError::Failed(error)); + } + if req.is_new && out.parsed.session_id.is_none() { return Err(RunError::Failed( "pi did not report a session id".to_string(), )); } - if parsed.assistant_failed { + if out.parsed.assistant_failed { return Err(RunError::Failed( "Pi assistant request failed; check Pi provider and authentication settings" .to_string(), )); } - let reply = parsed.reply.unwrap_or_default(); + let reply = out.parsed.reply.unwrap_or_default(); Ok(RunOutput { reply: final_reply("pi", &reply)?, - session_id: req.is_new.then_some(parsed.session_id).flatten(), + session_id: req.is_new.then_some(out.parsed.session_id).flatten(), }) } @@ -152,6 +158,70 @@ impl Runner { } } +struct ChildOutput { + status: std::process::ExitStatus, + stderr: Vec, + parsed: ParsedOutput, + parse_error: Option, +} + +async fn run_child( + cmd: &mut Command, + prompt: &[u8], + progress: Option>, +) -> io::Result { + let mut child = cmd.spawn()?; + let mut stdin = child + .stdin + .take() + .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "pi stdin unavailable"))?; + let write_result = stdin.write_all(prompt).await; + drop(stdin); + + let stdout = child + .stdout + .take() + .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "pi stdout unavailable"))?; + let mut stderr_pipe = child + .stderr + .take() + .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "pi stderr unavailable"))?; + + let stderr_task = tokio::spawn(async move { + let mut stderr = Vec::new(); + let _ = stderr_pipe.read_to_end(&mut stderr).await; + stderr + }); + + let mut parsed = ParsedOutput::default(); + let mut parse_error = None; + let mut lines = BufReader::new(stdout).lines(); + while let Some(line) = lines.next_line().await? { + if line.trim().is_empty() { + continue; + } + match apply_jsonl_line(&mut parsed, &line, progress.as_ref()) { + Ok(()) => {} + Err(error) => { + parse_error = Some(error); + break; + } + } + } + + let status = child.wait().await?; + let stderr = stderr_task.await.unwrap_or_default(); + if status.success() { + write_result?; + } + Ok(ChildOutput { + status, + stderr, + parsed, + parse_error, + }) +} + #[derive(Default)] struct ParsedOutput { session_id: Option, @@ -159,53 +229,84 @@ struct ParsedOutput { assistant_failed: bool, } -fn parse_jsonl(stdout: &[u8]) -> Result { - let stdout = std::str::from_utf8(stdout) - .map_err(|_| "pi returned malformed JSON output (invalid UTF-8)".to_string())?; - let mut parsed = ParsedOutput::default(); - for line in stdout.lines().filter(|line| !line.trim().is_empty()) { - let event: Value = serde_json::from_str(line) - .map_err(|_| "pi returned malformed JSON output".to_string())?; - match event.get("type").and_then(Value::as_str) { - Some("session") => { - parsed.session_id = event - .get("id") +fn apply_jsonl_line( + parsed: &mut ParsedOutput, + line: &str, + progress: Option<&mpsc::UnboundedSender>, +) -> Result<(), String> { + let event: Value = + serde_json::from_str(line).map_err(|_| "pi returned malformed JSON output".to_string())?; + match event.get("type").and_then(Value::as_str) { + Some("session") => { + parsed.session_id = event + .get("id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|id| !id.is_empty()) + .map(str::to_string); + } + Some("message_end") => { + let Some(message) = event.get("message") else { + return Ok(()); + }; + if message.get("role").and_then(Value::as_str) != Some("assistant") { + return Ok(()); + } + if matches!( + message.get("stopReason").and_then(Value::as_str), + Some("error" | "aborted") + ) { + parsed.reply = None; + parsed.assistant_failed = true; + return Ok(()); + } + let text = message + .get("content") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter(|part| part.get("type").and_then(Value::as_str) == Some("text")) + .filter_map(|part| part.get("text").and_then(Value::as_str)) + .collect::>() + .join(""); + parsed.reply = Some(text); + parsed.assistant_failed = false; + } + Some("tool_execution_start") => { + if let Some(tx) = progress { + let tool_name = event + .get("toolName") .and_then(Value::as_str) - .map(str::trim) - .filter(|id| !id.is_empty()) - .map(str::to_string); + .unwrap_or("") + .to_string(); + let args = event.get("args").cloned().unwrap_or(Value::Null); + let _ = tx.send(ProgressEvent { + preview: preview_from_args(&tool_name, &args), + tool_name, + phase: ProgressPhase::Start, + }); } - Some("message_end") => { - let Some(message) = event.get("message") else { - continue; - }; - if message.get("role").and_then(Value::as_str) != Some("assistant") { - continue; - } - if matches!( - message.get("stopReason").and_then(Value::as_str), - Some("error" | "aborted") - ) { - parsed.reply = None; - parsed.assistant_failed = true; - continue; + } + Some("tool_execution_end") => { + if let Some(tx) = progress { + if event.get("isError").and_then(Value::as_bool) == Some(true) { + let tool_name = event + .get("toolName") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + let args = event.get("args").cloned().unwrap_or(Value::Null); + let _ = tx.send(ProgressEvent { + preview: preview_from_args(&tool_name, &args), + tool_name, + phase: ProgressPhase::End { is_error: true }, + }); } - let text = message - .get("content") - .and_then(Value::as_array) - .into_iter() - .flatten() - .filter(|part| part.get("type").and_then(Value::as_str) == Some("text")) - .filter_map(|part| part.get("text").and_then(Value::as_str)) - .collect::>() - .join(""); - parsed.reply = Some(text); - parsed.assistant_failed = false; } - _ => {} } + _ => {} } - Ok(parsed) + Ok(()) } fn exit_diagnostic(status: Option, parse_error: Option<&str>) -> String { @@ -295,6 +396,34 @@ mod tests { assert!(!args.contains(&prompt)); } + #[tokio::test] + async fn streams_tool_progress_events_before_final_reply() { + let work_dir = temp_dir("pi-progress-work"); + let script = "#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' '{\"type\":\"session\",\"id\":\"progress-session\"}'\nprintf '%s\\n' '{\"type\":\"tool_execution_start\",\"toolCallId\":\"1\",\"toolName\":\"bash\",\"args\":{\"command\":\"ls -la\"}}'\nprintf '%s\\n' '{\"type\":\"tool_execution_end\",\"toolCallId\":\"1\",\"toolName\":\"bash\",\"result\":null,\"isError\":false}'\nprintf '%s\\n' '{\"type\":\"message_end\",\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"done\"}],\"stopReason\":\"stop\"}}'\n"; + let cli = FakeCli::new("pi", script); + let runner = Runner { bin: cli.bin() }; + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + + let output = runner + .run_with_progress( + request(work_dir.to_str().unwrap(), true), + Duration::from_secs(5), + Some(tx), + ) + .await + .unwrap(); + + assert_eq!(output.reply, "done"); + let event = rx.recv().await.expect("progress event"); + assert_eq!(event.tool_name, "bash"); + assert_eq!(event.preview, "ls -la"); + assert!(matches!(event.phase, crate::progress::ProgressPhase::Start)); + assert!( + rx.try_recv().is_err(), + "successful tool end should not emit" + ); + } + #[tokio::test] async fn unattended_run_trusts_project_local_resources() { let args_path = temp_path("pi-unattended-args"); diff --git a/src/progress.rs b/src/progress.rs new file mode 100644 index 0000000..b3a0d80 --- /dev/null +++ b/src/progress.rs @@ -0,0 +1,161 @@ +//! Cosmetic tool-progress feedback for chat. Never enters canonical history. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +/// One tool lifecycle event from a backend JSON stream. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProgressEvent { + pub tool_name: String, + pub preview: String, + pub phase: ProgressPhase, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ProgressPhase { + Start, + End { is_error: bool }, +} + +/// Per-conversation `/stream` toggle. In-memory only; restarts reset to off. +#[derive(Clone, Default)] +pub struct StreamPrefs { + enabled: Arc>>, +} + +impl StreamPrefs { + pub fn is_enabled(&self, thread: &str) -> bool { + self.enabled + .lock() + .unwrap() + .get(thread) + .copied() + .unwrap_or(false) + } + + pub fn set(&self, thread: &str, enabled: bool) { + self.enabled + .lock() + .unwrap() + .insert(thread.to_string(), enabled); + } + + pub fn toggle(&self, thread: &str) -> bool { + let enabled = !self.is_enabled(thread); + self.set(thread, enabled); + enabled + } +} + +const PREVIEW_MAX: usize = 100; + +/// Hermes-style cosmetic line: `⚡ Running ls -la …` +pub fn format_progress_line(event: &ProgressEvent) -> String { + let verb = tool_verb(&event.tool_name); + let body = if event.preview.is_empty() { + verb + } else { + format!("{verb} {}", &event.preview) + }; + let body = truncate_one_line(&body, PREVIEW_MAX); + match &event.phase { + ProgressPhase::Start => format!("⚡ {body}"), + ProgressPhase::End { is_error: true } => format!("⚡ Failed {body}"), + ProgressPhase::End { is_error: false } => format!("⚡ Done {body}"), + } +} + +fn tool_verb(tool_name: &str) -> String { + match tool_name { + "bash" | "terminal" | "execute_code" => "Running".to_string(), + "read" | "read_file" => "Reading".to_string(), + "write" | "write_file" => "Writing".to_string(), + "edit" | "patch" => "Editing".to_string(), + "grep" | "find" | "search_files" | "web_search" => "Searching".to_string(), + "ls" => "Listing".to_string(), + "" => "Working".to_string(), + other => format!("Using {other}"), + } +} + +/// Build a short one-line preview from Pi/tool JSON args. +pub fn preview_from_args(tool_name: &str, args: &serde_json::Value) -> String { + let raw = match tool_name { + "bash" | "terminal" => args + .get("command") + .or_else(|| args.get("cmd")) + .and_then(|v| v.as_str()) + .unwrap_or(""), + "read" | "read_file" | "write" | "write_file" | "edit" | "patch" | "ls" => args + .get("path") + .or_else(|| args.get("file_path")) + .or_else(|| args.get("file")) + .and_then(|v| v.as_str()) + .unwrap_or(""), + "grep" | "search_files" | "web_search" | "find" => args + .get("pattern") + .or_else(|| args.get("query")) + .or_else(|| args.get("path")) + .and_then(|v| v.as_str()) + .unwrap_or(""), + _ => args + .as_object() + .and_then(|obj| { + obj.values() + .find_map(|v| v.as_str()) + .or_else(|| obj.keys().next().map(|k| k.as_str())) + }) + .unwrap_or(""), + }; + truncate_one_line(raw, PREVIEW_MAX) +} + +fn truncate_one_line(text: &str, max: usize) -> String { + let flat: String = text.split_whitespace().collect::>().join(" "); + if flat.chars().count() <= max { + return flat; + } + let mut out = String::new(); + for ch in flat.chars() { + if out.chars().count() + 1 >= max.saturating_sub(1) { + break; + } + out.push(ch); + } + out.push('…'); + out +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn stream_prefs_default_off_and_toggle() { + let prefs = StreamPrefs::default(); + assert!(!prefs.is_enabled("telegram:dm:1")); + assert!(prefs.toggle("telegram:dm:1")); + assert!(prefs.is_enabled("telegram:dm:1")); + prefs.set("telegram:dm:1", false); + assert!(!prefs.is_enabled("telegram:dm:1")); + } + + #[test] + fn formats_hermes_style_bash_line() { + let line = format_progress_line(&ProgressEvent { + tool_name: "bash".into(), + preview: "curl https://example.com".into(), + phase: ProgressPhase::Start, + }); + assert_eq!(line, "⚡ Running curl https://example.com"); + } + + #[test] + fn preview_prefers_command_for_bash() { + assert_eq!( + preview_from_args("bash", &json!({"command": "ls -la"})), + "ls -la" + ); + } +} From 68f362d8daf74c2617d1cf74a18012028aa478eb Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Sun, 2 Aug 2026 20:10:19 +0200 Subject: [PATCH 2/4] fix(gateway): accept Telegram /cmd@bot slash forms Menu taps send /stream@BotName; normalize before matching so commands still work. Co-authored-by: Cursor --- src/gateway/mod.rs | 2 +- src/gateway/worker.rs | 19 ++++++++++++++++++- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 1879300..9215285 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -808,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; } diff --git a/src/gateway/worker.rs b/src/gateway/worker.rs index 8f92251..6faf3a0 100644 --- a/src/gateway/worker.rs +++ b/src/gateway/worker.rs @@ -768,7 +768,7 @@ fn complete_job(ctx: &Ctx, job: &Job, reason: &str) { /// Handles gateway-level slash commands before anything reaches the agent. fn command(ctx: &Ctx, job: &Job) -> Option { - let text = job.text.trim().to_lowercase(); + let text = normalize_slash_command(&job.text); match text.as_str() { "/clear" | "/new" | "/reset" => match ctx.store.lock().unwrap().rotate( &job.thread, @@ -806,6 +806,23 @@ fn command(ctx: &Ctx, job: &Job) -> Option { } } +/// Strip Telegram `@botname` suffixes so `/stream@MyBot` matches `/stream`. +pub(super) fn normalize_slash_command(text: &str) -> String { + let text = text.trim().to_lowercase(); + let Some(rest) = text.strip_prefix('/') else { + return text; + }; + let mut parts = rest.splitn(2, char::is_whitespace); + let cmd = parts.next().unwrap_or(""); + let args = parts.next().unwrap_or("").trim(); + let cmd = cmd.split('@').next().unwrap_or(cmd); + if args.is_empty() { + format!("/{cmd}") + } else { + format!("/{cmd} {args}") + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(super) enum DeliveryOutcome { Delivered, From a0591d3c955b78a828327a77db221a23107c8844 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Sun, 2 Aug 2026 20:31:00 +0200 Subject: [PATCH 3/4] fix(gateway): edit one Telegram progress message with Hermes Shell fences Accumulate tool progress into a single editable bubble (Shell + code fence) to avoid rate-limit stalls from many sends, and keep final replies unblocked. Co-authored-by: Cursor --- src/channel.rs | 23 ++++++++ src/gateway/worker.rs | 103 ++++++++++++++++++++++++++---------- src/progress.rs | 120 +++++++++++++++++++++++++++++++++--------- src/telegram.rs | 74 +++++++++++++++++++++++++- 4 files changed, 264 insertions(+), 56 deletions(-) diff --git a/src/channel.rs b/src/channel.rs index 3306ffe..bacc174 100644 --- a/src/channel.rs +++ b/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> { + 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 { match self { Self::IMessage(channel) => ChannelContract::download_voice(channel, voice).await, diff --git a/src/gateway/worker.rs b/src/gateway/worker.rs index 6faf3a0..e8f5ff0 100644 --- a/src/gateway/worker.rs +++ b/src/gateway/worker.rs @@ -11,7 +11,9 @@ use tracing::{error, info, warn}; use crate::agent::{Request, RunError}; use crate::history::{DeliveryStatus, OutboundMessage, OutboundOrigin}; -use crate::progress::{format_progress_line, ProgressEvent, ProgressPhase}; +use crate::progress::{ + append_progress_message, format_progress_block, ProgressEvent, ProgressPhase, +}; use crate::prompt::{ComposedPrompt, Composer}; use crate::voice::MAX_AUDIO_BYTES; @@ -377,7 +379,11 @@ where }; if let Some((tx, handle)) = progress_bundle { drop(tx); - let _ = handle.await; + // Never let a stuck progress edit block the final reply. + match tokio::time::timeout(Duration::from_secs(5), handle).await { + Ok(_) => {} + Err(_) => warn!("[{}] progress pump timed out after backend run", job.thread), + } } match result { @@ -1070,7 +1076,8 @@ fn backend_request<'a>( } } -const PROGRESS_MIN_INTERVAL: Duration = Duration::from_millis(300); +const PROGRESS_EDIT_INTERVAL: Duration = Duration::from_millis(450); +const PROGRESS_SEND_TIMEOUT: Duration = Duration::from_secs(8); async fn pump_progress( ctx: Ctx, @@ -1078,43 +1085,83 @@ async fn pump_progress( thread: String, mut rx: mpsc::UnboundedReceiver, ) { - let mut last = Instant::now() - .checked_sub(PROGRESS_MIN_INTERVAL) + let mut body = String::new(); + let mut message_id: Option = None; + let mut last_flush = Instant::now() + .checked_sub(PROGRESS_EDIT_INTERVAL) .unwrap_or_else(Instant::now); + let mut dirty = false; + while let Some(event) = rx.recv().await { match &event.phase { ProgressPhase::Start => {} ProgressPhase::End { is_error: true } => {} ProgressPhase::End { is_error: false } => continue, } - let elapsed = last.elapsed(); - if elapsed < PROGRESS_MIN_INTERVAL { - tokio::time::sleep(PROGRESS_MIN_INTERVAL - elapsed).await; + let block = format_progress_block(&event); + body = append_progress_message(&body, &block); + dirty = true; + + let elapsed = last_flush.elapsed(); + if message_id.is_some() && elapsed < PROGRESS_EDIT_INTERVAL { + continue; } - let text = format_progress_line(&event); - if let Err(error) = send_progress_ephemeral(&ctx, &target, &text).await { - warn!("[{thread}] progress send failed: {error}"); + if flush_progress(&ctx, &target, &thread, &body, &mut message_id).await { + last_flush = Instant::now(); + dirty = false; } - last = Instant::now(); } -} -/// Cosmetic progress delivery: never recorded in canonical history. -async fn send_progress_ephemeral(ctx: &Ctx, target: &str, text: &str) -> Result<()> { - #[cfg(test)] - { - ctx.sent_progress - .lock() - .unwrap() - .push((target.to_string(), text.to_string())); - Ok(()) + if dirty && !body.is_empty() { + let _ = flush_progress(&ctx, &target, &thread, &body, &mut message_id).await; } - #[cfg(not(test))] - { - let chunks = ctx.channel.outbound_chunks(text, ""); - for chunk in &chunks { - super::send_reply_chunk(ctx, target, chunk).await?; +} + +async fn flush_progress( + ctx: &Ctx, + target: &str, + thread: &str, + body: &str, + message_id: &mut Option, +) -> bool { + let send = async { + match message_id { + Some(id) => { + ctx.channel.edit_progress(target, *id, body).await?; + Ok::<_, anyhow::Error>(None) + } + None => Ok(ctx.channel.send_progress(target, body).await?), + } + }; + match tokio::time::timeout(PROGRESS_SEND_TIMEOUT, send).await { + Ok(Ok(Some(id))) => { + *message_id = Some(id); + #[cfg(test)] + { + ctx.sent_progress + .lock() + .unwrap() + .push((target.to_string(), body.to_string())); + } + true + } + Ok(Ok(None)) => { + #[cfg(test)] + { + ctx.sent_progress + .lock() + .unwrap() + .push((target.to_string(), body.to_string())); + } + true + } + Ok(Err(error)) => { + warn!("[{thread}] progress update failed: {error}"); + false + } + Err(_) => { + warn!("[{thread}] progress update timed out"); + false } - Ok(()) } } diff --git a/src/progress.rs b/src/progress.rs index b3a0d80..e1f2216 100644 --- a/src/progress.rs +++ b/src/progress.rs @@ -47,34 +47,84 @@ impl StreamPrefs { } } -const PREVIEW_MAX: usize = 100; +const PREVIEW_MAX: usize = 120; +const PROGRESS_MESSAGE_MAX: usize = 3500; + +/// Hermes-style Markdown block for one tool start (Telegram renders fences as copyable). +pub fn format_progress_block(event: &ProgressEvent) -> String { + let preview = truncate_one_line(&event.preview, PREVIEW_MAX); + let failed = matches!(event.phase, ProgressPhase::End { is_error: true }); + let prefix = if failed { "⚠️ " } else { "" }; + match event.tool_name.as_str() { + "bash" | "terminal" | "execute_code" => { + let cmd = if preview.is_empty() { "…" } else { &preview }; + format!("{prefix}💻 Shell\n```\n{cmd}\n```") + } + "read" | "read_file" => { + let path = if preview.is_empty() { "…" } else { &preview }; + format!("{prefix}📖 read\n`{path}`") + } + "write" | "write_file" => { + let path = if preview.is_empty() { "…" } else { &preview }; + format!("{prefix}✍️ write\n`{path}`") + } + "edit" | "patch" => { + let path = if preview.is_empty() { "…" } else { &preview }; + format!("{prefix}✏️ edit\n`{path}`") + } + "grep" | "find" | "search_files" | "web_search" => { + let q = if preview.is_empty() { "…" } else { &preview }; + format!("{prefix}🔍 search\n`{q}`") + } + "ls" => { + let path = if preview.is_empty() { "." } else { &preview }; + format!("{prefix}📁 ls\n`{path}`") + } + other => { + let name = if other.is_empty() { "tool" } else { other }; + if preview.is_empty() { + format!("{prefix}⚙️ {name}") + } else { + format!("{prefix}⚙️ {name}\n`{preview}`") + } + } + } +} -/// Hermes-style cosmetic line: `⚡ Running ls -la …` -pub fn format_progress_line(event: &ProgressEvent) -> String { - let verb = tool_verb(&event.tool_name); - let body = if event.preview.is_empty() { - verb +/// Append a new tool block into an accumulating progress message. +pub fn append_progress_message(existing: &str, block: &str) -> String { + let combined = if existing.is_empty() { + block.to_string() } else { - format!("{verb} {}", &event.preview) + format!("{existing}\n\n{block}") }; - let body = truncate_one_line(&body, PREVIEW_MAX); - match &event.phase { - ProgressPhase::Start => format!("⚡ {body}"), - ProgressPhase::End { is_error: true } => format!("⚡ Failed {body}"), - ProgressPhase::End { is_error: false } => format!("⚡ Done {body}"), - } + trim_progress_message(&combined, PROGRESS_MESSAGE_MAX) } -fn tool_verb(tool_name: &str) -> String { - match tool_name { - "bash" | "terminal" | "execute_code" => "Running".to_string(), - "read" | "read_file" => "Reading".to_string(), - "write" | "write_file" => "Writing".to_string(), - "edit" | "patch" => "Editing".to_string(), - "grep" | "find" | "search_files" | "web_search" => "Searching".to_string(), - "ls" => "Listing".to_string(), - "" => "Working".to_string(), - other => format!("Using {other}"), +fn trim_progress_message(text: &str, max: usize) -> String { + if text.chars().count() <= max { + return text.to_string(); + } + let mut blocks: Vec<&str> = text.split("\n\n").collect(); + while blocks.len() > 1 && blocks.join("\n\n").chars().count() > max { + blocks.remove(0); + } + let joined = blocks.join("\n\n"); + if joined.chars().count() <= max { + format!("…\n\n{joined}") + } else { + let mut out = String::from("…\n\n"); + for ch in joined + .chars() + .rev() + .take(max.saturating_sub(4)) + .collect::>() + .into_iter() + .rev() + { + out.push(ch); + } + out } } @@ -142,13 +192,14 @@ mod tests { } #[test] - fn formats_hermes_style_bash_line() { - let line = format_progress_line(&ProgressEvent { + fn formats_hermes_style_shell_fence() { + let block = format_progress_block(&ProgressEvent { tool_name: "bash".into(), preview: "curl https://example.com".into(), phase: ProgressPhase::Start, }); - assert_eq!(line, "⚡ Running curl https://example.com"); + assert!(block.contains("💻 Shell")); + assert!(block.contains("```\ncurl https://example.com\n```")); } #[test] @@ -158,4 +209,21 @@ mod tests { "ls -la" ); } + + #[test] + fn append_keeps_multiple_tool_blocks() { + let first = format_progress_block(&ProgressEvent { + tool_name: "bash".into(), + preview: "ls".into(), + phase: ProgressPhase::Start, + }); + let second = format_progress_block(&ProgressEvent { + tool_name: "read".into(), + preview: "a.txt".into(), + phase: ProgressPhase::Start, + }); + let msg = append_progress_message(&first, &second); + assert!(msg.contains("Shell")); + assert!(msg.contains("read")); + } } diff --git a/src/telegram.rs b/src/telegram.rs index e85fbdd..dc037df 100644 --- a/src/telegram.rs +++ b/src/telegram.rs @@ -249,6 +249,10 @@ impl Telegram { self.allow_user_ids.contains(&chat_id) || self.allow_chat_ids.contains(&chat_id) } + pub async fn send_plain(&self, target: &str, text: &str) -> Result<()> { + self.send_plain_with_id(target, text).await.map(|_| ()) + } + pub async fn send_rich(&self, target: &str, text: &str) -> Result<()> { if text.encode_utf16().count() > TEXT_LIMIT { bail!("Telegram rich message exceeds the {TEXT_LIMIT} character chunk limit"); @@ -271,9 +275,70 @@ impl Telegram { Ok(()) } - pub async fn send_plain(&self, target: &str, text: &str) -> Result<()> { + /// Send a rich progress bubble and return Telegram's message id for later edits. + #[cfg_attr(test, allow(dead_code))] + pub async fn send_progress(&self, target: &str, text: &str) -> Result { + if text.encode_utf16().count() > TEXT_LIMIT { + bail!("Telegram rich message exceeds the {TEXT_LIMIT} character chunk limit"); + } + let html = crate::markdown::to_telegram_html(text); + let mut payload = target_payload(target); + payload["text"] = json!(html); + payload["parse_mode"] = json!("HTML"); + match self.send_message_id(payload).await { + Ok(id) => Ok(id), + Err(_) => self.send_plain_with_id(target, text).await, + } + } + + /// Edit an existing progress bubble in place. + #[cfg_attr(test, allow(dead_code))] + pub async fn edit_progress(&self, target: &str, message_id: i64, text: &str) -> Result<()> { + if text.encode_utf16().count() > TEXT_LIMIT { + bail!("Telegram rich message exceeds the {TEXT_LIMIT} character chunk limit"); + } + let html = crate::markdown::to_telegram_html(text); + let mut payload = target_payload(target); + payload["message_id"] = json!(message_id); + payload["text"] = json!(html); + payload["parse_mode"] = json!("HTML"); + let transport_response = self + .post_with_topic_fallback("editMessageText", payload) + .await?; + let response: ApiResponse = serde_json::from_value(transport_response.body.clone()) + .map_err(|_| { + anyhow::anyhow!("Telegram editMessageText returned an invalid response") + })?; + if response.ok { + return Ok(()); + } + // Fall back to plain edit if HTML is rejected. + let mut plain = target_payload(target); + plain["message_id"] = json!(message_id); + plain["text"] = json!(text); + let transport_response = self + .post_with_topic_fallback("editMessageText", plain) + .await?; + let response: ApiResponse = serde_json::from_value(transport_response.body) + .map_err(|_| { + anyhow::anyhow!("Telegram editMessageText returned an invalid response") + })?; + if !response.ok { + bail!( + "Telegram editMessageText returned HTTP {}", + transport_response.status + ); + } + Ok(()) + } + + async fn send_plain_with_id(&self, target: &str, text: &str) -> Result { let mut payload = target_payload(target); payload["text"] = json!(text); + self.send_message_id(payload).await + } + + async fn send_message_id(&self, payload: Value) -> Result { let transport_response = self .post_with_topic_fallback("sendMessage", payload) .await?; @@ -285,7 +350,12 @@ impl Telegram { transport_response.status ); } - Ok(()) + response + .result + .as_ref() + .and_then(|value| value.get("message_id")) + .and_then(Value::as_i64) + .context("Telegram sendMessage omitted message_id") } pub async fn send_typing(&self, target: &str) -> Result<()> { From 495a413a28a1c7fc78f7f73d3c8e20bf9c6df00f Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Mon, 3 Aug 2026 13:30:04 +0200 Subject: [PATCH 4/4] fix(telegram): keep send_plain independent of message_id parsing Progress edits need message ids; ordinary delivery must keep working with API mocks and responses that only return ok=true. Co-authored-by: Cursor --- src/telegram.rs | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/src/telegram.rs b/src/telegram.rs index dc037df..149efd8 100644 --- a/src/telegram.rs +++ b/src/telegram.rs @@ -250,7 +250,20 @@ impl Telegram { } pub async fn send_plain(&self, target: &str, text: &str) -> Result<()> { - self.send_plain_with_id(target, text).await.map(|_| ()) + let mut payload = target_payload(target); + payload["text"] = json!(text); + let transport_response = self + .post_with_topic_fallback("sendMessage", payload) + .await?; + let response: ApiResponse = serde_json::from_value(transport_response.body) + .map_err(|_| anyhow::anyhow!("Telegram sendMessage returned an invalid response"))?; + if !response.ok { + bail!( + "Telegram sendMessage returned HTTP {}", + transport_response.status + ); + } + Ok(()) } pub async fn send_rich(&self, target: &str, text: &str) -> Result<()> {