From a414a35d2d118d861a43c58e03b8794224aba6c2 Mon Sep 17 00:00:00 2001 From: w3irdrobot Date: Thu, 29 Oct 2026 21:57:21 +0000 Subject: [PATCH 1/3] fix(agent): bound shell execution and show quiet command progress --- apps/maple-agent/README.md | 11 + apps/maple-agent/app/src/ui/chat/cache.rs | 3 + apps/maple-agent/app/src/ui/chat/tests.rs | 21 +- .../maple-agent/app/src/ui/chat/transcript.rs | 42 +++- apps/maple-agent/crates/UPSTREAM.md | 1 + .../crates/maple-agent/src/agent/runs.rs | 1 + .../maple-agent/src/agent/shell_tests.rs | 123 +++++++++++ .../crates/maple-agent/src/agent/tests.rs | 3 + .../crates/maple-agent/src/agent/timeline.rs | 18 ++ .../crates/maple-agent/src/agent/tools/mod.rs | 4 + .../maple-agent/src/agent/tools/tests.rs | 2 + .../crates/pi-coding-agent/src/tools/bash.rs | 166 +++++++++++--- .../pi-coding-agent/src/tools/bash/tests.rs | 202 ++++++++++++++++++ .../crates/pi-coding-agent/src/tools/tests.rs | 2 +- 14 files changed, 561 insertions(+), 38 deletions(-) create mode 100644 apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs diff --git a/apps/maple-agent/README.md b/apps/maple-agent/README.md index 9ea121c5a..73fa85561 100644 --- a/apps/maple-agent/README.md +++ b/apps/maple-agent/README.md @@ -70,6 +70,17 @@ agent runtime; its desktop Agent page points users to Maple Work. attachments (picker, paste, or drag and drop), a per-message Copy button, and a context-window indicator. Every tool call runs without asking for approval. +- Shell tool calls start in the task's saved folder on the execution computer. + An absolute `workdir` selects another checkout/worktree for one call; a `cd` + inside a command does not affect later calls. Maple supplies a 120-second + default deadline, and the call can request a longer `timeout` explicitly. + Quiet commands report their starting folder, elapsed time, time since output and + deadline every five seconds, including on collapsed tool cards. Stop and + deadlines kill the shell's process group on Unix (the process tree on Windows). + These are execution safeguards, not a filesystem sandbox: repository searches + must still choose the correct folder and bounded scope, and deliberately + detached processes can escape a Unix process group. `head` bounds output, + not runtime. - Slash commands in the composer: `/btw` asks a side question the task never sees, plus `/compact`, `/new`, `/pin`, `/web`, `/model`, and `/help`. Skills and prompt templates appear in the same list. diff --git a/apps/maple-agent/app/src/ui/chat/cache.rs b/apps/maple-agent/app/src/ui/chat/cache.rs index 21467f469..97eceaf67 100644 --- a/apps/maple-agent/app/src/ui/chat/cache.rs +++ b/apps/maple-agent/app/src/ui/chat/cache.rs @@ -264,6 +264,8 @@ pub(super) struct ItemDerived { pub(super) attachments: Vec, /// Readable tool output for the expanded card. pub(super) output_text: Option, + /// Shell execution status stays visible even on a collapsed card. + pub(super) shell_status: Option, /// `input: {json}` for the expanded card. pub(super) input_line: Option, /// +/- lines of an edit or write tool, capped at `MAX_DIFF_LINES`. @@ -290,6 +292,7 @@ impl ItemDerived { }) .collect(), output_text, + shell_status: super::transcript::shell_execution_status(item).map(SharedString::from), input_line: tool_input_line(item).map(SharedString::from), diff_lines: Arc::new(diff_lines_for(item)), external_agent: external_agent_activity(item).map(Arc::new), diff --git a/apps/maple-agent/app/src/ui/chat/tests.rs b/apps/maple-agent/app/src/ui/chat/tests.rs index cbfb7272d..863fce822 100644 --- a/apps/maple-agent/app/src/ui/chat/tests.rs +++ b/apps/maple-agent/app/src/ui/chat/tests.rs @@ -10688,8 +10688,14 @@ mod state_tests { chat.booting = false; chat.replace_timeline(vec![ AgentTimelineItem { - status: Some("completed".to_string()), + status: Some("running".to_string()), title: Some("Terminal: cargo test".to_string()), + output: Some(serde_json::json!({"structuredContent": {"shellExecution": { + "cwd": "/fixture/worktree", + "elapsedSeconds": 65, + "outputIdleSeconds": 60, + "timeoutSeconds": 120, + }}})), ..item("tool-1", "tool", None) }, AgentTimelineItem { @@ -10711,6 +10717,19 @@ mod state_tests { cx.simulate_resize(gpui::size(px(1200.), px(800.))); let card = cx.debug_bounds("tool-card").expect("the tool card renders"); + assert!( + cx.debug_bounds("shell-execution-status").is_some(), + "shell status remains visible on a collapsed card" + ); + chat.read_with(cx, |chat, _| { + let status = + super::super::transcript::shell_execution_status(&chat.timeline[0]).unwrap(); + assert!(status.contains("/fixture/worktree")); + assert!(status.contains("Running 65s")); + assert!(status.contains("no output for 60s")); + assert!(status.contains("deadline 120s")); + assert!(status.contains("Stop cancels command")); + }); let tool_title = cx .debug_bounds("tool-title") .expect("the tool card title renders"); diff --git a/apps/maple-agent/app/src/ui/chat/transcript.rs b/apps/maple-agent/app/src/ui/chat/transcript.rs index 392313a77..02d9c54ec 100644 --- a/apps/maple-agent/app/src/ui/chat/transcript.rs +++ b/apps/maple-agent/app/src/ui/chat/transcript.rs @@ -1130,7 +1130,7 @@ fn render_tool( let raw = item.title.as_deref().unwrap_or(&item.item_type); SharedString::from(tool_label_title(raw).to_string()) }); - let card = div() + let mut card = div() .id(gpui::SharedString::from(format!("tool-toggle-{item_id}"))) .debug_selector(|| "tool-card".to_string()) .role(gpui::Role::DisclosureTriangle) @@ -1201,8 +1201,20 @@ fn render_tool( theme::text_muted(), )), ); - // Collapsed: the header alone; the summary takes over the title when - // it lands. Only expanded cards pay for the derived payload strings. + if running + && shell_execution(item).is_some() + && let Some(status) = &transcript.derived.get(item, revision).shell_status + { + card = card.child( + div() + .text_xs() + .text_color(gpui::rgb(theme::text_secondary())) + .debug_selector(|| "shell-execution-status".to_string()) + .child(status.clone()), + ); + } + // Collapsed: the header and live shell status; the summary takes over the title when + // it lands. Expanded cards also show the cached arguments and output. if !details { return div().child(card); } @@ -1390,6 +1402,30 @@ pub(super) fn tool_output_markdown(item: &AgentTimelineItem) -> Option { (!trimmed.is_empty()).then(|| trimmed.to_string()) } +fn shell_execution(item: &AgentTimelineItem) -> Option<&serde_json::Value> { + item.output + .as_ref()? + .get("structuredContent")? + .get("shellExecution") +} + +pub(super) fn shell_execution_status(item: &AgentTimelineItem) -> Option { + let execution = shell_execution(item)?; + let cwd = execution.get("cwd")?.as_str()?; + let elapsed = execution.get("elapsedSeconds")?.as_u64()?; + let idle = execution.get("outputIdleSeconds")?.as_u64()?; + let deadline = execution + .get("timeoutSeconds") + .and_then(serde_json::Value::as_f64) + .map_or_else( + || "no deadline".to_string(), + |seconds| format!("deadline {seconds}s"), + ); + Some(format!( + "Started in {cwd}\nRunning {elapsed}s · no output for {idle}s · {deadline} · Stop cancels command" + )) +} + fn extract_output_text(value: &serde_json::Value) -> Option { match value { serde_json::Value::String(text) => Some(text.clone()), diff --git a/apps/maple-agent/crates/UPSTREAM.md b/apps/maple-agent/crates/UPSTREAM.md index a6a776236..a425bfeaf 100644 --- a/apps/maple-agent/crates/UPSTREAM.md +++ b/apps/maple-agent/crates/UPSTREAM.md @@ -19,6 +19,7 @@ complete audit of every earlier difference. | `SessionBase`, `SessionManager::open_partial` and storage context hooks | Resume a long conversation from the required compaction context without loading all saved history. Original entry IDs and parents remain intact. SQL, caches and paging stay in Maple. Complete-load behavior remains available; operations requiring an unloaded prefix must require a complete reload. | Full-versus-partial projection comparisons, inherited model/thinking/system state, compaction, retry and fork/export restrictions in `pi-coding-agent/src/session.rs`. | | Explicit `flush` and `has_unpersisted` | Expose pending persistence so an embedding host can retry. Maple decides whether shutdown or eviction may finish. Consuming an error must not discard the pending entries. | Persistence failure/retry tests in `pi-coding-agent/src/session.rs` and Maple store tests. | | Optional tool-output directory | Let an embedding host choose the lifetime and location of overflow files. Standalone defaults remain available. An invalid explicit directory fails without a fallback. Maple owns task paths, account policy and cleanup. | Shell, user-bash and private-directory tests in `pi-coding-agent`. | +| Optional shell default deadline, absolute per-call `workdir`, and quiet status updates | An embedding host needs to bound silent commands and make their execution folder and deadline visible. Pi retains no default deadline; Maple selects 120 seconds. Workdir is explicit per call and never changes the session folder. Status travels through existing partial-result events, with no prompt/output logging. | Disposable-folder and blocked FIFO pipeline tests in `tools/bash/tests.rs`; saved-task cold-load and quiet-progress/Stop tests in Maple's `agent/shell_tests.rs`. | These are deliberate embedding adaptations. They do not authorize changing the agent loop, compaction decisions or the meaning and ordering of Pi events. diff --git a/apps/maple-agent/crates/maple-agent/src/agent/runs.rs b/apps/maple-agent/crates/maple-agent/src/agent/runs.rs index 1f3a65f27..f18feec9d 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/runs.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/runs.rs @@ -2030,6 +2030,7 @@ impl RunWatch { } else { let tool = match event { pi_agent_core::AgentEvent::ToolExecutionStart { tool_call_id, .. } + | pi_agent_core::AgentEvent::ToolExecutionUpdate { tool_call_id, .. } | pi_agent_core::AgentEvent::ToolExecutionEnd { tool_call_id, .. } => { Some(tool_call_id.as_str()) } diff --git a/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs b/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs new file mode 100644 index 000000000..6764339a2 --- /dev/null +++ b/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs @@ -0,0 +1,123 @@ +//! Execution-host folder selection and quiet shell progress, without remote +//! accounts, live services or the real application's store. +#![cfg(unix)] + +use super::*; + +#[tokio::test] +async fn saved_task_worktree_survives_a_different_runtime_default_and_cold_load() { + let mut harness = Harness::new().await; + let worktree = tempfile::tempdir().unwrap(); + std::fs::write(worktree.path().join("window.rs"), "worktree-fixture").unwrap(); + let task = harness + .handle + .create_session(Some(AgentCreateSessionRequest { + project_root: Some(worktree.path().to_string_lossy().into_owned()), + title: None, + model: None, + context_limit: None, + mcp_server_names: Some(Vec::new()), + })) + .await + .unwrap() + .session + .id; + for cold in [false, true] { + if cold { + harness.restart().await; + harness.start().await; + } + harness.faux.push_message(vec![faux_tool_call( + "bash", + json!({"command": "pwd; cat ./window.rs"}), + )]); + harness.faux.push_text("Done."); + let mut run = harness.send(&task, "Read the fixture").await; + assert_eq!(finished(&mut run).await, AgentRunTerminal::Completed); + let rows = harness.recorder.live_rows(&run.run_id); + let row = rows + .iter() + .rev() + .find(|row| row.item_type == "tool") + .unwrap(); + let output = row.output.as_ref().unwrap()["text"].as_str().unwrap(); + let mut lines = output.lines(); + assert_eq!( + std::fs::canonicalize(lines.next().unwrap()).unwrap(), + worktree.path().canonicalize().unwrap() + ); + assert_eq!(lines.next(), Some("worktree-fixture")); + let detail = harness.handle.load_session(task.clone()).await.unwrap(); + assert_eq!( + Path::new(&detail.session.project_root) + .canonicalize() + .unwrap(), + worktree.path().canonicalize().unwrap() + ); + } + harness.handle.stop().await.unwrap(); +} + +#[tokio::test] +async fn quiet_shell_reports_effective_folder_deadline_and_idle_time_until_stop() { + let harness = Harness::new().await; + let worktree = tempfile::tempdir().unwrap(); + harness.faux.push_message(vec![faux_tool_call( + "bash", + json!({ + "command": "sleep 30 # fixture-private-command", + "workdir": worktree.path(), + }), + )]); + let task = harness.create_task().await; + let mut run = harness.send(&task, "Wait for cancellation").await; + let run_id = run.run_id.clone(); + eventually(|| { + harness.recorder.run_events(&run_id).iter().any(|event| { + let AgentRunEvent::TimelineItem(row) = event else { + return false; + }; + let Some(progress) = row + .output + .as_ref() + .and_then(|output| output.get("structuredContent")) + .and_then(|details| details.get("shellExecution")) + else { + return false; + }; + progress["elapsedSeconds"] + .as_u64() + .is_some_and(|seconds| seconds >= 5) + }) + }) + .await; + let rows = harness.recorder.live_rows(&run_id); + let tool = rows.iter().find(|row| row.item_type == "tool").unwrap(); + assert_eq!(tool.status.as_deref(), Some("running")); + let progress = &tool.output.as_ref().unwrap()["structuredContent"]["shellExecution"]; + assert_eq!(progress["cwd"], json!(worktree.path())); + assert_eq!(progress["timeoutSeconds"], 120.0); + assert!(progress["outputIdleSeconds"].as_u64().unwrap() >= 5); + assert!(!progress.to_string().contains("fixture-private-command")); + let identity = tool.live.clone(); + assert!(identity.is_some()); + harness + .handle + .cancel_desktop_run(run_id.clone()) + .await + .unwrap(); + assert_eq!(finished(&mut run).await, AgentRunTerminal::Cancelled); + // Partial results replace the original tool occurrence and never settle it + // early or create an extra row. The final cancellation removes live status. + let rows = harness.recorder.live_rows(&run_id); + let tools: Vec<_> = rows.iter().filter(|row| row.item_type == "tool").collect(); + assert_eq!(tools.len(), 1); + assert_eq!(tools[0].live, identity); + assert_ne!(tools[0].status.as_deref(), Some("running")); + assert!( + tools[0].output.as_ref().unwrap()["structuredContent"] + .get("shellExecution") + .is_none() + ); + harness.handle.stop().await.unwrap(); +} diff --git a/apps/maple-agent/crates/maple-agent/src/agent/tests.rs b/apps/maple-agent/crates/maple-agent/src/agent/tests.rs index 344667d07..c4dddd751 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/tests.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/tests.rs @@ -16,6 +16,9 @@ use crate::maple_api::test_maple_api_session; #[path = "image_persistence_tests.rs"] mod image_persistence_tests; +#[path = "shell_tests.rs"] +mod shell_tests; + const USER: &str = "slice@example.com"; #[derive(Default)] diff --git a/apps/maple-agent/crates/maple-agent/src/agent/timeline.rs b/apps/maple-agent/crates/maple-agent/src/agent/timeline.rs index de6aa273b..b9a98dbee 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/timeline.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/timeline.rs @@ -608,6 +608,24 @@ impl LiveTimeline { merge: "replace".to_string(), }] } + AgentEvent::ToolExecutionUpdate { + tool_call_id, + partial_result, + .. + } => { + let Some(id) = self.tool_rows.get(tool_call_id) else { + return Vec::new(); + }; + let mut item = tool_result_item( + id.clone(), + &partial_result.content, + partial_result.details.as_ref(), + false, + pi_ai::now_ms(), + ); + item.status = Some("running".into()); + vec![item] + } AgentEvent::ToolExecutionEnd { tool_call_id, result, diff --git a/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs b/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs index 445fbf239..fd0f8d6b4 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs @@ -33,6 +33,8 @@ use crate::maple_api::MapleWebTransport; /// The developer's choice of bash. pub(crate) const SHELL_ENV: &str = "MAPLE_SHELL"; +/// Silent searches must not hold a task forever. Builds can request longer explicitly. +const SHELL_DEFAULT_TIMEOUT: f64 = 120.0; /// How a task's tools are set up. pub(crate) struct TaskTools { @@ -67,9 +69,11 @@ pub(crate) fn task_tools(task: TaskToolsFor) -> TaskTools { .map(PathBuf::from); let hook = spawn_hook(task.session_id.clone(), task.login_path); let mut options = ToolsOptions::default(); + options.bash.default_timeout = Some(SHELL_DEFAULT_TIMEOUT); options.bash.shell_path = shell_path.clone(); options.bash.spawn_hook = Some(hook.clone()); options.powershell = PowerShellToolOptions { + default_timeout: Some(SHELL_DEFAULT_TIMEOUT), spawn_hook: Some(hook), ..PowerShellToolOptions::default() }; diff --git a/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs b/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs index 8f0a29a96..72866b3db 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs @@ -128,6 +128,8 @@ fn the_model_gets_read_a_shell_edit_and_write() { }; assert_eq!(tools.builtin, ["read", shell, "edit", "write"]); assert!(tools.options.powershell.spawn_hook.is_some()); + assert_eq!(tools.options.bash.default_timeout, Some(120.0)); + assert_eq!(tools.options.powershell.default_timeout, Some(120.0)); } fn tool(tools: &TaskTools, name: &str) -> Arc { diff --git a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs index 140357f1a..0c7b23dce 100644 --- a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs +++ b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs @@ -2,15 +2,15 @@ //! //! Output streams to the interface as it arrives. The model gets the last 2000 lines or //! 50KB, whichever is less, and when that cuts anything the full output is saved to a -//! file it can read. There is no default timeout; Stop or a timeout ends the command -//! and everything it started. +//! file it can read. The embedding host can choose a default timeout. Stop or a +//! timeout ends the command's process group (its process tree on Windows). use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::process::Stdio; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::time::{Duration, Instant}; use async_trait::async_trait; use pi_agent_core::{AgentTool, AgentToolResult, ToolError, ToolInvocation, ToolUpdates}; @@ -33,6 +33,8 @@ use super::{PREFER_STRICT, ToolContext}; const MAX_TIMEOUT_SECONDS: f64 = 2_147_483_647.0 / 1000.0; /// How often output updates reach the interface at most. const UPDATE_THROTTLE: Duration = Duration::from_millis(100); +/// Quiet commands still report their execution status to the interface. +const STATUS_INTERVAL: Duration = Duration::from_secs(5); /// After the shell exits, how long its output may stay quiet before collection stops. /// A background process can keep the output open; output still arriving keeps it open. const EXIT_STDIO_GRACE: Duration = Duration::from_millis(100); @@ -82,6 +84,8 @@ pub trait BashOperations: Send + Sync { /// Run `command` in `cwd`, streaming its output. The exit code is `None` when the /// command ended without one; a command killed by a signal reports 128 plus the /// signal's number. + /// A selected working directory that is unavailable must fail, never fall back + /// to the host's working directory or a temporary folder. async fn exec( &self, command: &str, @@ -97,7 +101,7 @@ enum LocalShell { } /// Commands on this machine, in their own process group (a process tree on Windows) -/// so that stopping one reaches everything it started. +/// so stopping one reaches the group's members, including pipeline processes. #[derive(Clone, Debug)] pub struct LocalShellOperations { shell: LocalShell, @@ -142,7 +146,15 @@ impl BashOperations for LocalShellOperations { format!("{POWERSHELL_UTF8_PREFIX}{command}"), ), }; - if tokio::fs::metadata(cwd).await.is_err() { + if !cwd.is_absolute() { + return Err(ExecError::Failed( + "Working directory must be an absolute path".into(), + )); + } + if !tokio::fs::metadata(cwd) + .await + .is_ok_and(|metadata| metadata.is_dir()) + { return Err(ExecError::Failed(format!( "Working directory does not exist: {}\nCannot execute {shell_name} commands.", cwd.display() @@ -240,10 +252,7 @@ async fn run_local( if let Some(stderr) = child.stderr.take() { readers.spawn(pump(stderr, options.on_data.clone(), activity.clone())); } - if let Some(mut stdin) = child.stdin.take() { - let _ = stdin.write_all(command.as_bytes()).await; - drop(stdin); - } + let stdin = child.stdin.take(); let kill = |pid: Option| { if let Some(pid) = pid { @@ -259,7 +268,14 @@ async fn run_local( } }; tokio::pin!(deadline); - let wait = child.wait(); + // Writing a large command to a stdin-based shell can block too. Keep that + // write inside the same cancellation/deadline boundary as waiting for exit. + let wait = async { + if let Some(mut stdin) = stdin { + let _ = stdin.write_all(command.as_bytes()).await; + } + child.wait().await + }; tokio::pin!(wait); let status = loop { tokio::select! { @@ -362,6 +378,9 @@ async fn pump(mut stream: R, on_data: OnData, activity: Ar /// Options for `bash`. #[derive(Clone)] pub struct BashToolOptions { + /// Seconds before a command without an explicit timeout is stopped. + /// Standalone Pi has no default; an embedding host can choose one. + pub default_timeout: Option, /// Full output goes here when present, without falling back on a failed path. pub output_dir: Option, /// Where commands run; this machine when `None`. @@ -379,6 +398,7 @@ pub struct BashToolOptions { impl Default for BashToolOptions { fn default() -> Self { Self { + default_timeout: None, output_dir: None, operations: None, command_prefix: None, @@ -392,6 +412,7 @@ impl Default for BashToolOptions { /// Options for `powershell`. #[derive(Clone)] pub struct PowerShellToolOptions { + pub default_timeout: Option, pub output_dir: Option, pub operations: Option>, pub expose_session_environment: bool, @@ -401,6 +422,7 @@ pub struct PowerShellToolOptions { impl Default for PowerShellToolOptions { fn default() -> Self { Self { + default_timeout: None, output_dir: None, operations: None, expose_session_environment: true, @@ -419,6 +441,7 @@ pub struct ShellTool { command_prefix: Option, expose_session_environment: bool, spawn_hook: Option, + default_timeout: Option, context: ToolContext, } @@ -426,23 +449,32 @@ pub struct ShellTool { struct ShellParams { command: String, #[serde(default)] + workdir: Option, + #[serde(default)] timeout: Option, } -fn shell_declaration(name: &str, shell_name: &str) -> Tool { +fn shell_declaration(name: &str, shell_name: &str, default_timeout: Option) -> Tool { + let timeout_description = match default_timeout { + Some(seconds) => format!( + "Timeout in seconds (optional, defaults to {seconds} seconds). Set explicitly for longer commands." + ), + None => "Timeout in seconds (optional, no default timeout)".to_string(), + }; Tool::new( name, format!( - "Execute a {shell_name} command in the current working directory. Returns stdout and stderr. Output is truncated to last {DEFAULT_MAX_LINES} lines or {}KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.", + "Execute a {shell_name} command in the session working directory, or an explicit absolute workdir. Each call starts in that folder; cd does not change later calls. Returns stdout and stderr. Output is truncated to last {DEFAULT_MAX_LINES} lines or {}KB (whichever is hit first). If truncated, full output is saved to a temp file. {timeout_description}", DEFAULT_MAX_BYTES / 1024 ), json!({ "type": "object", "properties": { "command": { "type": "string", "description": "Shell command to execute" }, + "workdir": { "type": "string", "description": "Absolute working directory on the execution computer. For repository searches, set this to the intended checkout or worktree." }, "timeout": { "type": "number", - "description": "Timeout in seconds (optional, no default timeout)" + "description": timeout_description } }, "required": ["command"] @@ -457,7 +489,7 @@ impl ShellTool { .operations .unwrap_or_else(|| Arc::new(LocalShellOperations::bash(options.shell_path))); Self { - declaration: shell_declaration("bash", "bash"), + declaration: shell_declaration("bash", "bash", options.default_timeout), name: "bash", cwd: cwd.into(), output_dir: options.output_dir, @@ -465,6 +497,7 @@ impl ShellTool { command_prefix: options.command_prefix, expose_session_environment: options.expose_session_environment, spawn_hook: options.spawn_hook, + default_timeout: options.default_timeout, context, } } @@ -478,7 +511,7 @@ impl ShellTool { .operations .unwrap_or_else(|| Arc::new(LocalShellOperations::powershell())); Self { - declaration: shell_declaration("powershell", "PowerShell"), + declaration: shell_declaration("powershell", "PowerShell", options.default_timeout), name: "powershell", cwd: cwd.into(), output_dir: options.output_dir, @@ -486,20 +519,23 @@ impl ShellTool { command_prefix: None, expose_session_environment: options.expose_session_environment, spawn_hook: options.spawn_hook, + default_timeout: options.default_timeout, context, } } /// The guideline that goes with the tool's snippet. pub fn guidelines(&self) -> Vec { + let mut rules = vec![ + "For repository searches, set workdir to the absolute intended checkout/worktree and search within it. Verify the folder before using parent paths; never guess ../../ from a temporary folder. Avoid recursive searches of filesystem roots or virtual filesystems such as /proc, /sys and /dev. head limits output, not runtime; use a timeout for searches.".to_string(), + ]; if self.expose_session_environment { - vec![format!( + rules.push(format!( "You can inspect {}_* environment variables for current model and session details.", self.context.env_prefix() - )] - } else { - Vec::new() + )); } + rules } fn spawn_context(&self, command: String, cwd: PathBuf) -> BashSpawnContext { @@ -545,23 +581,50 @@ struct UpdateThrottle { task: tokio::task::JoinHandle<()>, output: Arc>, updates: ToolUpdates, + progress: Arc, +} + +struct ShellProgress { + cwd: PathBuf, + timeout: Option, + started: Instant, + last_output: Mutex, } impl UpdateThrottle { - fn start(output: Arc>, updates: ToolUpdates) -> Self { - updates.send(AgentToolResult { - content: Vec::new(), - ..AgentToolResult::default() + fn start( + output: Arc>, + updates: ToolUpdates, + cwd: PathBuf, + timeout: Option, + ) -> Self { + let progress = Arc::new(ShellProgress { + cwd, + timeout, + started: Instant::now(), + last_output: Mutex::new(Instant::now()), }); - let dirty = Arc::new(AtomicBool::new(false)); + let dirty = Arc::new(AtomicBool::new(true)); + send_update(&dirty, &output, &updates, &progress); let wake = Arc::new(Notify::new()); let task = tokio::spawn({ - let (dirty, wake, output, updates) = - (dirty.clone(), wake.clone(), output.clone(), updates.clone()); + let (dirty, wake, output, updates, progress) = ( + dirty.clone(), + wake.clone(), + output.clone(), + updates.clone(), + progress.clone(), + ); async move { + let mut status = tokio::time::interval(STATUS_INTERVAL); + status.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + status.tick().await; loop { - wake.notified().await; - send_update(&dirty, &output, &updates); + tokio::select! { + _ = wake.notified() => {}, + _ = status.tick() => dirty.store(true, Ordering::Release), + } + send_update(&dirty, &output, &updates, &progress); tokio::time::sleep(UPDATE_THROTTLE).await; } } @@ -572,10 +635,16 @@ impl UpdateThrottle { task, output, updates, + progress, } } fn mark(&self) { + *self + .progress + .last_output + .lock() + .unwrap_or_else(|error| error.into_inner()) = Instant::now(); self.dirty.store(true, Ordering::Release); self.wake.notify_one(); } @@ -583,11 +652,22 @@ impl UpdateThrottle { /// Stop, sending whatever has not been sent. fn finish(self) { self.task.abort(); - send_update(&self.dirty, &self.output, &self.updates); + send_update(&self.dirty, &self.output, &self.updates, &self.progress); } } -fn send_update(dirty: &AtomicBool, output: &Mutex, updates: &ToolUpdates) { +impl Drop for UpdateThrottle { + fn drop(&mut self) { + self.task.abort(); + } +} + +fn send_update( + dirty: &AtomicBool, + output: &Mutex, + updates: &ToolUpdates, + progress: &ShellProgress, +) { if !dirty.swap(false, Ordering::AcqRel) { return; } @@ -595,9 +675,16 @@ fn send_update(dirty: &AtomicBool, output: &Mutex, updates: & .lock() .unwrap_or_else(|error| error.into_inner()) .snapshot(true); + let mut details = snapshot_details(&snapshot).unwrap_or_else(|| json!({})); + details["shellExecution"] = json!({ + "cwd": progress.cwd.to_string_lossy(), + "timeoutSeconds": progress.timeout, + "elapsedSeconds": progress.started.elapsed().as_secs(), + "outputIdleSeconds": progress.last_output.lock().unwrap_or_else(|error| error.into_inner()).elapsed().as_secs(), + }); updates.send(AgentToolResult { content: vec![Content::text(snapshot.content.clone())], - details: snapshot_details(&snapshot), + details: Some(details), ..AgentToolResult::default() }); } @@ -673,7 +760,18 @@ impl AgentTool for ShellTool { Some(prefix) => format!("{prefix}\n{}", params.command), None => params.command, }; - let cwd = self.context.cwd().unwrap_or_else(|| self.cwd.clone()); + let timeout = params.timeout.or(self.default_timeout); + resolve_timeout(timeout)?; + if params + .workdir + .as_ref() + .is_some_and(|cwd| !cwd.is_absolute()) + { + return Err("Working directory must be an absolute path".into()); + } + let cwd = params + .workdir + .unwrap_or_else(|| self.context.cwd().unwrap_or_else(|| self.cwd.clone())); let spawn = self.spawn_context(command, cwd); let output = Arc::new(Mutex::new(OutputAccumulator::new_with_output_dir( @@ -683,6 +781,8 @@ impl AgentTool for ShellTool { let throttle = Arc::new(Mutex::new(Some(UpdateThrottle::start( output.clone(), invocation.updates.clone(), + spawn.cwd.clone(), + timeout, )))); let on_data: OnData = { let (output, throttle) = (output.clone(), throttle.clone()); @@ -708,7 +808,7 @@ impl AgentTool for ShellTool { ExecOptions { on_data, cancel: invocation.cancel.clone(), - timeout: params.timeout, + timeout, env: Some(spawn.env), }, ) diff --git a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs index b84184474..82568b462 100644 --- a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs +++ b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs @@ -147,6 +147,208 @@ async fn a_timeout_stops_the_command() { assert_eq!(error, "Invalid timeout: must be a finite number of seconds"); } +#[cfg(unix)] +#[tokio::test] +async fn each_call_uses_the_explicit_worktree_or_the_session_folder() { + let fixture = tempfile::tempdir().unwrap(); + let project = fixture.path().join("checkout"); + let worktree = fixture.path().join("linked-worktree"); + fs::create_dir(&project).unwrap(); + fs::create_dir(&worktree).unwrap(); + fs::write(project.join("window.rs"), "checkout").unwrap(); + fs::write(worktree.join("window.rs"), "worktree").unwrap(); + let tool = bash(&project, BashToolOptions::default()); + let result = run( + &tool, + json!({"command": "cat ./window.rs", "workdir": worktree}), + ) + .await + .unwrap(); + assert_eq!(content_text(&result.content), "worktree"); + run(&tool, json!({"command": "cd .."})).await.unwrap(); + let result = run(&tool, json!({"command": "cat ./window.rs"})) + .await + .unwrap(); + assert_eq!(content_text(&result.content), "checkout"); + + for workdir in [fixture.path().join("missing"), project.join("window.rs")] { + let error = run( + &tool, + json!({"command": "touch must-not-run", "workdir": workdir}), + ) + .await + .unwrap_err(); + assert!( + error.starts_with("Working directory does not exist"), + "{error}" + ); + } + let error = run( + &tool, + json!({"command": "touch must-not-run", "workdir": "../../"}), + ) + .await + .unwrap_err(); + assert_eq!(error, "Working directory must be an absolute path"); + let relative = bash(Path::new("."), BashToolOptions::default()); + assert_eq!( + run(&relative, json!({"command": "pwd"})).await.unwrap_err(), + "Working directory must be an absolute path" + ); + assert!(!project.join("must-not-run").exists()); +} + +#[cfg(unix)] +#[tokio::test] +async fn the_host_default_deadline_is_used_unless_the_call_overrides_it() { + let dir = tempfile::tempdir().unwrap(); + let tool = bash( + dir.path(), + BashToolOptions { + default_timeout: Some(0.05), + ..BashToolOptions::default() + }, + ); + let error = tokio::time::timeout( + Duration::from_secs(5), + run(&tool, json!({"command": "sleep 30"})), + ) + .await + .expect("default deadline ends the call") + .unwrap_err(); + assert_eq!(error, "Command timed out after 0.05 seconds"); + let result = run( + &tool, + json!({"command": "sleep 0.1; printf done", "timeout": 5}), + ) + .await + .unwrap(); + assert_eq!(content_text(&result.content), "done"); +} + +/// A finite, disposable substitute for a kernel file that waits indefinitely for +/// new data. The held writer prevents EOF; neither side of the pipeline produces +/// output, so head cannot bound its runtime. +#[cfg(unix)] +#[tokio::test] +async fn blocked_pipeline_is_stopped_by_cancel_deadline_and_dropped_future() { + use std::ffi::CString; + use std::os::unix::fs::OpenOptionsExt; + + for end in ["cancel", "deadline", "drop"] { + let dir = tempfile::tempdir().unwrap(); + fs::create_dir_all(dir.path().join("scratch/tmp")).unwrap(); + let fifo = CString::new(dir.path().join("blocked").as_os_str().as_encoded_bytes()).unwrap(); + // SAFETY: a valid, NUL-terminated path inside a disposable test directory. + assert_eq!(unsafe { libc::mkfifo(fifo.as_ptr(), 0o600) }, 0); + let _reader = fs::OpenOptions::new() + .read(true) + .custom_flags(libc::O_NONBLOCK) + .open(dir.path().join("blocked")) + .unwrap(); + let _writer = fs::OpenOptions::new() + .write(true) + .custom_flags(libc::O_NONBLOCK) + .open(dir.path().join("blocked")) + .unwrap(); + let tool = Arc::new(bash(dir.path(), BashToolOptions::default())); + let cancel = CancellationToken::new(); + let call = tokio::spawn({ + let (tool, cancel) = (tool.clone(), cancel.clone()); + async move { + run_with(&tool, json!({ + "command": "printf '%s' $$ > shell.pid; cd scratch/tmp && sh -c 'printf %s $$ > ../../reader.pid; exec cat ../../blocked' | sh -c 'printf %s $$ > ../../head.pid; exec head -c 1'", + "timeout": if end == "deadline" { 0.5 } else { 30.0 }, + }), cancel).await + } + }); + let mut pids = Vec::new(); + for name in ["shell.pid", "reader.pid", "head.pid"] { + let pid = tokio::time::timeout(Duration::from_secs(5), async { + loop { + if let Some(pid) = fs::read_to_string(dir.path().join(name)) + .ok() + .and_then(|text| text.parse::().ok()) + { + break pid; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("pipeline member records its pid"); + pids.push(pid); + } + match end { + "cancel" => cancel.cancel(), + "drop" => call.abort(), + _ => {} + } + let result = tokio::time::timeout(Duration::from_secs(5), call) + .await + .expect("blocked call ends"); + match end { + "drop" => assert!(result.unwrap_err().is_cancelled()), + "cancel" => assert_eq!(result.unwrap().unwrap_err(), "Command aborted"), + _ => assert_eq!( + result.unwrap().unwrap_err(), + "Command timed out after 0.5 seconds" + ), + } + // Check every member, not just the shell. Linux PID 1 need not reap an + // orphan immediately, so a zombie counts as terminated, never as alive. + for pid in pids { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + // SAFETY: signal zero only checks existence. + if unsafe { libc::kill(pid, 0) } != 0 { + break; + } + #[cfg(target_os = "linux")] + if fs::read_to_string(format!("/proc/{pid}/stat")).is_ok_and(|stat| { + stat.rsplit_once(") ") + .is_some_and(|(_, rest)| rest.starts_with("Z ")) + }) { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("pipeline member is terminated"); + } + } +} + +#[cfg(unix)] +#[tokio::test] +async fn deadline_also_covers_a_shell_that_does_not_read_its_command_from_stdin() { + let dir = tempfile::tempdir().unwrap(); + let config = ShellConfig { + shell: PathBuf::from("/bin/bash"), + args: vec!["-c".into(), "sleep 30".into()], + transport: CommandTransport::Stdin, + }; + let result = tokio::time::timeout( + Duration::from_secs(5), + run_local( + &config, + &"x".repeat(1024 * 1024), + dir.path(), + Some(Duration::from_millis(100)), + ExecOptions { + on_data: Arc::new(|_| {}), + cancel: CancellationToken::new(), + timeout: Some(0.1), + env: None, + }, + ), + ) + .await + .expect("deadline covers blocked stdin write"); + assert_eq!(result, Err(ExecError::TimedOut(0.1))); +} + #[tokio::test] async fn a_stopped_or_timed_out_command_points_at_its_full_output() { let dir = tempfile::tempdir().unwrap(); diff --git a/apps/maple-agent/crates/pi-coding-agent/src/tools/tests.rs b/apps/maple-agent/crates/pi-coding-agent/src/tools/tests.rs index d043aa03f..7cba1ea5e 100644 --- a/apps/maple-agent/crates/pi-coding-agent/src/tools/tests.rs +++ b/apps/maple-agent/crates/pi-coding-agent/src/tools/tests.rs @@ -616,7 +616,7 @@ fn every_built_in_tool_is_created_with_the_defaults_active() { ); assert_eq!( bash.guidelines, - ["You can inspect TEST_* environment variables for current model and session details."] + ShellTool::bash(dir.path(), BashToolOptions::default(), context()).guidelines() ); } From 301633f9f02d72094cd3117b5334b39fa0b8eace Mon Sep 17 00:00:00 2001 From: w3irdrobot Date: Thu, 29 Oct 2026 22:43:04 +0000 Subject: [PATCH 2/3] fix(agent): settle shell progress before completion and allow slow builds --- apps/maple-agent/README.md | 7 ++- apps/maple-agent/crates/UPSTREAM.md | 2 +- .../maple-agent/src/agent/shell_tests.rs | 2 +- .../crates/maple-agent/src/agent/timeline.rs | 6 ++- .../maple-agent/src/agent/timeline/tests.rs | 54 +++++++++++++++++++ .../crates/maple-agent/src/agent/tools/mod.rs | 5 +- .../maple-agent/src/agent/tools/tests.rs | 4 +- .../crates/pi-coding-agent/src/tools/bash.rs | 16 +++--- .../pi-coding-agent/src/tools/bash/tests.rs | 39 ++++++++++++++ 9 files changed, 118 insertions(+), 17 deletions(-) diff --git a/apps/maple-agent/README.md b/apps/maple-agent/README.md index 73fa85561..bf98474f1 100644 --- a/apps/maple-agent/README.md +++ b/apps/maple-agent/README.md @@ -72,8 +72,11 @@ agent runtime; its desktop Agent page points users to Maple Work. approval. - Shell tool calls start in the task's saved folder on the execution computer. An absolute `workdir` selects another checkout/worktree for one call; a `cd` - inside a command does not affect later calls. Maple supplies a 120-second - default deadline, and the call can request a longer `timeout` explicitly. + inside a command does not affect later calls. Maple supplies a 30-minute + default deadline to allow slow builds, and the call can request a longer + `timeout` explicitly. Search guidance asks for a short timeout (normally + 30 seconds) and a narrower scope if it expires; that shorter limit applies + only when the call requests it. Quiet commands report their starting folder, elapsed time, time since output and deadline every five seconds, including on collapsed tool cards. Stop and deadlines kill the shell's process group on Unix (the process tree on Windows). diff --git a/apps/maple-agent/crates/UPSTREAM.md b/apps/maple-agent/crates/UPSTREAM.md index a425bfeaf..50a97064f 100644 --- a/apps/maple-agent/crates/UPSTREAM.md +++ b/apps/maple-agent/crates/UPSTREAM.md @@ -19,7 +19,7 @@ complete audit of every earlier difference. | `SessionBase`, `SessionManager::open_partial` and storage context hooks | Resume a long conversation from the required compaction context without loading all saved history. Original entry IDs and parents remain intact. SQL, caches and paging stay in Maple. Complete-load behavior remains available; operations requiring an unloaded prefix must require a complete reload. | Full-versus-partial projection comparisons, inherited model/thinking/system state, compaction, retry and fork/export restrictions in `pi-coding-agent/src/session.rs`. | | Explicit `flush` and `has_unpersisted` | Expose pending persistence so an embedding host can retry. Maple decides whether shutdown or eviction may finish. Consuming an error must not discard the pending entries. | Persistence failure/retry tests in `pi-coding-agent/src/session.rs` and Maple store tests. | | Optional tool-output directory | Let an embedding host choose the lifetime and location of overflow files. Standalone defaults remain available. An invalid explicit directory fails without a fallback. Maple owns task paths, account policy and cleanup. | Shell, user-bash and private-directory tests in `pi-coding-agent`. | -| Optional shell default deadline, absolute per-call `workdir`, and quiet status updates | An embedding host needs to bound silent commands and make their execution folder and deadline visible. Pi retains no default deadline; Maple selects 120 seconds. Workdir is explicit per call and never changes the session folder. Status travels through existing partial-result events, with no prompt/output logging. | Disposable-folder and blocked FIFO pipeline tests in `tools/bash/tests.rs`; saved-task cold-load and quiet-progress/Stop tests in Maple's `agent/shell_tests.rs`. | +| Optional shell default deadline, absolute per-call `workdir`, and quiet status updates | An embedding host needs to bound silent commands and make their execution folder and deadline visible. Pi retains no default deadline; Maple selects 30 minutes for slow builds and guides searches to request a shorter timeout. Workdir is explicit per call and never changes the session folder. Status travels through existing partial-result events, with no prompt/output logging. | Disposable-folder and blocked FIFO pipeline tests in `tools/bash/tests.rs`; saved-task cold-load and quiet-progress/Stop tests in Maple's `agent/shell_tests.rs`. | These are deliberate embedding adaptations. They do not authorize changing the agent loop, compaction decisions or the meaning and ordering of Pi events. diff --git a/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs b/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs index 6764339a2..a4a98825f 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs @@ -96,7 +96,7 @@ async fn quiet_shell_reports_effective_folder_deadline_and_idle_time_until_stop( assert_eq!(tool.status.as_deref(), Some("running")); let progress = &tool.output.as_ref().unwrap()["structuredContent"]["shellExecution"]; assert_eq!(progress["cwd"], json!(worktree.path())); - assert_eq!(progress["timeoutSeconds"], 120.0); + assert_eq!(progress["timeoutSeconds"], 1800.0); assert!(progress["outputIdleSeconds"].as_u64().unwrap() >= 5); assert!(!progress.to_string().contains("fixture-private-command")); let identity = tool.live.clone(); diff --git a/apps/maple-agent/crates/maple-agent/src/agent/timeline.rs b/apps/maple-agent/crates/maple-agent/src/agent/timeline.rs index b9a98dbee..93787e7f8 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/timeline.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/timeline.rs @@ -632,11 +632,13 @@ impl LiveTimeline { is_error, .. } => { - let Some(id) = self.tool_rows.get(tool_call_id) else { + // Retire the occurrence so a delayed partial result cannot + // overwrite its final status or output while other tools run. + let Some(id) = self.tool_rows.remove(tool_call_id) else { return Vec::new(); }; vec![tool_result_item( - id.clone(), + id, &result.content, result.details.as_ref(), *is_error, diff --git a/apps/maple-agent/crates/maple-agent/src/agent/timeline/tests.rs b/apps/maple-agent/crates/maple-agent/src/agent/timeline/tests.rs index 5973b6614..fbd106fbf 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/timeline/tests.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/timeline/tests.rs @@ -87,6 +87,60 @@ fn shown(items: &[AgentTimelineItem]) -> Vec { .collect() } +#[test] +fn late_tool_progress_cannot_replace_a_final_result_while_another_tool_runs() { + for (text, is_error) in [ + ("finished", false), + ("Command aborted", true), + ("Command timed out after 30 seconds", true), + ] { + let mut live = LiveTimeline::default(); + let mut reply = AssistantMessage::empty(&FauxProvider::new().model()); + reply.content = vec![ + faux_tool_call("bash", json!({"command": "first"})), + faux_tool_call("bash", json!({"command": "second"})), + ]; + let calls: Vec<_> = reply.tool_calls().cloned().collect(); + let mut rows = live.rows(&AgentEvent::MessageEnd { + message: SessionMessage::Llm(Message::Assistant(reply)), + }); + let update = |call: &ToolCall| AgentEvent::ToolExecutionUpdate { + tool_call_id: call.id.clone(), + tool_name: call.name.clone(), + args: Value::Object(call.arguments.clone()), + partial_result: AgentToolResult::text("still running"), + }; + for call in &calls { + for row in live.rows(&AgentEvent::ToolExecutionStart { + tool_call_id: call.id.clone(), + tool_name: call.name.clone(), + args: Value::Object(call.arguments.clone()), + }) { + merge_into(&mut rows, row); + } + assert_eq!(live.rows(&update(call)).len(), 1); + } + for row in live.rows(&AgentEvent::ToolExecutionEnd { + tool_call_id: calls[0].id.clone(), + tool_name: calls[0].name.clone(), + result: AgentToolResult::text(text), + is_error, + }) { + merge_into(&mut rows, row); + } + // An already-running reporter can deliver a partial result after End. + assert!(live.rows(&update(&calls[0])).is_empty()); + let active = live.rows(&update(&calls[1])); + assert_eq!(active.len(), 1); + assert_eq!(active[0].status.as_deref(), Some("running")); + assert_eq!( + rows[0].status.as_deref(), + Some(if is_error { "failed" } else { "completed" }) + ); + assert_eq!(rows[0].output.as_ref().unwrap()["text"], text); + } +} + #[tokio::test] async fn live_rows_are_the_rows_a_reload_shows() { let faux = FauxProvider::new(); diff --git a/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs b/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs index fd0f8d6b4..c91691615 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs @@ -33,8 +33,9 @@ use crate::maple_api::MapleWebTransport; /// The developer's choice of bash. pub(crate) const SHELL_ENV: &str = "MAPLE_SHELL"; -/// Silent searches must not hold a task forever. Builds can request longer explicitly. -const SHELL_DEFAULT_TIMEOUT: f64 = 120.0; +/// Leave room for slow builds while bounding commands that never finish. +/// Searches request a shorter timeout through the shell tool's guidelines. +const SHELL_DEFAULT_TIMEOUT: f64 = 30.0 * 60.0; /// How a task's tools are set up. pub(crate) struct TaskTools { diff --git a/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs b/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs index 72866b3db..9efdb1862 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs @@ -128,8 +128,8 @@ fn the_model_gets_read_a_shell_edit_and_write() { }; assert_eq!(tools.builtin, ["read", shell, "edit", "write"]); assert!(tools.options.powershell.spawn_hook.is_some()); - assert_eq!(tools.options.bash.default_timeout, Some(120.0)); - assert_eq!(tools.options.powershell.default_timeout, Some(120.0)); + assert_eq!(tools.options.bash.default_timeout, Some(1800.0)); + assert_eq!(tools.options.powershell.default_timeout, Some(1800.0)); } fn tool(tools: &TaskTools, name: &str) -> Arc { diff --git a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs index 0c7b23dce..08ab605e1 100644 --- a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs +++ b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs @@ -527,7 +527,7 @@ impl ShellTool { /// The guideline that goes with the tool's snippet. pub fn guidelines(&self) -> Vec { let mut rules = vec![ - "For repository searches, set workdir to the absolute intended checkout/worktree and search within it. Verify the folder before using parent paths; never guess ../../ from a temporary folder. Avoid recursive searches of filesystem roots or virtual filesystems such as /proc, /sys and /dev. head limits output, not runtime; use a timeout for searches.".to_string(), + "For repository searches, set workdir to the absolute intended checkout/worktree and search within it. Verify the folder before using parent paths; never guess ../../ from a temporary folder. Avoid recursive searches of filesystem roots or virtual filesystems such as /proc, /sys and /dev. head limits output, not runtime; request a short timeout (normally 30 seconds) for searches and narrow the scope if it expires.".to_string(), ]; if self.expose_session_environment { rules.push(format!( @@ -649,9 +649,11 @@ impl UpdateThrottle { self.wake.notify_one(); } - /// Stop, sending whatever has not been sent. - fn finish(self) { + /// Join the reporter before the tool returns: abort alone cannot stop a + /// synchronous send already running on another executor thread. + async fn finish(mut self) { self.task.abort(); + let _ = (&mut self.task).await; send_update(&self.dirty, &self.output, &self.updates, &self.progress); } } @@ -819,12 +821,12 @@ impl AgentTool for ShellTool { output.finish(); (output.snapshot(true), output.last_line_bytes()) }; - if let Some(throttle) = throttle + let finished_throttle = throttle .lock() .unwrap_or_else(|error| error.into_inner()) - .take() - { - throttle.finish(); + .take(); + if let Some(throttle) = finished_throttle { + throttle.finish().await; } // The kept output reaches the model either way; a failed save loses only the // rest, and the note says the file could not be saved. diff --git a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs index 82568b462..7f5e1641b 100644 --- a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs +++ b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs @@ -73,6 +73,45 @@ fn full_output_path(text: &str) -> PathBuf { PathBuf::from(&text[start..end]) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn finishing_progress_waits_for_an_in_flight_reporter() { + // Model a reporter inside synchronous send_update on another worker. Abort + // cannot interrupt that code; the final result must wait until it returns. + let gate = Arc::new(std::sync::Barrier::new(2)); + let reported = Arc::new(AtomicBool::new(false)); + let (started, ready) = tokio::sync::oneshot::channel(); + let task = tokio::spawn({ + let (gate, reported) = (gate.clone(), reported.clone()); + async move { + started.send(()).unwrap(); + gate.wait(); + reported.store(true, Ordering::Release); + } + }); + ready.await.unwrap(); + let throttle = UpdateThrottle { + dirty: Arc::new(AtomicBool::new(false)), + wake: Arc::new(Notify::new()), + task, + output: Arc::new(Mutex::new(OutputAccumulator::new("progress-fixture"))), + updates: ToolUpdates::none(), + progress: Arc::new(ShellProgress { + cwd: PathBuf::new(), + timeout: None, + started: Instant::now(), + last_output: Mutex::new(Instant::now()), + }), + }; + let finish = throttle.finish(); + tokio::pin!(finish); + let first_poll = futures_util::poll!(&mut finish); + // Always release the worker before asserting, including on a regression. + gate.wait(); + assert!(first_poll.is_pending(), "completion must join the reporter"); + finish.await; + assert!(reported.load(Ordering::Acquire)); +} + #[cfg(unix)] #[tokio::test] async fn commands_run_and_report_their_exit_code() { From 0a48340accc4bb083fd2f4c189903ead1742bebf Mon Sep 17 00:00:00 2001 From: w3irdrobot Date: Fri, 30 Oct 2026 00:56:38 +0000 Subject: [PATCH 3/3] fix(agent): keep shell safeguards in Maple integration --- apps/maple-agent/CLAUDE.md | 6 + apps/maple-agent/README.md | 12 +- apps/maple-agent/crates/UPSTREAM.md | 1 - .../crates/maple-agent/src/agent/runs.rs | 16 +- .../src/agent/runs/shell_progress.rs | 275 ++++++++++++++++++ .../crates/maple-agent/src/agent/runtime.rs | 6 +- .../maple-agent/src/agent/shell_tests.rs | 20 +- .../crates/maple-agent/src/agent/tools/mod.rs | 13 +- .../maple-agent/src/agent/tools/shell.rs | 90 ++++++ .../src/agent/tools/shell/tests.rs | 270 +++++++++++++++++ .../maple-agent/src/agent/tools/tests.rs | 4 +- .../crates/pi-coding-agent/src/tools/bash.rs | 180 +++--------- .../pi-coding-agent/src/tools/bash/tests.rs | 241 --------------- .../crates/pi-coding-agent/src/tools/tests.rs | 2 +- 14 files changed, 730 insertions(+), 406 deletions(-) create mode 100644 apps/maple-agent/crates/maple-agent/src/agent/runs/shell_progress.rs create mode 100644 apps/maple-agent/crates/maple-agent/src/agent/tools/shell.rs create mode 100644 apps/maple-agent/crates/maple-agent/src/agent/tools/shell/tests.rs diff --git a/apps/maple-agent/CLAUDE.md b/apps/maple-agent/CLAUDE.md index 7e04621de..8d271ea11 100644 --- a/apps/maple-agent/CLAUDE.md +++ b/apps/maple-agent/CLAUDE.md @@ -9,6 +9,12 @@ GPUI desktop app for Maple. Workspace crates: `app` (package `maple-agent-app`, `crates/pi-agent-core`, `crates/pi-coding-agent`, which must not depend on Maple crates. See `README.md` for the layout, prerequisites, and command line modes. +Keep `crates/pi-*` unchanged for Maple product work. Implement policy and UI +behavior through the existing host, extension, and operations interfaces in +`crates/maple-agent` and `app`. Exceptions must address a critical need and +explain why existing interfaces are insufficient; see `crates/UPSTREAM.md` for +upstream review work. + Packaged Dev and Prod OAuth use their compiled Auth origins and a one-attempt loopback listener in `app/src/hosted_oauth.rs`; unpackaged builds retain the provider callback paste flow. Keep the listener bound to `127.0.0.1` through diff --git a/apps/maple-agent/README.md b/apps/maple-agent/README.md index bf98474f1..d6c106946 100644 --- a/apps/maple-agent/README.md +++ b/apps/maple-agent/README.md @@ -71,15 +71,19 @@ agent runtime; its desktop Agent page points users to Maple Work. and a context-window indicator. Every tool call runs without asking for approval. - Shell tool calls start in the task's saved folder on the execution computer. - An absolute `workdir` selects another checkout/worktree for one call; a `cd` - inside a command does not affect later calls. Maple supplies a 30-minute - default deadline to allow slow builds, and the call can request a longer + A `cd` inside a command does not affect later calls. Repository searches + should name the intended checkout/worktree with explicit absolute paths. + Maple supplies a 30-minute default deadline to allow slow builds, and the + call can request a longer `timeout` explicitly. Search guidance asks for a short timeout (normally 30 seconds) and a narrower scope if it expires; that shorter limit applies only when the call requests it. Quiet commands report their starting folder, elapsed time, time since output and deadline every five seconds, including on collapsed tool cards. Stop and - deadlines kill the shell's process group on Unix (the process tree on Windows). + deadlines use Pi's process-group cleanup on Unix (process-tree cleanup on Windows). + Maple bounds the whole operation, including setup and stdin delivery, and allows + up to one second for cooperative cancellation before dropping a stuck operation. + The execution policy and progress reporting live in Maple; the Pi crates are unchanged. These are execution safeguards, not a filesystem sandbox: repository searches must still choose the correct folder and bounded scope, and deliberately detached processes can escape a Unix process group. `head` bounds output, diff --git a/apps/maple-agent/crates/UPSTREAM.md b/apps/maple-agent/crates/UPSTREAM.md index 50a97064f..a6a776236 100644 --- a/apps/maple-agent/crates/UPSTREAM.md +++ b/apps/maple-agent/crates/UPSTREAM.md @@ -19,7 +19,6 @@ complete audit of every earlier difference. | `SessionBase`, `SessionManager::open_partial` and storage context hooks | Resume a long conversation from the required compaction context without loading all saved history. Original entry IDs and parents remain intact. SQL, caches and paging stay in Maple. Complete-load behavior remains available; operations requiring an unloaded prefix must require a complete reload. | Full-versus-partial projection comparisons, inherited model/thinking/system state, compaction, retry and fork/export restrictions in `pi-coding-agent/src/session.rs`. | | Explicit `flush` and `has_unpersisted` | Expose pending persistence so an embedding host can retry. Maple decides whether shutdown or eviction may finish. Consuming an error must not discard the pending entries. | Persistence failure/retry tests in `pi-coding-agent/src/session.rs` and Maple store tests. | | Optional tool-output directory | Let an embedding host choose the lifetime and location of overflow files. Standalone defaults remain available. An invalid explicit directory fails without a fallback. Maple owns task paths, account policy and cleanup. | Shell, user-bash and private-directory tests in `pi-coding-agent`. | -| Optional shell default deadline, absolute per-call `workdir`, and quiet status updates | An embedding host needs to bound silent commands and make their execution folder and deadline visible. Pi retains no default deadline; Maple selects 30 minutes for slow builds and guides searches to request a shorter timeout. Workdir is explicit per call and never changes the session folder. Status travels through existing partial-result events, with no prompt/output logging. | Disposable-folder and blocked FIFO pipeline tests in `tools/bash/tests.rs`; saved-task cold-load and quiet-progress/Stop tests in Maple's `agent/shell_tests.rs`. | These are deliberate embedding adaptations. They do not authorize changing the agent loop, compaction decisions or the meaning and ordering of Pi events. diff --git a/apps/maple-agent/crates/maple-agent/src/agent/runs.rs b/apps/maple-agent/crates/maple-agent/src/agent/runs.rs index f18feec9d..fd59b85af 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/runs.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/runs.rs @@ -41,7 +41,9 @@ use super::{ }; mod live; +mod shell_progress; use live::LiveSnapshot; +use shell_progress::ShellProgress; const AGENT_RUN_EVENT_CAPACITY: usize = 256; const MAX_DESKTOP_QUEUE_ITEMS: usize = 16; @@ -1588,7 +1590,11 @@ impl AgentRuntime { } = run; let attachments = AgentAttachmentStore::new(Arc::clone(self.store.store()), self.store.account()); - let watch = RunWatch::default(); + let watch = RunWatch { + state: Arc::default(), + shells: ShellProgress::new(session.extension_context().cwd()), + }; + let shell_status = watch.shells.start(events.clone()); let listener = session.subscribe({ let events = events.clone(); let watch = watch.clone(); @@ -1749,6 +1755,7 @@ impl AgentRuntime { run.accepting = false; } session.unsubscribe(listener); + shell_status.stop().await; let cancelled = stopped.is_cancelled(); if cancelled || failure.is_some() || deferred.failure.is_some() { @@ -1982,9 +1989,10 @@ struct DrivenRun { /// What a run's listener saw: the usage of its replies and whether Pi /// retried. -#[derive(Clone, Default)] +#[derive(Clone)] struct RunWatch { state: Arc>, + shells: Arc, } #[derive(Default)] @@ -2038,9 +2046,7 @@ impl RunWatch { }; (projected, tool, false) }; - for row in rows { - events.publish_row(AgentRunEvent::TimelineItem(row), tool, completed); - } + self.shells.publish(event, rows, tool, completed, events); } AgentSessionEvent::EntryAppended { entry } => events.note_source(entry), AgentSessionEvent::RetryStart { .. } => lock(&self.state).retried = true, diff --git a/apps/maple-agent/crates/maple-agent/src/agent/runs/shell_progress.rs b/apps/maple-agent/crates/maple-agent/src/agent/runs/shell_progress.rs new file mode 100644 index 000000000..b1c3e464a --- /dev/null +++ b/apps/maple-agent/crates/maple-agent/src/agent/runs/shell_progress.rs @@ -0,0 +1,275 @@ +//! Run-owned shell status. Pi's partial output contract stays unchanged. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use pi_agent_core::AgentEvent; +use pi_ai::Content; +use pi_coding_agent::SessionMessage; +use serde_json::json; +use tokio::sync::Notify; + +use super::{AgentRunEvent, AgentTimelineItem, RunEvents, lock}; +use crate::agent::tools::shell::timeout_seconds; + +const STATUS_INTERVAL: Duration = Duration::from_secs(5); + +pub(super) struct ShellProgress { + cwd: Option, + calls: Mutex>, + wake: Notify, +} + +struct RunningShell { + row: AgentTimelineItem, + started: Instant, + last_output: Instant, + timeout: f64, +} + +impl RunningShell { + fn decorate(&self, row: &mut AgentTimelineItem, cwd: &std::path::Path) { + let output = row.output.get_or_insert_with(|| json!({})); + let details = &mut output["structuredContent"]; + if !details.is_object() { + *details = json!({}); + } + details["shellExecution"] = json!({ + "cwd": cwd.to_string_lossy(), + "timeoutSeconds": self.timeout, + "elapsedSeconds": self.started.elapsed().as_secs(), + "outputIdleSeconds": self.last_output.elapsed().as_secs(), + }); + } +} + +impl ShellProgress { + pub(super) fn new(cwd: Option) -> Arc { + Arc::new(Self { + cwd, + calls: Mutex::new(HashMap::new()), + wake: Notify::new(), + }) + } + + pub(super) fn publish( + &self, + event: &AgentEvent, + mut rows: Vec, + tool: Option<&str>, + completed: bool, + events: &RunEvents, + ) { + // Hold this through publication: a heartbeat and End cannot overtake + // each other between checking ownership and sending their rows. + let mut calls = lock(&self.calls); + match event { + AgentEvent::ToolExecutionStart { + tool_call_id, + tool_name, + args, + } if matches!(tool_name.as_str(), "bash" | "powershell") => { + if self.cwd.is_some() + && let Some(row) = rows.first() + { + let now = Instant::now(); + calls.insert( + tool_call_id.clone(), + RunningShell { + row: row.clone(), + started: now, + last_output: now, + timeout: timeout_seconds(args.get("timeout").and_then(|v| v.as_f64())), + }, + ); + self.wake.notify_one(); + } + } + AgentEvent::ToolExecutionUpdate { + tool_call_id, + partial_result, + tool_name, + .. + } => { + // Projection and publication are separate steps. An update + // projected before End may arrive here after it; fence that + // stale row using the same lock as terminal publication. + if matches!(tool_name.as_str(), "bash" | "powershell") + && self.cwd.is_some() + && !calls.contains_key(tool_call_id) + { + return; + } + if let Some(call) = calls.get_mut(tool_call_id) + && partial_result.content.iter().any( + |content| matches!(content, Content::Text(text) if !text.text.is_empty()), + ) + { + call.last_output = Instant::now(); + } + } + AgentEvent::ToolExecutionEnd { tool_call_id, .. } => { + calls.remove(tool_call_id); + } + _ => {} + } + for row in &mut rows { + if let Some(call) = tool.and_then(|tool| calls.get_mut(tool)) + && let Some(cwd) = &self.cwd + { + call.decorate(row, cwd); + call.row = row.clone(); + } + events.publish_row(AgentRunEvent::TimelineItem(row.clone()), tool, completed); + } + } + + pub(super) fn start(self: &Arc, events: RunEvents) -> ShellStatusTask { + let progress = Arc::clone(self); + ShellStatusTask(tokio::spawn(async move { + loop { + progress.wake.notified().await; + loop { + tokio::time::sleep(STATUS_INTERVAL).await; + if !progress.heartbeat(&events) { + break; + } + } + } + })) + } + + fn heartbeat(&self, events: &RunEvents) -> bool { + let calls = lock(&self.calls); + if let Some(cwd) = &self.cwd { + for (id, call) in calls.iter() { + let mut row = call.row.clone(); + call.decorate(&mut row, cwd); + events.publish_row(AgentRunEvent::TimelineItem(row), Some(id), false); + } + } + !calls.is_empty() + } +} + +pub(super) struct ShellStatusTask(tokio::task::JoinHandle<()>); + +impl ShellStatusTask { + pub(super) async fn stop(mut self) { + self.0.abort(); + let _ = (&mut self.0).await; + } +} + +impl Drop for ShellStatusTask { + fn drop(&mut self) { + self.0.abort(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::agent::{AgentEventDispatcher, tests::Recorder, timeline}; + use pi_agent_core::AgentToolResult; + + #[test] + fn heartbeat_keeps_output_and_truncation_but_never_revives_a_finished_call() { + let dir = tempfile::tempdir().unwrap(); + let progress = ShellProgress::new(Some(dir.path().to_path_buf())); + let (events, mut receiver) = RunEvents::new( + AgentEventDispatcher::new(Arc::new(Recorder::default())), + "task", + "run", + ); + for id in ["first", "second"] { + let mut row = timeline::tool_result_item(id.into(), &[], None, false, 0); + row.status = Some("running".into()); + progress.publish( + &AgentEvent::ToolExecutionStart { + tool_call_id: id.into(), + tool_name: "bash".into(), + args: json!({"command": "fixture"}), + }, + vec![row], + Some(id), + false, + &events, + ); + } + let partial = AgentToolResult::text("retained output") + .with_details(json!({"fullOutputPath": "fixture.log"})); + let mut row = timeline::tool_result_item( + "second".into(), + &partial.content, + partial.details.as_ref(), + false, + 0, + ); + row.status = Some("running".into()); + progress.publish( + &AgentEvent::ToolExecutionUpdate { + tool_call_id: "second".into(), + tool_name: "bash".into(), + args: json!({}), + partial_result: partial, + }, + vec![row], + Some("second"), + false, + &events, + ); + progress.publish( + &AgentEvent::ToolExecutionEnd { + tool_call_id: "first".into(), + tool_name: "bash".into(), + result: AgentToolResult::text("done"), + is_error: false, + }, + vec![timeline::tool_result_item( + "first".into(), + &[Content::text("done")], + None, + false, + 0, + )], + Some("first"), + false, + &events, + ); + while receiver.try_recv().is_ok() {} + let mut stale = + timeline::tool_result_item("first".into(), &[Content::text("late")], None, false, 0); + stale.status = Some("running".into()); + progress.publish( + &AgentEvent::ToolExecutionUpdate { + tool_call_id: "first".into(), + tool_name: "bash".into(), + args: json!({}), + partial_result: AgentToolResult::text("late"), + }, + vec![stale], + Some("first"), + false, + &events, + ); + assert!( + receiver.try_recv().is_err(), + "a previously projected update cannot overtake End" + ); + assert!(progress.heartbeat(&events)); + let AgentRunEvent::TimelineItem(row) = receiver.try_recv().unwrap() else { + panic!("heartbeat row") + }; + assert_eq!(row.id, "second"); + let output = row.output.unwrap(); + assert_eq!(output["text"], "retained output"); + assert_eq!(output["structuredContent"]["fullOutputPath"], "fixture.log"); + assert!( + receiver.try_recv().is_err(), + "the finished call emits no heartbeat" + ); + } +} diff --git a/apps/maple-agent/crates/maple-agent/src/agent/runtime.rs b/apps/maple-agent/crates/maple-agent/src/agent/runtime.rs index 020328c28..f0f94000d 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/runtime.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/runtime.rs @@ -793,7 +793,11 @@ impl AgentRuntime { /// delivered together, the saved thinking default, and the host's opening /// instructions after Pi's own. fn settings(&self) -> Settings { - let appended = self.host.harness_instructions().trim().to_string(); + let appended = format!( + "{}\n\n{}", + tools::shell::GUIDANCE, + self.host.harness_instructions().trim() + ); Settings { retry: RETRY, // The transport times out a response that does not start or diff --git a/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs b/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs index a4a98825f..afa033466 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/shell_tests.rs @@ -65,11 +65,22 @@ async fn quiet_shell_reports_effective_folder_deadline_and_idle_time_until_stop( harness.faux.push_message(vec![faux_tool_call( "bash", json!({ - "command": "sleep 30 # fixture-private-command", - "workdir": worktree.path(), + "command": "printf fixture-output; sleep 30 # fixture-private-command", }), )]); - let task = harness.create_task().await; + let task = harness + .handle + .create_session(Some(AgentCreateSessionRequest { + project_root: Some(worktree.path().to_string_lossy().into_owned()), + title: None, + model: None, + context_limit: None, + mcp_server_names: Some(Vec::new()), + })) + .await + .unwrap() + .session + .id; let mut run = harness.send(&task, "Wait for cancellation").await; let run_id = run.run_id.clone(); eventually(|| { @@ -94,10 +105,11 @@ async fn quiet_shell_reports_effective_folder_deadline_and_idle_time_until_stop( let rows = harness.recorder.live_rows(&run_id); let tool = rows.iter().find(|row| row.item_type == "tool").unwrap(); assert_eq!(tool.status.as_deref(), Some("running")); + assert_eq!(tool.output.as_ref().unwrap()["text"], "fixture-output"); let progress = &tool.output.as_ref().unwrap()["structuredContent"]["shellExecution"]; assert_eq!(progress["cwd"], json!(worktree.path())); assert_eq!(progress["timeoutSeconds"], 1800.0); - assert!(progress["outputIdleSeconds"].as_u64().unwrap() >= 5); + assert!(progress["outputIdleSeconds"].as_u64().unwrap() >= 4); assert!(!progress.to_string().contains("fixture-private-command")); let identity = tool.live.clone(); assert!(identity.is_some()); diff --git a/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs b/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs index c91691615..60c8c4146 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/tools/mod.rs @@ -13,6 +13,7 @@ mod desktop; mod read_image; +pub(super) mod shell; pub(crate) mod web; pub(super) use desktop::parse_user_questions; @@ -24,7 +25,8 @@ use std::sync::Arc; use pi_coding_agent::AgentSession; use pi_coding_agent::extensions::{Extension, RegisteredTool}; use pi_coding_agent::tools::{ - BashSpawnContext, BashSpawnHook, PowerShellToolOptions, ToolsOptions, set_env_var, + BashSpawnContext, BashSpawnHook, LocalShellOperations, PowerShellToolOptions, ToolsOptions, + set_env_var, }; use super::questions::QuestionBroker; @@ -33,9 +35,6 @@ use crate::maple_api::MapleWebTransport; /// The developer's choice of bash. pub(crate) const SHELL_ENV: &str = "MAPLE_SHELL"; -/// Leave room for slow builds while bounding commands that never finish. -/// Searches request a shorter timeout through the shell tool's guidelines. -const SHELL_DEFAULT_TIMEOUT: f64 = 30.0 * 60.0; /// How a task's tools are set up. pub(crate) struct TaskTools { @@ -70,11 +69,13 @@ pub(crate) fn task_tools(task: TaskToolsFor) -> TaskTools { .map(PathBuf::from); let hook = spawn_hook(task.session_id.clone(), task.login_path); let mut options = ToolsOptions::default(); - options.bash.default_timeout = Some(SHELL_DEFAULT_TIMEOUT); + options.bash.operations = Some(shell::bounded(LocalShellOperations::bash( + shell_path.clone(), + ))); options.bash.shell_path = shell_path.clone(); options.bash.spawn_hook = Some(hook.clone()); options.powershell = PowerShellToolOptions { - default_timeout: Some(SHELL_DEFAULT_TIMEOUT), + operations: Some(shell::bounded(LocalShellOperations::powershell())), spawn_hook: Some(hook), ..PowerShellToolOptions::default() }; diff --git a/apps/maple-agent/crates/maple-agent/src/agent/tools/shell.rs b/apps/maple-agent/crates/maple-agent/src/agent/tools/shell.rs new file mode 100644 index 000000000..3f7026034 --- /dev/null +++ b/apps/maple-agent/crates/maple-agent/src/agent/tools/shell.rs @@ -0,0 +1,90 @@ +//! Maple's execution policy around Pi's unchanged shell operations. + +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use pi_coding_agent::tools::{BashOperations, ExecError, ExecOptions, LocalShellOperations}; + +pub(crate) const DEFAULT_TIMEOUT_SECONDS: f64 = 30.0 * 60.0; +const CLEANUP_GRACE: Duration = Duration::from_secs(1); + +pub(crate) const GUIDANCE: &str = "Maple shell policy: bash and powershell have a 30-minute default runtime deadline, overriding the generic tool description's no-default behavior. Request a longer timeout explicitly for slow builds. Each command starts in the task's saved folder; cd does not persist between calls. For repository searches, use explicit absolute paths to the intended checkout/worktree. Verify the folder and never guess ../../ from a temporary folder. Avoid recursive searches of filesystem roots, /proc, /sys and /dev. head limits output, not runtime: request a short timeout (normally 30 seconds) for searches and narrow the scope if it expires."; + +pub(super) fn bounded(inner: LocalShellOperations) -> Arc { + Arc::new(BoundedShell { + inner: Arc::new(inner), + default_timeout: DEFAULT_TIMEOUT_SECONDS, + }) +} + +struct BoundedShell { + inner: Arc, + default_timeout: f64, +} + +pub(crate) fn timeout_seconds(requested: Option) -> f64 { + requested.unwrap_or(DEFAULT_TIMEOUT_SECONDS) +} + +#[async_trait] +impl BashOperations for BoundedShell { + async fn exec( + &self, + command: &str, + cwd: &Path, + mut options: ExecOptions, + ) -> Result, ExecError> { + let seconds = options.timeout.unwrap_or(self.default_timeout); + let duration = Duration::try_from_secs_f64(seconds) + .ok() + .filter(|duration| !duration.is_zero()) + .ok_or_else(|| { + ExecError::Failed( + "Invalid timeout: must be a positive finite number of seconds".into(), + ) + })?; + let cancel = options.cancel.clone(); + if cancel.is_cancelled() { + return Err(ExecError::Aborted); + } + let execution_cancel = cancel.child_token(); + options.cancel = execution_cancel.clone(); + options.timeout = Some(seconds); + // Bound setup and command delivery as well as process waiting. In + // particular, Pi's legacy stdin transport can block before its timer. + let execution = async { + if !cwd.is_absolute() { + return Err(ExecError::Failed( + "Working directory must be an absolute path".into(), + )); + } + if !tokio::fs::metadata(cwd) + .await + .is_ok_and(|metadata| metadata.is_dir()) + { + return Err(ExecError::Failed( + "Working directory is unavailable or is not a directory".into(), + )); + } + self.inner.exec(command, cwd, options).await + }; + tokio::pin!(execution); + let reason = tokio::select! { + biased; + _ = cancel.cancelled() => ExecError::Aborted, + result = &mut execution => return result, + _ = tokio::time::sleep(duration) => ExecError::TimedOut(seconds), + }; + // Let Pi reap the shell and clean up remaining pipe writers normally. + // If setup/stdin cannot respond, dropping the future invokes Pi's own + // process-tree guard. Never wait indefinitely for that cooperation. + execution_cancel.cancel(); + let _ = tokio::time::timeout(CLEANUP_GRACE, &mut execution).await; + Err(reason) + } +} + +#[cfg(test)] +mod tests; diff --git a/apps/maple-agent/crates/maple-agent/src/agent/tools/shell/tests.rs b/apps/maple-agent/crates/maple-agent/src/agent/tools/shell/tests.rs new file mode 100644 index 000000000..0957e5d01 --- /dev/null +++ b/apps/maple-agent/crates/maple-agent/src/agent/tools/shell/tests.rs @@ -0,0 +1,270 @@ +use super::*; +use std::fs; +use std::sync::atomic::{AtomicBool, Ordering}; +use tokio_util::sync::CancellationToken; + +fn options() -> ExecOptions { + ExecOptions { + on_data: Arc::new(|_| {}), + cancel: CancellationToken::new(), + timeout: None, + env: None, + } +} + +#[tokio::test] +async fn unavailable_or_relative_folders_fail_before_execution() { + let dir = tempfile::tempdir().unwrap(); + fs::write(dir.path().join("file"), "fixture").unwrap(); + let shell = bounded(LocalShellOperations::bash(None)); + for cwd in [ + dir.path().join("missing"), + dir.path().join("file"), + "../../".into(), + ] { + assert!(matches!( + shell.exec("must-not-execute", &cwd, options()).await, + Err(ExecError::Failed(_)) + )); + } +} + +struct InspectTimeout; +#[async_trait] +impl BashOperations for InspectTimeout { + async fn exec( + &self, + _: &str, + _: &Path, + options: ExecOptions, + ) -> Result, ExecError> { + assert_eq!(options.timeout, Some(DEFAULT_TIMEOUT_SECONDS)); + Ok(Some(0)) + } +} + +#[tokio::test] +async fn maple_supplies_a_default_deadline_to_the_existing_operations_interface() { + let dir = tempfile::tempdir().unwrap(); + let shell = BoundedShell { + inner: Arc::new(InspectTimeout), + default_timeout: DEFAULT_TIMEOUT_SECONDS, + }; + assert_eq!( + shell.exec("fixture", dir.path(), options()).await.unwrap(), + Some(0) + ); +} + +#[cfg(unix)] +#[tokio::test] +async fn explicit_deadlines_override_the_default_and_calls_keep_the_saved_folder() { + let dir = tempfile::tempdir().unwrap(); + fs::write(dir.path().join("window.rs"), "fixture").unwrap(); + let shell = BoundedShell { + inner: Arc::new(LocalShellOperations::bash(None)), + default_timeout: 0.05, + }; + assert_eq!( + shell.exec("sleep 30", dir.path(), options()).await, + Err(ExecError::TimedOut(0.05)) + ); + assert_eq!( + shell + .exec( + "sleep 0.1", + dir.path(), + ExecOptions { + timeout: Some(5.0), + ..options() + } + ) + .await + .unwrap(), + Some(0) + ); + shell.exec("cd ..", dir.path(), options()).await.unwrap(); + assert_eq!( + shell + .exec("test -f ./window.rs", dir.path(), options()) + .await + .unwrap(), + Some(0) + ); +} + +struct BlockedDelivery(Arc, Arc); +struct Dropped(Arc); +impl Drop for Dropped { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } +} +#[async_trait] +impl BashOperations for BlockedDelivery { + async fn exec(&self, _: &str, _: &Path, _: ExecOptions) -> Result, ExecError> { + let _dropped = Dropped(self.0.clone()); + self.1.notify_one(); + // An operation stuck before its own cancellation/deadline select, as + // with command delivery to a shell that does not read its stdin. + std::future::pending().await + } +} + +#[tokio::test] +async fn uncooperative_command_delivery_is_bounded_and_dropped() { + let dir = tempfile::tempdir().unwrap(); + let dropped = Arc::new(AtomicBool::new(false)); + let shell = BoundedShell { + inner: Arc::new(BlockedDelivery(dropped.clone(), Arc::default())), + default_timeout: 0.01, + }; + let parent_cancel = CancellationToken::new(); + let result = tokio::time::timeout( + Duration::from_secs(5), + shell.exec( + "fixture", + dir.path(), + ExecOptions { + cancel: parent_cancel.clone(), + ..options() + }, + ), + ) + .await + .unwrap(); + assert_eq!(result, Err(ExecError::TimedOut(0.01))); + assert!(dropped.load(Ordering::SeqCst)); + assert!( + !parent_cancel.is_cancelled(), + "a command deadline must not cancel its caller" + ); +} + +#[tokio::test] +async fn uncooperative_command_delivery_is_cancelled_and_dropped() { + let dir = tempfile::tempdir().unwrap(); + let dropped = Arc::new(AtomicBool::new(false)); + let entered = Arc::new(tokio::sync::Notify::new()); + let shell = BoundedShell { + inner: Arc::new(BlockedDelivery(dropped.clone(), entered.clone())), + default_timeout: DEFAULT_TIMEOUT_SECONDS, + }; + let cancel = CancellationToken::new(); + let (result, ()) = tokio::time::timeout(Duration::from_secs(5), async { + tokio::join!( + shell.exec( + "fixture", + dir.path(), + ExecOptions { + cancel: cancel.clone(), + ..options() + } + ), + async { + entered.notified().await; + cancel.cancel(); + } + ) + }) + .await + .unwrap(); + assert_eq!(result, Err(ExecError::Aborted)); + assert!(dropped.load(Ordering::SeqCst)); +} + +/// A finite, disposable substitute for a kernel file that waits indefinitely for +/// new data. The held writer prevents EOF; neither side of the pipeline produces +/// output, so head cannot bound its runtime. +#[cfg(unix)] +#[tokio::test] +async fn blocked_pipeline_is_stopped_by_cancel_deadline_and_dropped_future() { + use std::ffi::CString; + use std::os::unix::fs::OpenOptionsExt; + + for end in ["cancel", "deadline", "drop"] { + let dir = tempfile::tempdir().unwrap(); + fs::create_dir_all(dir.path().join("scratch/tmp")).unwrap(); + let fifo = CString::new(dir.path().join("blocked").as_os_str().as_encoded_bytes()).unwrap(); + // SAFETY: a valid, NUL-terminated path inside a disposable test directory. + assert_eq!(unsafe { libc::mkfifo(fifo.as_ptr(), 0o600) }, 0); + let _reader = fs::OpenOptions::new() + .read(true) + .custom_flags(libc::O_NONBLOCK) + .open(dir.path().join("blocked")) + .unwrap(); + let _writer = fs::OpenOptions::new() + .write(true) + .custom_flags(libc::O_NONBLOCK) + .open(dir.path().join("blocked")) + .unwrap(); + let tool = Arc::new(BoundedShell { + inner: Arc::new(LocalShellOperations::bash(None)), + default_timeout: DEFAULT_TIMEOUT_SECONDS, + }); + let cwd = dir.path().to_path_buf(); + let cancel = CancellationToken::new(); + let call = tokio::spawn({ + let (tool, cancel) = (tool.clone(), cancel.clone()); + async move { + tool.exec( + "printf '%s' $$ > shell.pid; cd scratch/tmp && sh -c 'printf %s $$ > ../../reader.pid; exec cat ../../blocked' | sh -c 'printf %s $$ > ../../head.pid; exec head -c 1'", + &cwd, + ExecOptions { timeout: Some(if end == "deadline" { 0.5 } else { 30.0 }), cancel, ..options() }, + ).await + } + }); + let mut pids = Vec::new(); + for name in ["shell.pid", "reader.pid", "head.pid"] { + let pid = tokio::time::timeout(Duration::from_secs(5), async { + loop { + if let Some(pid) = fs::read_to_string(dir.path().join(name)) + .ok() + .and_then(|text| text.parse::().ok()) + { + break pid; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("pipeline member records its pid"); + pids.push(pid); + } + match end { + "cancel" => cancel.cancel(), + "drop" => call.abort(), + _ => {} + } + let result = tokio::time::timeout(Duration::from_secs(5), call) + .await + .expect("blocked call ends"); + match end { + "drop" => assert!(result.unwrap_err().is_cancelled()), + "cancel" => assert_eq!(result.unwrap().unwrap_err(), ExecError::Aborted), + _ => assert_eq!(result.unwrap().unwrap_err(), ExecError::TimedOut(0.5)), + } + // Check every member, not just the shell. Linux PID 1 need not reap an + // orphan immediately, so a zombie counts as terminated, never as alive. + for pid in pids { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + // SAFETY: signal zero only checks existence. + if unsafe { libc::kill(pid, 0) } != 0 { + break; + } + #[cfg(target_os = "linux")] + if fs::read_to_string(format!("/proc/{pid}/stat")).is_ok_and(|stat| { + stat.rsplit_once(") ") + .is_some_and(|(_, rest)| rest.starts_with("Z ")) + }) { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("pipeline member is terminated"); + } + } +} diff --git a/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs b/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs index 9efdb1862..0434c3f47 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/tools/tests.rs @@ -128,8 +128,8 @@ fn the_model_gets_read_a_shell_edit_and_write() { }; assert_eq!(tools.builtin, ["read", shell, "edit", "write"]); assert!(tools.options.powershell.spawn_hook.is_some()); - assert_eq!(tools.options.bash.default_timeout, Some(1800.0)); - assert_eq!(tools.options.powershell.default_timeout, Some(1800.0)); + assert!(tools.options.bash.operations.is_some()); + assert!(tools.options.powershell.operations.is_some()); } fn tool(tools: &TaskTools, name: &str) -> Arc { diff --git a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs index 08ab605e1..140357f1a 100644 --- a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs +++ b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash.rs @@ -2,15 +2,15 @@ //! //! Output streams to the interface as it arrives. The model gets the last 2000 lines or //! 50KB, whichever is less, and when that cuts anything the full output is saved to a -//! file it can read. The embedding host can choose a default timeout. Stop or a -//! timeout ends the command's process group (its process tree on Windows). +//! file it can read. There is no default timeout; Stop or a timeout ends the command +//! and everything it started. use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::process::Stdio; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; -use std::time::{Duration, Instant}; +use std::time::Duration; use async_trait::async_trait; use pi_agent_core::{AgentTool, AgentToolResult, ToolError, ToolInvocation, ToolUpdates}; @@ -33,8 +33,6 @@ use super::{PREFER_STRICT, ToolContext}; const MAX_TIMEOUT_SECONDS: f64 = 2_147_483_647.0 / 1000.0; /// How often output updates reach the interface at most. const UPDATE_THROTTLE: Duration = Duration::from_millis(100); -/// Quiet commands still report their execution status to the interface. -const STATUS_INTERVAL: Duration = Duration::from_secs(5); /// After the shell exits, how long its output may stay quiet before collection stops. /// A background process can keep the output open; output still arriving keeps it open. const EXIT_STDIO_GRACE: Duration = Duration::from_millis(100); @@ -84,8 +82,6 @@ pub trait BashOperations: Send + Sync { /// Run `command` in `cwd`, streaming its output. The exit code is `None` when the /// command ended without one; a command killed by a signal reports 128 plus the /// signal's number. - /// A selected working directory that is unavailable must fail, never fall back - /// to the host's working directory or a temporary folder. async fn exec( &self, command: &str, @@ -101,7 +97,7 @@ enum LocalShell { } /// Commands on this machine, in their own process group (a process tree on Windows) -/// so stopping one reaches the group's members, including pipeline processes. +/// so that stopping one reaches everything it started. #[derive(Clone, Debug)] pub struct LocalShellOperations { shell: LocalShell, @@ -146,15 +142,7 @@ impl BashOperations for LocalShellOperations { format!("{POWERSHELL_UTF8_PREFIX}{command}"), ), }; - if !cwd.is_absolute() { - return Err(ExecError::Failed( - "Working directory must be an absolute path".into(), - )); - } - if !tokio::fs::metadata(cwd) - .await - .is_ok_and(|metadata| metadata.is_dir()) - { + if tokio::fs::metadata(cwd).await.is_err() { return Err(ExecError::Failed(format!( "Working directory does not exist: {}\nCannot execute {shell_name} commands.", cwd.display() @@ -252,7 +240,10 @@ async fn run_local( if let Some(stderr) = child.stderr.take() { readers.spawn(pump(stderr, options.on_data.clone(), activity.clone())); } - let stdin = child.stdin.take(); + if let Some(mut stdin) = child.stdin.take() { + let _ = stdin.write_all(command.as_bytes()).await; + drop(stdin); + } let kill = |pid: Option| { if let Some(pid) = pid { @@ -268,14 +259,7 @@ async fn run_local( } }; tokio::pin!(deadline); - // Writing a large command to a stdin-based shell can block too. Keep that - // write inside the same cancellation/deadline boundary as waiting for exit. - let wait = async { - if let Some(mut stdin) = stdin { - let _ = stdin.write_all(command.as_bytes()).await; - } - child.wait().await - }; + let wait = child.wait(); tokio::pin!(wait); let status = loop { tokio::select! { @@ -378,9 +362,6 @@ async fn pump(mut stream: R, on_data: OnData, activity: Ar /// Options for `bash`. #[derive(Clone)] pub struct BashToolOptions { - /// Seconds before a command without an explicit timeout is stopped. - /// Standalone Pi has no default; an embedding host can choose one. - pub default_timeout: Option, /// Full output goes here when present, without falling back on a failed path. pub output_dir: Option, /// Where commands run; this machine when `None`. @@ -398,7 +379,6 @@ pub struct BashToolOptions { impl Default for BashToolOptions { fn default() -> Self { Self { - default_timeout: None, output_dir: None, operations: None, command_prefix: None, @@ -412,7 +392,6 @@ impl Default for BashToolOptions { /// Options for `powershell`. #[derive(Clone)] pub struct PowerShellToolOptions { - pub default_timeout: Option, pub output_dir: Option, pub operations: Option>, pub expose_session_environment: bool, @@ -422,7 +401,6 @@ pub struct PowerShellToolOptions { impl Default for PowerShellToolOptions { fn default() -> Self { Self { - default_timeout: None, output_dir: None, operations: None, expose_session_environment: true, @@ -441,7 +419,6 @@ pub struct ShellTool { command_prefix: Option, expose_session_environment: bool, spawn_hook: Option, - default_timeout: Option, context: ToolContext, } @@ -449,32 +426,23 @@ pub struct ShellTool { struct ShellParams { command: String, #[serde(default)] - workdir: Option, - #[serde(default)] timeout: Option, } -fn shell_declaration(name: &str, shell_name: &str, default_timeout: Option) -> Tool { - let timeout_description = match default_timeout { - Some(seconds) => format!( - "Timeout in seconds (optional, defaults to {seconds} seconds). Set explicitly for longer commands." - ), - None => "Timeout in seconds (optional, no default timeout)".to_string(), - }; +fn shell_declaration(name: &str, shell_name: &str) -> Tool { Tool::new( name, format!( - "Execute a {shell_name} command in the session working directory, or an explicit absolute workdir. Each call starts in that folder; cd does not change later calls. Returns stdout and stderr. Output is truncated to last {DEFAULT_MAX_LINES} lines or {}KB (whichever is hit first). If truncated, full output is saved to a temp file. {timeout_description}", + "Execute a {shell_name} command in the current working directory. Returns stdout and stderr. Output is truncated to last {DEFAULT_MAX_LINES} lines or {}KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.", DEFAULT_MAX_BYTES / 1024 ), json!({ "type": "object", "properties": { "command": { "type": "string", "description": "Shell command to execute" }, - "workdir": { "type": "string", "description": "Absolute working directory on the execution computer. For repository searches, set this to the intended checkout or worktree." }, "timeout": { "type": "number", - "description": timeout_description + "description": "Timeout in seconds (optional, no default timeout)" } }, "required": ["command"] @@ -489,7 +457,7 @@ impl ShellTool { .operations .unwrap_or_else(|| Arc::new(LocalShellOperations::bash(options.shell_path))); Self { - declaration: shell_declaration("bash", "bash", options.default_timeout), + declaration: shell_declaration("bash", "bash"), name: "bash", cwd: cwd.into(), output_dir: options.output_dir, @@ -497,7 +465,6 @@ impl ShellTool { command_prefix: options.command_prefix, expose_session_environment: options.expose_session_environment, spawn_hook: options.spawn_hook, - default_timeout: options.default_timeout, context, } } @@ -511,7 +478,7 @@ impl ShellTool { .operations .unwrap_or_else(|| Arc::new(LocalShellOperations::powershell())); Self { - declaration: shell_declaration("powershell", "PowerShell", options.default_timeout), + declaration: shell_declaration("powershell", "PowerShell"), name: "powershell", cwd: cwd.into(), output_dir: options.output_dir, @@ -519,23 +486,20 @@ impl ShellTool { command_prefix: None, expose_session_environment: options.expose_session_environment, spawn_hook: options.spawn_hook, - default_timeout: options.default_timeout, context, } } /// The guideline that goes with the tool's snippet. pub fn guidelines(&self) -> Vec { - let mut rules = vec![ - "For repository searches, set workdir to the absolute intended checkout/worktree and search within it. Verify the folder before using parent paths; never guess ../../ from a temporary folder. Avoid recursive searches of filesystem roots or virtual filesystems such as /proc, /sys and /dev. head limits output, not runtime; request a short timeout (normally 30 seconds) for searches and narrow the scope if it expires.".to_string(), - ]; if self.expose_session_environment { - rules.push(format!( + vec![format!( "You can inspect {}_* environment variables for current model and session details.", self.context.env_prefix() - )); + )] + } else { + Vec::new() } - rules } fn spawn_context(&self, command: String, cwd: PathBuf) -> BashSpawnContext { @@ -581,50 +545,23 @@ struct UpdateThrottle { task: tokio::task::JoinHandle<()>, output: Arc>, updates: ToolUpdates, - progress: Arc, -} - -struct ShellProgress { - cwd: PathBuf, - timeout: Option, - started: Instant, - last_output: Mutex, } impl UpdateThrottle { - fn start( - output: Arc>, - updates: ToolUpdates, - cwd: PathBuf, - timeout: Option, - ) -> Self { - let progress = Arc::new(ShellProgress { - cwd, - timeout, - started: Instant::now(), - last_output: Mutex::new(Instant::now()), + fn start(output: Arc>, updates: ToolUpdates) -> Self { + updates.send(AgentToolResult { + content: Vec::new(), + ..AgentToolResult::default() }); - let dirty = Arc::new(AtomicBool::new(true)); - send_update(&dirty, &output, &updates, &progress); + let dirty = Arc::new(AtomicBool::new(false)); let wake = Arc::new(Notify::new()); let task = tokio::spawn({ - let (dirty, wake, output, updates, progress) = ( - dirty.clone(), - wake.clone(), - output.clone(), - updates.clone(), - progress.clone(), - ); + let (dirty, wake, output, updates) = + (dirty.clone(), wake.clone(), output.clone(), updates.clone()); async move { - let mut status = tokio::time::interval(STATUS_INTERVAL); - status.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - status.tick().await; loop { - tokio::select! { - _ = wake.notified() => {}, - _ = status.tick() => dirty.store(true, Ordering::Release), - } - send_update(&dirty, &output, &updates, &progress); + wake.notified().await; + send_update(&dirty, &output, &updates); tokio::time::sleep(UPDATE_THROTTLE).await; } } @@ -635,41 +572,22 @@ impl UpdateThrottle { task, output, updates, - progress, } } fn mark(&self) { - *self - .progress - .last_output - .lock() - .unwrap_or_else(|error| error.into_inner()) = Instant::now(); self.dirty.store(true, Ordering::Release); self.wake.notify_one(); } - /// Join the reporter before the tool returns: abort alone cannot stop a - /// synchronous send already running on another executor thread. - async fn finish(mut self) { - self.task.abort(); - let _ = (&mut self.task).await; - send_update(&self.dirty, &self.output, &self.updates, &self.progress); - } -} - -impl Drop for UpdateThrottle { - fn drop(&mut self) { + /// Stop, sending whatever has not been sent. + fn finish(self) { self.task.abort(); + send_update(&self.dirty, &self.output, &self.updates); } } -fn send_update( - dirty: &AtomicBool, - output: &Mutex, - updates: &ToolUpdates, - progress: &ShellProgress, -) { +fn send_update(dirty: &AtomicBool, output: &Mutex, updates: &ToolUpdates) { if !dirty.swap(false, Ordering::AcqRel) { return; } @@ -677,16 +595,9 @@ fn send_update( .lock() .unwrap_or_else(|error| error.into_inner()) .snapshot(true); - let mut details = snapshot_details(&snapshot).unwrap_or_else(|| json!({})); - details["shellExecution"] = json!({ - "cwd": progress.cwd.to_string_lossy(), - "timeoutSeconds": progress.timeout, - "elapsedSeconds": progress.started.elapsed().as_secs(), - "outputIdleSeconds": progress.last_output.lock().unwrap_or_else(|error| error.into_inner()).elapsed().as_secs(), - }); updates.send(AgentToolResult { content: vec![Content::text(snapshot.content.clone())], - details: Some(details), + details: snapshot_details(&snapshot), ..AgentToolResult::default() }); } @@ -762,18 +673,7 @@ impl AgentTool for ShellTool { Some(prefix) => format!("{prefix}\n{}", params.command), None => params.command, }; - let timeout = params.timeout.or(self.default_timeout); - resolve_timeout(timeout)?; - if params - .workdir - .as_ref() - .is_some_and(|cwd| !cwd.is_absolute()) - { - return Err("Working directory must be an absolute path".into()); - } - let cwd = params - .workdir - .unwrap_or_else(|| self.context.cwd().unwrap_or_else(|| self.cwd.clone())); + let cwd = self.context.cwd().unwrap_or_else(|| self.cwd.clone()); let spawn = self.spawn_context(command, cwd); let output = Arc::new(Mutex::new(OutputAccumulator::new_with_output_dir( @@ -783,8 +683,6 @@ impl AgentTool for ShellTool { let throttle = Arc::new(Mutex::new(Some(UpdateThrottle::start( output.clone(), invocation.updates.clone(), - spawn.cwd.clone(), - timeout, )))); let on_data: OnData = { let (output, throttle) = (output.clone(), throttle.clone()); @@ -810,7 +708,7 @@ impl AgentTool for ShellTool { ExecOptions { on_data, cancel: invocation.cancel.clone(), - timeout, + timeout: params.timeout, env: Some(spawn.env), }, ) @@ -821,12 +719,12 @@ impl AgentTool for ShellTool { output.finish(); (output.snapshot(true), output.last_line_bytes()) }; - let finished_throttle = throttle + if let Some(throttle) = throttle .lock() .unwrap_or_else(|error| error.into_inner()) - .take(); - if let Some(throttle) = finished_throttle { - throttle.finish().await; + .take() + { + throttle.finish(); } // The kept output reaches the model either way; a failed save loses only the // rest, and the note says the file could not be saved. diff --git a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs index 7f5e1641b..b84184474 100644 --- a/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs +++ b/apps/maple-agent/crates/pi-coding-agent/src/tools/bash/tests.rs @@ -73,45 +73,6 @@ fn full_output_path(text: &str) -> PathBuf { PathBuf::from(&text[start..end]) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn finishing_progress_waits_for_an_in_flight_reporter() { - // Model a reporter inside synchronous send_update on another worker. Abort - // cannot interrupt that code; the final result must wait until it returns. - let gate = Arc::new(std::sync::Barrier::new(2)); - let reported = Arc::new(AtomicBool::new(false)); - let (started, ready) = tokio::sync::oneshot::channel(); - let task = tokio::spawn({ - let (gate, reported) = (gate.clone(), reported.clone()); - async move { - started.send(()).unwrap(); - gate.wait(); - reported.store(true, Ordering::Release); - } - }); - ready.await.unwrap(); - let throttle = UpdateThrottle { - dirty: Arc::new(AtomicBool::new(false)), - wake: Arc::new(Notify::new()), - task, - output: Arc::new(Mutex::new(OutputAccumulator::new("progress-fixture"))), - updates: ToolUpdates::none(), - progress: Arc::new(ShellProgress { - cwd: PathBuf::new(), - timeout: None, - started: Instant::now(), - last_output: Mutex::new(Instant::now()), - }), - }; - let finish = throttle.finish(); - tokio::pin!(finish); - let first_poll = futures_util::poll!(&mut finish); - // Always release the worker before asserting, including on a regression. - gate.wait(); - assert!(first_poll.is_pending(), "completion must join the reporter"); - finish.await; - assert!(reported.load(Ordering::Acquire)); -} - #[cfg(unix)] #[tokio::test] async fn commands_run_and_report_their_exit_code() { @@ -186,208 +147,6 @@ async fn a_timeout_stops_the_command() { assert_eq!(error, "Invalid timeout: must be a finite number of seconds"); } -#[cfg(unix)] -#[tokio::test] -async fn each_call_uses_the_explicit_worktree_or_the_session_folder() { - let fixture = tempfile::tempdir().unwrap(); - let project = fixture.path().join("checkout"); - let worktree = fixture.path().join("linked-worktree"); - fs::create_dir(&project).unwrap(); - fs::create_dir(&worktree).unwrap(); - fs::write(project.join("window.rs"), "checkout").unwrap(); - fs::write(worktree.join("window.rs"), "worktree").unwrap(); - let tool = bash(&project, BashToolOptions::default()); - let result = run( - &tool, - json!({"command": "cat ./window.rs", "workdir": worktree}), - ) - .await - .unwrap(); - assert_eq!(content_text(&result.content), "worktree"); - run(&tool, json!({"command": "cd .."})).await.unwrap(); - let result = run(&tool, json!({"command": "cat ./window.rs"})) - .await - .unwrap(); - assert_eq!(content_text(&result.content), "checkout"); - - for workdir in [fixture.path().join("missing"), project.join("window.rs")] { - let error = run( - &tool, - json!({"command": "touch must-not-run", "workdir": workdir}), - ) - .await - .unwrap_err(); - assert!( - error.starts_with("Working directory does not exist"), - "{error}" - ); - } - let error = run( - &tool, - json!({"command": "touch must-not-run", "workdir": "../../"}), - ) - .await - .unwrap_err(); - assert_eq!(error, "Working directory must be an absolute path"); - let relative = bash(Path::new("."), BashToolOptions::default()); - assert_eq!( - run(&relative, json!({"command": "pwd"})).await.unwrap_err(), - "Working directory must be an absolute path" - ); - assert!(!project.join("must-not-run").exists()); -} - -#[cfg(unix)] -#[tokio::test] -async fn the_host_default_deadline_is_used_unless_the_call_overrides_it() { - let dir = tempfile::tempdir().unwrap(); - let tool = bash( - dir.path(), - BashToolOptions { - default_timeout: Some(0.05), - ..BashToolOptions::default() - }, - ); - let error = tokio::time::timeout( - Duration::from_secs(5), - run(&tool, json!({"command": "sleep 30"})), - ) - .await - .expect("default deadline ends the call") - .unwrap_err(); - assert_eq!(error, "Command timed out after 0.05 seconds"); - let result = run( - &tool, - json!({"command": "sleep 0.1; printf done", "timeout": 5}), - ) - .await - .unwrap(); - assert_eq!(content_text(&result.content), "done"); -} - -/// A finite, disposable substitute for a kernel file that waits indefinitely for -/// new data. The held writer prevents EOF; neither side of the pipeline produces -/// output, so head cannot bound its runtime. -#[cfg(unix)] -#[tokio::test] -async fn blocked_pipeline_is_stopped_by_cancel_deadline_and_dropped_future() { - use std::ffi::CString; - use std::os::unix::fs::OpenOptionsExt; - - for end in ["cancel", "deadline", "drop"] { - let dir = tempfile::tempdir().unwrap(); - fs::create_dir_all(dir.path().join("scratch/tmp")).unwrap(); - let fifo = CString::new(dir.path().join("blocked").as_os_str().as_encoded_bytes()).unwrap(); - // SAFETY: a valid, NUL-terminated path inside a disposable test directory. - assert_eq!(unsafe { libc::mkfifo(fifo.as_ptr(), 0o600) }, 0); - let _reader = fs::OpenOptions::new() - .read(true) - .custom_flags(libc::O_NONBLOCK) - .open(dir.path().join("blocked")) - .unwrap(); - let _writer = fs::OpenOptions::new() - .write(true) - .custom_flags(libc::O_NONBLOCK) - .open(dir.path().join("blocked")) - .unwrap(); - let tool = Arc::new(bash(dir.path(), BashToolOptions::default())); - let cancel = CancellationToken::new(); - let call = tokio::spawn({ - let (tool, cancel) = (tool.clone(), cancel.clone()); - async move { - run_with(&tool, json!({ - "command": "printf '%s' $$ > shell.pid; cd scratch/tmp && sh -c 'printf %s $$ > ../../reader.pid; exec cat ../../blocked' | sh -c 'printf %s $$ > ../../head.pid; exec head -c 1'", - "timeout": if end == "deadline" { 0.5 } else { 30.0 }, - }), cancel).await - } - }); - let mut pids = Vec::new(); - for name in ["shell.pid", "reader.pid", "head.pid"] { - let pid = tokio::time::timeout(Duration::from_secs(5), async { - loop { - if let Some(pid) = fs::read_to_string(dir.path().join(name)) - .ok() - .and_then(|text| text.parse::().ok()) - { - break pid; - } - tokio::time::sleep(Duration::from_millis(5)).await; - } - }) - .await - .expect("pipeline member records its pid"); - pids.push(pid); - } - match end { - "cancel" => cancel.cancel(), - "drop" => call.abort(), - _ => {} - } - let result = tokio::time::timeout(Duration::from_secs(5), call) - .await - .expect("blocked call ends"); - match end { - "drop" => assert!(result.unwrap_err().is_cancelled()), - "cancel" => assert_eq!(result.unwrap().unwrap_err(), "Command aborted"), - _ => assert_eq!( - result.unwrap().unwrap_err(), - "Command timed out after 0.5 seconds" - ), - } - // Check every member, not just the shell. Linux PID 1 need not reap an - // orphan immediately, so a zombie counts as terminated, never as alive. - for pid in pids { - tokio::time::timeout(Duration::from_secs(5), async { - loop { - // SAFETY: signal zero only checks existence. - if unsafe { libc::kill(pid, 0) } != 0 { - break; - } - #[cfg(target_os = "linux")] - if fs::read_to_string(format!("/proc/{pid}/stat")).is_ok_and(|stat| { - stat.rsplit_once(") ") - .is_some_and(|(_, rest)| rest.starts_with("Z ")) - }) { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - }) - .await - .expect("pipeline member is terminated"); - } - } -} - -#[cfg(unix)] -#[tokio::test] -async fn deadline_also_covers_a_shell_that_does_not_read_its_command_from_stdin() { - let dir = tempfile::tempdir().unwrap(); - let config = ShellConfig { - shell: PathBuf::from("/bin/bash"), - args: vec!["-c".into(), "sleep 30".into()], - transport: CommandTransport::Stdin, - }; - let result = tokio::time::timeout( - Duration::from_secs(5), - run_local( - &config, - &"x".repeat(1024 * 1024), - dir.path(), - Some(Duration::from_millis(100)), - ExecOptions { - on_data: Arc::new(|_| {}), - cancel: CancellationToken::new(), - timeout: Some(0.1), - env: None, - }, - ), - ) - .await - .expect("deadline covers blocked stdin write"); - assert_eq!(result, Err(ExecError::TimedOut(0.1))); -} - #[tokio::test] async fn a_stopped_or_timed_out_command_points_at_its_full_output() { let dir = tempfile::tempdir().unwrap(); diff --git a/apps/maple-agent/crates/pi-coding-agent/src/tools/tests.rs b/apps/maple-agent/crates/pi-coding-agent/src/tools/tests.rs index 7cba1ea5e..d043aa03f 100644 --- a/apps/maple-agent/crates/pi-coding-agent/src/tools/tests.rs +++ b/apps/maple-agent/crates/pi-coding-agent/src/tools/tests.rs @@ -616,7 +616,7 @@ fn every_built_in_tool_is_created_with_the_defaults_active() { ); assert_eq!( bash.guidelines, - ShellTool::bash(dir.path(), BashToolOptions::default(), context()).guidelines() + ["You can inspect TEST_* environment variables for current model and session details."] ); }