From 35959165b887e8ee4ecf40a773cb97edfbc0aa53 Mon Sep 17 00:00:00 2001 From: calvinsturm Date: Tue, 30 Jun 2026 18:31:04 -0700 Subject: [PATCH] Stream shell output to the TUI tail --- src/agent/tool_helpers.rs | 379 +++++++++++++++++++++++++++++++++++++- src/agent_tool_exec.rs | 9 +- src/events.rs | 1 + src/target.rs | 155 +++++++++++++++- src/tools.rs | 14 +- src/tools/exec_shell.rs | 15 +- src/tui/state.rs | 1 + src/tui/state/events.rs | 31 ++++ src/tui/state/tests.rs | 44 +++++ 9 files changed, 633 insertions(+), 16 deletions(-) diff --git a/src/agent/tool_helpers.rs b/src/agent/tool_helpers.rs index 7927dae..d092491 100644 --- a/src/agent/tool_helpers.rs +++ b/src/agent/tool_helpers.rs @@ -101,7 +101,292 @@ pub(super) enum FailedRepeatGuardDecision { Finalize(Box), } +/// Total live shell output bytes forwarded to the event stream per command. +/// Bounds memory and event-log growth; the full output still appears in the +/// final result envelope (subject to its own truncation policy). +const SHELL_STREAM_MAX_BYTES: usize = 64 * 1024; +const SHELL_STREAM_CHANNEL_CAPACITY: usize = 128; + +#[derive(Default)] +struct ShellUtf8Decoder { + pending: Vec, +} + +impl ShellUtf8Decoder { + fn push(&mut self, bytes: &[u8]) -> String { + self.pending.extend_from_slice(bytes); + let mut out = String::new(); + loop { + match std::str::from_utf8(&self.pending) { + Ok(s) => { + out.push_str(s); + self.pending.clear(); + break; + } + Err(err) => { + let valid_up_to = err.valid_up_to(); + if valid_up_to > 0 { + out.push_str( + std::str::from_utf8(&self.pending[..valid_up_to]) + .expect("valid prefix from utf8 error"), + ); + self.pending.drain(..valid_up_to); + } + match err.error_len() { + Some(len) => { + out.push('\u{FFFD}'); + let drain = len.min(self.pending.len()); + self.pending.drain(..drain); + } + None => break, + } + } + } + } + out + } + + fn flush_lossy(&mut self) -> String { + if self.pending.is_empty() { + return String::new(); + } + let out = String::from_utf8_lossy(&self.pending).to_string(); + self.pending.clear(); + out + } +} + +/// Coalesces incremental shell output chunks into `ShellOutputChunk` events and +/// enforces a total byte budget so live streaming cannot grow unbounded. +struct ShellStreamCoalescer { + tool_call_id: String, + streamed_bytes: usize, + budget_notice_emitted: bool, + stdout_decoder: ShellUtf8Decoder, + stderr_decoder: ShellUtf8Decoder, +} + +#[derive(Debug, PartialEq, Eq)] +enum ShellStreamAction { + Emit { stream: &'static str, text: String }, + BudgetReached, +} + +impl ShellStreamCoalescer { + fn new(tool_call_id: String) -> Self { + Self { + tool_call_id, + streamed_bytes: 0, + budget_notice_emitted: false, + stdout_decoder: ShellUtf8Decoder::default(), + stderr_decoder: ShellUtf8Decoder::default(), + } + } + + fn decoder_mut(&mut self, kind: crate::target::ShellStreamKind) -> &mut ShellUtf8Decoder { + match kind { + crate::target::ShellStreamKind::Stdout => &mut self.stdout_decoder, + crate::target::ShellStreamKind::Stderr => &mut self.stderr_decoder, + } + } + + fn emit_text( + &mut self, + kind: crate::target::ShellStreamKind, + text: String, + actions: &mut Vec, + ) { + if text.is_empty() { + return; + } + if self.streamed_bytes >= SHELL_STREAM_MAX_BYTES { + self.emit_budget_notice(actions); + return; + } + let remaining = SHELL_STREAM_MAX_BYTES - self.streamed_bytes; + let (text, truncated) = truncate_str_to_utf8_budget(&text, remaining); + if !text.is_empty() { + self.streamed_bytes += text.len(); + actions.push(ShellStreamAction::Emit { + stream: kind.as_str(), + text, + }); + } + if truncated { + self.emit_budget_notice(actions); + } + } + + fn emit_budget_notice(&mut self, actions: &mut Vec) { + if !self.budget_notice_emitted { + self.budget_notice_emitted = true; + actions.push(ShellStreamAction::BudgetReached); + } + } + + fn ingest_merged( + &mut self, + kind: crate::target::ShellStreamKind, + bytes: &[u8], + actions: &mut Vec, + ) { + let text = self.decoder_mut(kind).push(bytes); + self.emit_text(kind, text, actions); + } + + /// Merge adjacent chunks from the same stream and return display actions to + /// emit, respecting UTF-8 boundaries and the live byte budget. + fn ingest(&mut self, batch: &[crate::target::ShellOutputChunk]) -> Vec { + let mut actions = Vec::new(); + let mut current_stream = None; + let mut merged = Vec::new(); + for chunk in batch { + if current_stream == Some(chunk.stream) { + merged.extend_from_slice(&chunk.bytes); + } else { + if let Some(kind) = current_stream { + self.ingest_merged(kind, &merged, &mut actions); + merged.clear(); + } + current_stream = Some(chunk.stream); + merged.extend_from_slice(&chunk.bytes); + } + } + if let Some(kind) = current_stream { + self.ingest_merged(kind, &merged, &mut actions); + } + actions + } + + fn flush_pending(&mut self) -> Vec { + let mut actions = Vec::new(); + let stdout = self.stdout_decoder.flush_lossy(); + self.emit_text(crate::target::ShellStreamKind::Stdout, stdout, &mut actions); + let stderr = self.stderr_decoder.flush_lossy(); + self.emit_text(crate::target::ShellStreamKind::Stderr, stderr, &mut actions); + actions + } +} + +fn truncate_str_to_utf8_budget(input: &str, max_bytes: usize) -> (String, bool) { + if input.len() <= max_bytes { + return (input.to_string(), false); + } + let mut end = max_bytes.min(input.len()); + while end > 0 && !input.is_char_boundary(end) { + end -= 1; + } + (input[..end].to_string(), true) +} + impl Agent

{ + fn should_stream_shell_output(&self, tc: &ToolCall) -> bool { + self.event_sink.is_some() + && tc.name == "shell" + && matches!( + self.tool_rt.exec_target_kind, + crate::target::ExecTargetKind::Host + ) + } + + fn emit_shell_stream_action( + &mut self, + run_id: &str, + step: u32, + coalescer: &ShellStreamCoalescer, + action: ShellStreamAction, + ) { + match action { + ShellStreamAction::Emit { stream, text } => { + self.emit_event( + run_id, + step, + EventKind::ShellOutputChunk, + serde_json::json!({ + "tool_call_id": coalescer.tool_call_id, + "stream": stream, + "chunk": text, + }), + ); + } + ShellStreamAction::BudgetReached => { + self.emit_event( + run_id, + step, + EventKind::ShellOutputChunk, + serde_json::json!({ + "tool_call_id": coalescer.tool_call_id, + "stream": "meta", + "chunk": "[live output truncated; see final result]", + }), + ); + } + } + } + + /// Run a host `shell` tool call while forwarding live output chunks as + /// `ShellOutputChunk` events. Returns `Err(())` if the tool exceeds `dur` + /// (matching the non-streaming timeout branch); dropping the tool future on + /// timeout triggers `kill_on_drop` on the child. The final `ToolRunOutcome` + /// is identical to the non-streaming path. + async fn run_tool_once_with_live_stream( + &mut self, + run_id: &str, + step: u32, + tc: &ToolCall, + dur: std::time::Duration, + ) -> Result { + let (chunk_tx, mut chunk_rx) = tokio::sync::mpsc::channel::( + SHELL_STREAM_CHANNEL_CAPACITY, + ); + // Own the tool inputs so the future does not borrow `self`, leaving + // `self` free for synchronous event emission on this same task. + let tool_rt = self.tool_rt.clone(); + let mcp = self.mcp_registry.clone(); + let tc_owned = tc.clone(); + let tool_fut = run_tool_once(&tool_rt, &tc_owned, mcp.as_ref(), Some(chunk_tx)); + tokio::pin!(tool_fut); + let deadline = tokio::time::sleep(dur); + tokio::pin!(deadline); + let mut coalescer = ShellStreamCoalescer::new(tc.id.clone()); + + loop { + tokio::select! { + biased; + res = &mut tool_fut => { + // Flush any chunks that arrived before completion. + let mut batch = Vec::new(); + while let Ok(c) = chunk_rx.try_recv() { + batch.push(c); + } + for action in coalescer.ingest(&batch) { + self.emit_shell_stream_action(run_id, step, &coalescer, action); + } + for action in coalescer.flush_pending() { + self.emit_shell_stream_action(run_id, step, &coalescer, action); + } + return Ok(res); + } + maybe = chunk_rx.recv() => { + if let Some(first) = maybe { + // Coalesce all immediately-available chunks into one batch + // to keep the UI responsive under noisy output. + let mut batch = vec![first]; + while let Ok(c) = chunk_rx.try_recv() { + batch.push(c); + } + for action in coalescer.ingest(&batch) { + self.emit_shell_stream_action(run_id, step, &coalescer, action); + } + } + } + _ = &mut deadline => { + return Err(()); + } + } + } + } + pub(super) async fn run_tool_with_timeout_and_emit_mcp_events( &mut self, run_id: &str, @@ -110,14 +395,21 @@ impl Agent

{ phase: &str, ) -> Message { let tool_exec_timeout_ms = self.effective_tool_exec_timeout_ms(); - let outcome = match tokio::time::timeout( - std::time::Duration::from_millis(tool_exec_timeout_ms), - run_tool_once(&self.tool_rt, tc, self.mcp_registry.as_ref()), - ) - .await - { + let dur = std::time::Duration::from_millis(tool_exec_timeout_ms); + let run_result = if self.should_stream_shell_output(tc) { + self.run_tool_once_with_live_stream(run_id, step, tc, dur) + .await + } else { + tokio::time::timeout( + dur, + run_tool_once(&self.tool_rt, tc, self.mcp_registry.as_ref(), None), + ) + .await + .map_err(|_| ()) + }; + let outcome = match run_result { Ok(outcome) => outcome, - Err(_) => { + Err(()) => { let reason = format!( "runtime tool execution timeout: '{}' exceeded {}ms", tc.name, tool_exec_timeout_ms @@ -1369,3 +1661,76 @@ impl Agent

{ AllowedToolResultDecision::Continue } } + +#[cfg(test)] +mod shell_stream_tests { + use super::{ShellStreamAction, ShellStreamCoalescer}; + use crate::target::{ShellOutputChunk, ShellStreamKind}; + + #[test] + fn coalescer_preserves_stream_identity() { + let mut coalescer = ShellStreamCoalescer::new("tc1".to_string()); + let actions = coalescer.ingest(&[ + ShellOutputChunk { + stream: ShellStreamKind::Stdout, + bytes: b"out".to_vec(), + }, + ShellOutputChunk { + stream: ShellStreamKind::Stderr, + bytes: b"err".to_vec(), + }, + ]); + + assert_eq!( + actions, + vec![ + ShellStreamAction::Emit { + stream: "stdout", + text: "out".to_string(), + }, + ShellStreamAction::Emit { + stream: "stderr", + text: "err".to_string(), + }, + ] + ); + } + + #[test] + fn coalescer_waits_for_complete_utf8_codepoint() { + let mut coalescer = ShellStreamCoalescer::new("tc1".to_string()); + let euro = "€".as_bytes(); + let first = coalescer.ingest(&[ShellOutputChunk { + stream: ShellStreamKind::Stdout, + bytes: euro[..2].to_vec(), + }]); + assert!(first.is_empty(), "incomplete codepoint should not emit"); + + let second = coalescer.ingest(&[ShellOutputChunk { + stream: ShellStreamKind::Stdout, + bytes: euro[2..].to_vec(), + }]); + assert_eq!( + second, + vec![ShellStreamAction::Emit { + stream: "stdout", + text: "€".to_string(), + }] + ); + } + + #[test] + fn coalescer_truncates_live_output_on_utf8_boundary() { + let mut coalescer = ShellStreamCoalescer::new("tc1".to_string()); + let actions = coalescer.ingest(&[ShellOutputChunk { + stream: ShellStreamKind::Stdout, + bytes: "€".repeat(30_000).into_bytes(), + }]); + let ShellStreamAction::Emit { text, .. } = &actions[0] else { + panic!("first action should emit text"); + }; + assert!(text.len() <= super::SHELL_STREAM_MAX_BYTES); + assert!(text.is_char_boundary(text.len())); + assert_eq!(actions.last(), Some(&ShellStreamAction::BudgetReached)); + } +} diff --git a/src/agent_tool_exec.rs b/src/agent_tool_exec.rs index 9d509f9..52b7ab3 100644 --- a/src/agent_tool_exec.rs +++ b/src/agent_tool_exec.rs @@ -53,6 +53,7 @@ pub(crate) async fn run_tool_once( tool_rt: &ToolRuntime, tc: &ToolCall, mcp_registry: Option<&std::sync::Arc>, + shell_stream: Option, ) -> ToolRunOutcome { if tc.name.starts_with("mcp.") { match mcp_registry { @@ -110,8 +111,14 @@ pub(crate) async fn run_tool_once( }, } } else { + // Route through the streaming variant only when a stream is attached + // (host shell calls); every other call takes the exact original path. + let message = match shell_stream { + Some(_) => crate::tools::execute_tool_streaming(tool_rt, tc, shell_stream).await, + None => execute_tool(tool_rt, tc).await, + }; ToolRunOutcome { - message: execute_tool(tool_rt, tc).await, + message, mcp_meta: None, } } diff --git a/src/events.rs b/src/events.rs index c622ae8..8870519 100644 --- a/src/events.rs +++ b/src/events.rs @@ -18,6 +18,7 @@ pub enum EventKind { ToolExecTarget, ToolExecStart, ToolExecEnd, + ShellOutputChunk, PostWriteVerifyStart, PostWriteVerifyEnd, ToolRetry, diff --git a/src/target.rs b/src/target.rs index 2dc8d27..cdad183 100644 --- a/src/target.rs +++ b/src/target.rs @@ -61,6 +61,36 @@ impl TargetResult { } } +/// Which standard stream a live output chunk came from. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ShellStreamKind { + Stdout, + Stderr, +} + +impl ShellStreamKind { + pub fn as_str(self) -> &'static str { + match self { + ShellStreamKind::Stdout => "stdout", + ShellStreamKind::Stderr => "stderr", + } + } +} + +/// An incremental slice of shell output produced while the command is still +/// running. Emitted for live progress display only; the final result envelope +/// is unaffected and remains the source of truth. +#[derive(Debug, Clone)] +pub struct ShellOutputChunk { + pub stream: ShellStreamKind, + pub bytes: Vec, +} + +/// Sender used to forward live shell output chunks to a consumer (e.g. the TUI +/// tail). Cloneable and bounded; producers use best-effort non-blocking sends +/// so noisy output cannot backpressure or grow memory without bound. +pub type ShellOutputTx = tokio::sync::mpsc::Sender; + #[derive(Debug, Clone)] pub struct ShellReq { pub workdir: PathBuf, @@ -73,6 +103,11 @@ pub struct ShellReq { /// unrepresentable. Timeout enforcement currently applies to the host /// execution target only; the docker target ignores this field (follow-up). pub timeout_ms: u64, + /// Optional sink for live output chunks while the command runs. `None` + /// disables streaming (unchanged behavior). Honored by the host target only; + /// the docker target ignores it (follow-up), and the final result envelope + /// is identical either way. + pub stream: Option, } #[derive(Debug, Clone)] @@ -149,7 +184,7 @@ impl ExecTarget for HostTarget { None => req.workdir.clone(), }; command.current_dir(cwd); - match spawn_and_wait_managed(command, req.timeout_ms, None).await { + match spawn_and_wait_managed(command, req.timeout_ms, None, req.stream.clone()).await { Ok(managed) => build_shell_target_result( ExecTargetKind::Host, None, @@ -1040,6 +1075,7 @@ async fn spawn_and_wait_managed( mut command: Command, timeout_ms: u64, stdin_bytes: Option<&[u8]>, + stream: Option, ) -> std::io::Result { use std::sync::{Arc, Mutex}; use tokio::io::AsyncReadExt; @@ -1067,7 +1103,16 @@ async fn spawn_and_wait_managed( // buffers (rather than `read_to_end` returning the bytes) let the timeout // path recover whatever was captured so far even when the drain task is // still blocked on a pipe held open by a surviving grandchild process. - fn drain(pipe: Option) -> (Arc>>, tokio::task::JoinHandle<()>) + // + // When `stream` is provided, each read is also forwarded as a live + // `ShellOutputChunk` (a best-effort, non-blocking send) so a consumer can + // display progress while the command runs. The forwarded bytes never affect + // the accumulated buffers or the final result envelope. + fn drain( + pipe: Option, + stream: Option, + kind: ShellStreamKind, + ) -> (Arc>>, tokio::task::JoinHandle<()>) where R: tokio::io::AsyncRead + Unpin + Send + 'static, { @@ -1083,6 +1128,14 @@ async fn spawn_and_wait_managed( if let Ok(mut guard) = sink.lock() { guard.extend_from_slice(&chunk[..n]); } + if let Some(tx) = &stream { + // Ignore send errors: a dropped receiver just + // means nobody is watching the live stream. + let _ = tx.try_send(ShellOutputChunk { + stream: kind, + bytes: chunk[..n].to_vec(), + }); + } } } } @@ -1091,8 +1144,9 @@ async fn spawn_and_wait_managed( (buf, handle) } - let (stdout_buf, stdout_task) = drain(child.stdout.take()); - let (stderr_buf, stderr_task) = drain(child.stderr.take()); + let (stdout_buf, stdout_task) = + drain(child.stdout.take(), stream.clone(), ShellStreamKind::Stdout); + let (stderr_buf, stderr_task) = drain(child.stderr.take(), stream, ShellStreamKind::Stderr); let stdout_abort = stdout_task.abort_handle(); let stderr_abort = stderr_task.abort_handle(); @@ -1265,7 +1319,10 @@ fn shell_escape(s: &str) -> String { mod tests { use std::path::PathBuf; - use super::{resolve_path_scoped, DockerTarget, ExecTargetKind, HostTarget, ReadReq, ShellReq}; + use super::{ + resolve_path_scoped, DockerTarget, ExecTargetKind, HostTarget, ReadReq, ShellReq, + ShellStreamKind, + }; use crate::target::ExecTarget; use clap::ValueEnum; @@ -1309,6 +1366,7 @@ mod tests { cwd: Some("../".to_string()), max_tool_output_bytes: 200_000, timeout_ms: 0, + stream: None, }) .await; assert!(!out.ok); @@ -1324,6 +1382,7 @@ mod tests { cwd: None, max_tool_output_bytes: 200_000, timeout_ms: 0, + stream: None, } } else { ShellReq { @@ -1333,6 +1392,7 @@ mod tests { cwd: None, max_tool_output_bytes: 200_000, timeout_ms: 0, + stream: None, } } } @@ -1346,6 +1406,7 @@ mod tests { cwd: None, max_tool_output_bytes: 200_000, timeout_ms, + stream: None, } } else { ShellReq { @@ -1355,6 +1416,37 @@ mod tests { cwd: None, max_tool_output_bytes: 200_000, timeout_ms, + stream: None, + } + } + } + + fn stream_shell_req() -> ShellReq { + if cfg!(windows) { + ShellReq { + workdir: PathBuf::from("."), + cmd: "cmd".to_string(), + args: vec![ + "/C".to_string(), + "echo stream-out & echo stream-err 1>&2".to_string(), + ], + cwd: None, + max_tool_output_bytes: 200_000, + timeout_ms: 5_000, + stream: None, + } + } else { + ShellReq { + workdir: PathBuf::from("."), + cmd: "sh".to_string(), + args: vec![ + "-c".to_string(), + "printf stream-out; printf stream-err >&2".to_string(), + ], + cwd: None, + max_tool_output_bytes: 200_000, + timeout_ms: 5_000, + stream: None, } } } @@ -1400,6 +1492,58 @@ mod tests { ); } + #[tokio::test] + async fn host_shell_streaming_preserves_final_result_envelope_fields() { + let plain = HostTarget.exec_shell(stream_shell_req()).await; + let (tx, mut rx) = tokio::sync::mpsc::channel(16); + let mut streamed_req = stream_shell_req(); + streamed_req.stream = Some(tx); + let streamed = HostTarget.exec_shell(streamed_req).await; + + assert!(plain.ok, "plain command should succeed: {}", plain.content); + assert!( + streamed.ok, + "streamed command should succeed: {}", + streamed.content + ); + let plain_json: serde_json::Value = + serde_json::from_str(&plain.content).expect("plain shell result JSON"); + let streamed_json: serde_json::Value = + serde_json::from_str(&streamed.content).expect("streamed shell result JSON"); + for key in [ + "status", + "stdout", + "stderr", + "stdout_truncated", + "stderr_truncated", + "max_tool_output_bytes", + ] { + assert_eq!(streamed_json[key], plain_json[key], "field {key}"); + } + assert_eq!(streamed.stdout_truncated, plain.stdout_truncated); + assert_eq!(streamed.stderr_truncated, plain.stderr_truncated); + assert_eq!(streamed.exit_code, plain.exit_code); + + let mut live_stdout = Vec::new(); + let mut live_stderr = Vec::new(); + while let Ok(chunk) = rx.try_recv() { + match chunk.stream { + ShellStreamKind::Stdout => live_stdout.extend_from_slice(&chunk.bytes), + ShellStreamKind::Stderr => live_stderr.extend_from_slice(&chunk.bytes), + } + } + assert!( + String::from_utf8_lossy(&live_stdout).contains("stream-out"), + "stdout chunks: {:?}", + String::from_utf8_lossy(&live_stdout) + ); + assert!( + String::from_utf8_lossy(&live_stderr).contains("stream-err"), + "stderr chunks: {:?}", + String::from_utf8_lossy(&live_stderr) + ); + } + #[test] fn docker_command_assembly_is_deterministic() { let t = DockerTarget::new( @@ -1451,6 +1595,7 @@ mod tests { cwd: None, max_tool_output_bytes: 200_000, timeout_ms: 500, + stream: None, }) .await; assert!(!out.ok, "timeout on docker target must be rejected"); diff --git a/src/tools.rs b/src/tools.rs index 007694d..72cf668 100644 --- a/src/tools.rs +++ b/src/tools.rs @@ -168,6 +168,18 @@ pub struct ToolErrorDetail { } pub async fn execute_tool(rt: &ToolRuntime, tc: &ToolCall) -> Message { + execute_tool_streaming(rt, tc, None).await +} + +/// Like [`execute_tool`], but forwards live shell output chunks to `shell_stream` +/// while a `shell` command runs. Only the `shell` tool consults the stream; all +/// other tools behave identically to [`execute_tool`]. The final result envelope +/// is unaffected by streaming. +pub async fn execute_tool_streaming( + rt: &ToolRuntime, + tc: &ToolCall, + shell_stream: Option, +) -> Message { let normalized_args = catalog::normalize_builtin_tool_args(&tc.name, &tc.arguments); let side_effects = tool_side_effects(&tc.name); if let Err(e) = validate_builtin_tool_args(&tc.name, &normalized_args, rt.tool_args_strict) { @@ -186,7 +198,7 @@ pub async fn execute_tool(rt: &ToolRuntime, tc: &ToolCall) -> Message { "read_file" => exec_fs::run_read_file(rt, &normalized_args).await, "glob" => exec_fs::run_glob(rt, &normalized_args).await, "grep" => exec_fs::run_grep(rt, &normalized_args).await, - "shell" => exec_shell::run_shell(rt, &normalized_args).await, + "shell" => exec_shell::run_shell(rt, &normalized_args, shell_stream).await, "write_file" => exec_write::run_write_file(rt, &normalized_args).await, "apply_patch" => exec_write::run_apply_patch(rt, &normalized_args).await, "edit" => exec_write::run_edit(rt, &normalized_args).await, diff --git a/src/tools/exec_shell.rs b/src/tools/exec_shell.rs index 6dc1a54..d7dcaa9 100644 --- a/src/tools/exec_shell.rs +++ b/src/tools/exec_shell.rs @@ -1,6 +1,6 @@ use serde_json::Value; -use crate::target::{ExecTargetKind, ShellReq, TargetResult}; +use crate::target::{ExecTargetKind, ShellOutputTx, ShellReq, TargetResult}; use crate::types::SideEffects; use super::exec_support::{failed_exec, ToolExecution}; @@ -31,7 +31,16 @@ pub(super) fn resolve_shell_timeout_ms(args: &Value, target: ExecTargetKind) -> } } -pub(super) async fn run_shell(rt: &ToolRuntime, args: &Value) -> ToolExecution { +pub(super) async fn run_shell( + rt: &ToolRuntime, + args: &Value, + stream: Option, +) -> ToolExecution { + // Live streaming is host-only; the docker target ignores ShellReq.stream. + let stream = match rt.exec_target_kind { + ExecTargetKind::Host => stream, + ExecTargetKind::Docker => None, + }; let shell_allowed = rt.allow_shell || (rt.allow_shell_in_workdir_only && shell_cwd_is_workdir_scoped(args)); if !shell_allowed && !rt.unsafe_bypass_allow_flags { @@ -76,6 +85,7 @@ pub(super) async fn run_shell(rt: &ToolRuntime, args: &Value) -> ToolExecution { cwd: cwd.clone(), max_tool_output_bytes: rt.max_tool_output_bytes, timeout_ms, + stream: stream.clone(), }; let mut out = rt.exec_target.exec_shell(req).await; if !out.ok && shell_spawn_not_found(&out.content) { @@ -91,6 +101,7 @@ pub(super) async fn run_shell(rt: &ToolRuntime, args: &Value) -> ToolExecution { cwd, max_tool_output_bytes: rt.max_tool_output_bytes, timeout_ms, + stream, }) .await; out = annotate_shell_repair(repaired, repair_strategy); diff --git a/src/tui/state.rs b/src/tui/state.rs index 76b678c..071e634 100644 --- a/src/tui/state.rs +++ b/src/tui/state.rs @@ -137,6 +137,7 @@ impl UiState { EventKind::ToolDecision => self.apply_tool_decision_event(ev), EventKind::ToolExecStart => self.apply_tool_exec_start_event(ev), EventKind::ToolExecEnd => self.apply_tool_exec_end_event(ev), + EventKind::ShellOutputChunk => self.apply_shell_output_chunk_event(ev), EventKind::PostWriteVerifyStart => self.apply_post_write_verify_start_event(ev), EventKind::PostWriteVerifyEnd => self.apply_post_write_verify_end_event(ev), EventKind::PolicyLoaded => self.apply_policy_loaded_event(ev), diff --git a/src/tui/state/events.rs b/src/tui/state/events.rs index 98eb5da..da91d69 100644 --- a/src/tui/state/events.rs +++ b/src/tui/state/events.rs @@ -699,4 +699,35 @@ impl UiState { self.push_log("compaction performed".to_string()); } } + + /// Render a live shell output chunk into the log/tail area while the command + /// is still running. stderr is distinguished with an `err>` prefix; stdout + /// with `out>`. Each incoming line is pushed separately so the ring-buffered + /// log bounds growth automatically. + pub(super) fn apply_shell_output_chunk_event(&mut self, ev: &Event) { + let stream = ev + .data + .get("stream") + .and_then(|v| v.as_str()) + .unwrap_or("out"); + let chunk = ev + .data + .get("chunk") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + if chunk.is_empty() { + return; + } + let prefix = match stream { + "stderr" => "err>", + "meta" => "···", + _ => "out>", + }; + for line in chunk.split('\n') { + if line.is_empty() { + continue; + } + self.push_log(format!("{prefix} {line}")); + } + } } diff --git a/src/tui/state/tests.rs b/src/tui/state/tests.rs index f7e7fe4..c69ab07 100644 --- a/src/tui/state/tests.rs +++ b/src/tui/state/tests.rs @@ -168,6 +168,50 @@ fn logs_are_capped() { assert_eq!(s.logs, vec!["b".to_string(), "c".to_string()]); } +#[test] +fn shell_output_chunk_logs_stream_identity_and_caps() { + let mut s = UiState::new(3); + s.apply_event(&Event::new( + "r1".to_string(), + 1, + EventKind::ShellOutputChunk, + serde_json::json!({ + "tool_call_id":"tc1", + "stream":"stdout", + "chunk":"out-a\nout-b" + }), + )); + s.apply_event(&Event::new( + "r1".to_string(), + 1, + EventKind::ShellOutputChunk, + serde_json::json!({ + "tool_call_id":"tc1", + "stream":"stderr", + "chunk":"err-a" + }), + )); + s.apply_event(&Event::new( + "r1".to_string(), + 1, + EventKind::ShellOutputChunk, + serde_json::json!({ + "tool_call_id":"tc1", + "stream":"meta", + "chunk":"[live output truncated; see final result]" + }), + )); + + assert_eq!( + s.logs, + vec![ + "out> out-b".to_string(), + "err> err-a".to_string(), + "··· [live output truncated; see final result]".to_string(), + ] + ); +} + #[test] fn approvals_refresh_and_transition() { let tmp = tempdir().expect("tmp");