Skip to main content

pi/core/tools/
bash.rs

1//! Bash tool: execute a shell command with streamed, tail-truncated output.
2//!
3//! Ports `.references/pi/packages/coding-agent/src/core/tools/bash.ts` plus the
4//! local process-tree kill path from `utils/shell.ts`. stdout and stderr are
5//! merged by arrival order into an [`OutputAccumulator`] with spill prefix
6//! `pi-bash`. Partial tool updates are throttled to 100 ms.
7
8use std::collections::HashMap;
9use std::fmt::Write as _;
10use std::path::{Path, PathBuf};
11use std::process::Stdio;
12use std::sync::Arc;
13use std::time::{Duration, Instant};
14
15use futures::FutureExt as _;
16use futures::future::BoxFuture;
17use pi_agent::{AgentTool, AgentToolResult, ToolError, ToolUpdates};
18use pi_ai::ToolResultContent;
19use pi_ai::types::TextContent;
20use schemars::JsonSchema;
21use serde::{Deserialize, Serialize};
22use serde_json::{Map, Value};
23use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
24use tokio::process::{Child, Command};
25use tokio::sync::Mutex;
26use tokio::sync::mpsc;
27use tokio_util::sync::CancellationToken;
28
29use super::{
30    DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES, OutputAccumulator, OutputAccumulatorError,
31    OutputAccumulatorOptions, OutputSnapshot, TruncatedBy, TruncationResult, format_size,
32};
33
34/// Maximum timeout duration in milliseconds (TypeScript `MAX_TIMEOUT_MS`).
35const MAX_TIMEOUT_MS: u64 = 2_147_483_647;
36
37/// Numeric form of the maximum timeout in seconds.
38const MAX_TIMEOUT_SECONDS: f64 = 2_147_483.647;
39
40/// Display form of the maximum timeout in seconds (`MAX_TIMEOUT_MS / 1000`).
41const MAX_TIMEOUT_SECONDS_DISPLAY: &str = "2147483.647";
42
43/// Throttle window for streaming tool updates.
44const BASH_UPDATE_THROTTLE: Duration = Duration::from_millis(100);
45
46/// Temp-file prefix for bash spill paths (`pi-bash-{16hex}.log`).
47const BASH_TEMP_FILE_PREFIX: &str = "pi-bash";
48
49/// TypeBox-compatible bash arguments (fixture `bash.json`).
50#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
51pub struct BashToolInput {
52    /// Bash command to execute.
53    #[schemars(description = "Bash command to execute")]
54    pub command: String,
55    /// Timeout in seconds (optional, no default timeout).
56    #[serde(default, skip_serializing_if = "Option::is_none")]
57    #[schemars(description = "Timeout in seconds (optional, no default timeout)")]
58    pub timeout: Option<f64>,
59}
60
61/// Structured details returned when bash output is truncated.
62#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
63#[serde(rename_all = "camelCase")]
64pub struct BashToolDetails {
65    /// Tail truncation metadata when the stream exceeded limits.
66    #[serde(default, skip_serializing_if = "Option::is_none")]
67    pub truncation: Option<TruncationResult>,
68    /// Absolute path of the full-output spill file, when one was opened.
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub full_output_path: Option<String>,
71}
72
73/// Pluggable command execution backend (TypeScript `BashOperations`).
74pub trait BashOperations: Send + Sync {
75    /// Execute `command` in `cwd`, streaming merged output chunks through
76    /// `on_data`. Returns the process exit code (`None` when killed without a
77    /// status). Failures use the TypeScript internal markers `"aborted"` and
78    /// `"timeout:{secs}"` so the tool wrapper can append status text.
79    fn exec(
80        &self,
81        command: String,
82        cwd: PathBuf,
83        on_data: Box<dyn FnMut(Vec<u8>) + Send>,
84        cancel: CancellationToken,
85        timeout: Option<f64>,
86        env: HashMap<String, String>,
87    ) -> BoxFuture<'static, Result<Option<i32>, ToolError>>;
88}
89
90/// Spawn rewrite hook context (TypeScript `BashSpawnContext`).
91#[derive(Clone, Debug)]
92pub struct BashSpawnContext {
93    /// Command string to execute.
94    pub command: String,
95    /// Working directory for the child process.
96    pub cwd: PathBuf,
97    /// Environment map for the child process.
98    pub env: HashMap<String, String>,
99}
100
101/// Optional rewrite applied before local spawn (TypeScript `BashSpawnHook`).
102pub type BashSpawnHook = Arc<dyn Fn(BashSpawnContext) -> BashSpawnContext + Send + Sync>;
103
104/// Options for [`BashTool`].
105#[derive(Clone)]
106pub struct BashToolOptions {
107    /// Working directory used when no spawn hook rewrites it.
108    pub cwd: PathBuf,
109    /// Optional absolute shell path (TypeScript `shellPath`).
110    pub shell_path: Option<PathBuf>,
111    /// Optional command prefix prepended as `{prefix}\n{command}`.
112    pub command_prefix: Option<String>,
113    /// Optional spawn rewrite hook.
114    pub spawn_hook: Option<BashSpawnHook>,
115    /// Custom operations; default is local shell execution.
116    pub operations: Option<Arc<dyn BashOperations>>,
117}
118
119impl std::fmt::Debug for BashToolOptions {
120    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
121        f.debug_struct("BashToolOptions")
122            .field("cwd", &self.cwd)
123            .field("shell_path", &self.shell_path)
124            .field("command_prefix", &self.command_prefix)
125            .field("spawn_hook", &self.spawn_hook.as_ref().map(|_| "Some(..)"))
126            .field("operations", &self.operations.as_ref().map(|_| "Some(..)"))
127            .finish()
128    }
129}
130
131impl BashToolOptions {
132    /// Builds options for `cwd`.
133    #[must_use]
134    pub fn new(cwd: impl Into<PathBuf>) -> Self {
135        Self {
136            cwd: cwd.into(),
137            shell_path: None,
138            command_prefix: None,
139            spawn_hook: None,
140            operations: None,
141        }
142    }
143}
144
145/// Local shell backend matching TypeScript `createLocalBashOperations`.
146#[derive(Clone, Debug, Default)]
147pub struct LocalBashOperations {
148    shell_path: Option<PathBuf>,
149}
150
151impl LocalBashOperations {
152    /// Creates local operations with optional custom shell path.
153    #[must_use]
154    pub fn new(shell_path: Option<PathBuf>) -> Self {
155        Self { shell_path }
156    }
157}
158
159impl BashOperations for LocalBashOperations {
160    fn exec(
161        &self,
162        command: String,
163        cwd: PathBuf,
164        mut on_data: Box<dyn FnMut(Vec<u8>) + Send>,
165        cancel: CancellationToken,
166        timeout: Option<f64>,
167        env: HashMap<String, String>,
168    ) -> BoxFuture<'static, Result<Option<i32>, ToolError>> {
169        let shell_path = self.shell_path.clone();
170        async move {
171            let timeout_ms = resolve_timeout_ms(timeout)?;
172            if cancel.is_cancelled() {
173                return Err(ToolError::new("aborted"));
174            }
175            if !cwd.is_dir() {
176                return Err(ToolError::new(format!(
177                    "Working directory does not exist: {}\nCannot execute bash commands.",
178                    cwd.display()
179                )));
180            }
181
182            let shell = resolve_shell_config(shell_path.as_deref())?;
183            let mut child = spawn_shell_command(&shell, &command, &cwd, &env)?;
184            let pid = child.id();
185            let stdout = child.stdout.take();
186            let stderr = child.stderr.take();
187
188            let (chunk_tx, mut chunk_rx) = mpsc::unbounded_channel::<Vec<u8>>();
189            let readers = spawn_stream_readers(stdout, stderr, chunk_tx);
190
191            let outcome = run_child_loop(
192                &mut child,
193                pid,
194                &mut chunk_rx,
195                &mut on_data,
196                &cancel,
197                timeout_ms,
198            )
199            .await;
200
201            // Always reap and drain; cancellation wins over timeout/exit.
202            let exit_code = finalize_child(
203                &mut child,
204                pid,
205                &mut chunk_rx,
206                &mut on_data,
207                readers,
208                outcome,
209            )
210            .await?;
211
212            if cancel.is_cancelled() {
213                return Err(ToolError::new("aborted"));
214            }
215            if matches!(outcome, ChildLoopOutcome::TimedOut) {
216                return Err(ToolError::new(format!(
217                    "timeout:{}",
218                    timeout_seconds_label(timeout)
219                )));
220            }
221            Ok(exit_code)
222        }
223        .boxed()
224    }
225}
226
227#[derive(Clone, Copy, Debug, Eq, PartialEq)]
228enum ChildLoopOutcome {
229    Exited,
230    Cancelled,
231    TimedOut,
232    WaitFailed,
233}
234
235async fn run_child_loop(
236    child: &mut Child,
237    pid: Option<u32>,
238    chunk_rx: &mut mpsc::UnboundedReceiver<Vec<u8>>,
239    on_data: &mut (dyn FnMut(Vec<u8>) + Send),
240    cancel: &CancellationToken,
241    timeout_ms: Option<u64>,
242) -> ChildLoopOutcome {
243    let timeout_at = timeout_ms.map(|ms| Instant::now() + Duration::from_millis(ms));
244    let mut exited = false;
245
246    loop {
247        if cancel.is_cancelled() {
248            kill_process(pid);
249            return ChildLoopOutcome::Cancelled;
250        }
251        if timeout_at.is_some_and(|deadline| Instant::now() >= deadline) {
252            kill_process(pid);
253            return ChildLoopOutcome::TimedOut;
254        }
255
256        let wait_timeout = timeout_at.map(|deadline| {
257            deadline
258                .saturating_duration_since(Instant::now())
259                .max(Duration::from_millis(1))
260        });
261
262        tokio::select! {
263            biased;
264            () = cancel.cancelled() => {
265                kill_process(pid);
266                return ChildLoopOutcome::Cancelled;
267            }
268            () = tokio::time::sleep(wait_timeout.unwrap_or(Duration::from_hours(8760))),
269                if wait_timeout.is_some() =>
270            {
271                kill_process(pid);
272                return ChildLoopOutcome::TimedOut;
273            }
274            chunk = chunk_rx.recv() => {
275                if let Some(bytes) = chunk {
276                    on_data(bytes);
277                } else {
278                    // Both stream readers finished. Wait for process if needed.
279                    if !exited {
280                        return match child.wait().await {
281                            Ok(_) => ChildLoopOutcome::Exited,
282                            Err(_) => ChildLoopOutcome::WaitFailed,
283                        };
284                    }
285                    return ChildLoopOutcome::Exited;
286                }
287            }
288            status = child.wait(), if !exited => {
289                match status {
290                    Ok(_) => {
291                        exited = true;
292                        // Drain remaining pipe data until readers disconnect.
293                    }
294                    Err(_) => return ChildLoopOutcome::WaitFailed,
295                }
296            }
297        }
298
299        if exited {
300            return drain_exited_output(pid, chunk_rx, on_data, cancel).await;
301        }
302    }
303}
304
305async fn drain_exited_output(
306    pid: Option<u32>,
307    chunk_rx: &mut mpsc::UnboundedReceiver<Vec<u8>>,
308    on_data: &mut (dyn FnMut(Vec<u8>) + Send),
309    cancel: &CancellationToken,
310) -> ChildLoopOutcome {
311    let idle_deadline = Instant::now() + Duration::from_millis(100);
312    loop {
313        if cancel.is_cancelled() {
314            kill_process(pid);
315            return ChildLoopOutcome::Cancelled;
316        }
317        match chunk_rx.try_recv() {
318            Ok(bytes) => on_data(bytes),
319            Err(mpsc::error::TryRecvError::Empty) => {
320                if Instant::now() >= idle_deadline {
321                    return ChildLoopOutcome::Exited;
322                }
323                tokio::select! {
324                    () = cancel.cancelled() => {
325                        kill_process(pid);
326                        return ChildLoopOutcome::Cancelled;
327                    }
328                    chunk = chunk_rx.recv() => {
329                        match chunk {
330                            Some(bytes) => on_data(bytes),
331                            None => return ChildLoopOutcome::Exited,
332                        }
333                    }
334                    () = tokio::time::sleep(Duration::from_millis(5)) => {}
335                }
336            }
337            Err(mpsc::error::TryRecvError::Disconnected) => {
338                return ChildLoopOutcome::Exited;
339            }
340        }
341    }
342}
343
344fn kill_process(pid: Option<u32>) {
345    if let Some(pid) = pid {
346        kill_process_tree(pid);
347    }
348}
349
350async fn finalize_child(
351    child: &mut Child,
352    pid: Option<u32>,
353    chunk_rx: &mut mpsc::UnboundedReceiver<Vec<u8>>,
354    on_data: &mut (dyn FnMut(Vec<u8>) + Send),
355    readers: Vec<tokio::task::JoinHandle<()>>,
356    outcome: ChildLoopOutcome,
357) -> Result<Option<i32>, ToolError> {
358    if matches!(
359        outcome,
360        ChildLoopOutcome::Cancelled | ChildLoopOutcome::TimedOut
361    ) {
362        kill_process(pid);
363    }
364
365    // Drain any remaining output.
366    let drain_deadline = Instant::now() + Duration::from_millis(200);
367    while Instant::now() < drain_deadline {
368        match chunk_rx.try_recv() {
369            Ok(bytes) => on_data(bytes),
370            Err(mpsc::error::TryRecvError::Empty) => {
371                tokio::task::yield_now().await;
372            }
373            Err(mpsc::error::TryRecvError::Disconnected) => break,
374        }
375    }
376    while let Ok(bytes) = chunk_rx.try_recv() {
377        on_data(bytes);
378    }
379
380    for handle in readers {
381        let _ = handle.await;
382    }
383
384    match child.try_wait() {
385        Ok(Some(status)) => Ok(status.code()),
386        Ok(None) => {
387            if let Some(pid) = pid {
388                kill_process_tree(pid);
389            }
390            match child.wait().await {
391                Ok(status) => Ok(status.code()),
392                Err(error) => Err(ToolError::new(error.to_string())),
393            }
394        }
395        Err(error) => Err(ToolError::new(error.to_string())),
396    }
397}
398
399/// Agent tool that runs bash commands in a working directory.
400#[derive(Clone)]
401pub struct BashTool {
402    cwd: PathBuf,
403    command_prefix: Option<String>,
404    spawn_hook: Option<BashSpawnHook>,
405    operations: Arc<dyn BashOperations>,
406    parameters: Value,
407    description: String,
408}
409
410impl std::fmt::Debug for BashTool {
411    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
412        f.debug_struct("BashTool")
413            .field("cwd", &self.cwd)
414            .field("command_prefix", &self.command_prefix)
415            .field("spawn_hook", &self.spawn_hook.as_ref().map(|_| "Some(..)"))
416            .field("operations", &"<dyn BashOperations>")
417            .field("parameters", &self.parameters)
418            .field("description", &self.description)
419            .finish()
420    }
421}
422
423impl BashTool {
424    /// Creates a bash tool rooted at `cwd`.
425    #[must_use]
426    pub fn new(cwd: impl Into<PathBuf>) -> Self {
427        Self::with_options(BashToolOptions::new(cwd))
428    }
429
430    /// Creates a bash tool from explicit options.
431    #[must_use]
432    pub fn with_options(options: BashToolOptions) -> Self {
433        let operations = options.operations.unwrap_or_else(|| {
434            Arc::new(LocalBashOperations::new(options.shell_path.clone()))
435                as Arc<dyn BashOperations>
436        });
437        Self {
438            cwd: options.cwd,
439            command_prefix: options.command_prefix,
440            spawn_hook: options.spawn_hook,
441            operations,
442            parameters: bash_parameters_schema(),
443            description: bash_description(),
444        }
445    }
446
447    /// Returns the JSON Schema for bash arguments (normalized `TypeBox` shape).
448    #[must_use]
449    pub fn parameters_schema() -> Value {
450        bash_parameters_schema()
451    }
452
453    /// Validates raw tool arguments into [`BashToolInput`].
454    ///
455    /// # Errors
456    ///
457    /// Returns [`ToolError`] when required fields are missing or mistyped, or
458    /// when `timeout` fails TypeScript `resolveTimeoutMs` checks.
459    pub fn parse_input(args: &Map<String, Value>) -> Result<BashToolInput, ToolError> {
460        let input: BashToolInput = serde_json::from_value(Value::Object(args.clone()))
461            .map_err(|error| ToolError::new(format!("Bash tool input is invalid. {error}")))?;
462        let _ = resolve_timeout_ms(input.timeout)?;
463        Ok(input)
464    }
465}
466
467impl AgentTool for BashTool {
468    fn name(&self) -> &'static str {
469        "bash"
470    }
471
472    fn label(&self) -> &'static str {
473        "bash"
474    }
475
476    fn description(&self) -> &str {
477        &self.description
478    }
479
480    fn parameters(&self) -> &Value {
481        &self.parameters
482    }
483
484    fn validate_arguments(
485        &self,
486        args: &Map<String, Value>,
487    ) -> Result<Map<String, Value>, ToolError> {
488        let _ = Self::parse_input(args)?;
489        Ok(args.clone())
490    }
491
492    fn execute(
493        &self,
494        _tool_call_id: &str,
495        args: Map<String, Value>,
496        cancel: CancellationToken,
497        updates: ToolUpdates,
498    ) -> BoxFuture<'static, Result<AgentToolResult, ToolError>> {
499        let cwd = self.cwd.clone();
500        let command_prefix = self.command_prefix.clone();
501        let spawn_hook = self.spawn_hook.clone();
502        let operations = Arc::clone(&self.operations);
503
504        async move {
505            let input = BashTool::parse_input(&args)?;
506            let resolved_command = match command_prefix {
507                Some(prefix) => format!("{prefix}\n{}", input.command),
508                None => input.command,
509            };
510            let spawn_context = resolve_spawn_context(resolved_command, cwd, spawn_hook.as_ref());
511            let collected =
512                collect_command_output(operations, spawn_context, input.timeout, cancel, updates)
513                    .await?;
514            command_result(collected)
515        }
516        .boxed()
517    }
518}
519
520struct CollectedCommandOutput {
521    exec_result: Result<Option<i32>, ToolError>,
522    snapshot: OutputSnapshot,
523    last_line_bytes: usize,
524}
525
526async fn collect_command_output(
527    operations: Arc<dyn BashOperations>,
528    spawn_context: BashSpawnContext,
529    timeout: Option<f64>,
530    cancel: CancellationToken,
531    updates: ToolUpdates,
532) -> Result<CollectedCommandOutput, ToolError> {
533    updates.send(AgentToolResult {
534        content: Vec::new(),
535        details: Value::Null,
536        added_tool_names: None,
537        terminate: None,
538    });
539    let output = Arc::new(Mutex::new(OutputAccumulator::new(
540        OutputAccumulatorOptions {
541            max_lines: DEFAULT_MAX_LINES,
542            max_bytes: DEFAULT_MAX_BYTES,
543            temp_file_prefix: BASH_TEMP_FILE_PREFIX.to_owned(),
544        },
545    )));
546    let throttle = Arc::new(Mutex::new(UpdateThrottle::new()));
547    let (data_tx, mut data_rx) = mpsc::unbounded_channel::<Vec<u8>>();
548    let pump_output = Arc::clone(&output);
549    let pump_updates = updates.clone();
550    let pump_throttle = Arc::clone(&throttle);
551    let pump = tokio::spawn(async move {
552        while let Some(chunk) = data_rx.recv().await {
553            pump_output.lock().await.append(&chunk)?;
554            schedule_throttled_update(
555                pump_updates.clone(),
556                Arc::clone(&pump_output),
557                Arc::clone(&pump_throttle),
558            );
559        }
560        Ok::<(), OutputAccumulatorError>(())
561    });
562    let on_data = Box::new(move |bytes: Vec<u8>| {
563        let _ = data_tx.send(bytes);
564    });
565    let exec_result = operations
566        .exec(
567            spawn_context.command,
568            spawn_context.cwd,
569            on_data,
570            cancel,
571            timeout,
572            spawn_context.env,
573        )
574        .await;
575    finish_collected_output(exec_result, pump, output, throttle, updates).await
576}
577
578async fn finish_collected_output(
579    exec_result: Result<Option<i32>, ToolError>,
580    pump: tokio::task::JoinHandle<Result<(), OutputAccumulatorError>>,
581    output: Arc<Mutex<OutputAccumulator>>,
582    throttle: Arc<Mutex<UpdateThrottle>>,
583    updates: ToolUpdates,
584) -> Result<CollectedCommandOutput, ToolError> {
585    match pump.await {
586        Ok(Ok(())) => {}
587        Ok(Err(error)) => return Err(accumulator_error(&error)),
588        Err(error) => return Err(ToolError::new(format!("bash output pump failed: {error}"))),
589    }
590    for _ in 0..20 {
591        if !throttle.lock().await.timer_armed {
592            break;
593        }
594        tokio::time::sleep(Duration::from_millis(10)).await;
595    }
596    let mut output_guard = output.lock().await;
597    output_guard
598        .finish()
599        .map_err(|error| accumulator_error(&error))?;
600    let snapshot = output_guard
601        .snapshot(true)
602        .map_err(|error| accumulator_error(&error))?;
603    updates.send(snapshot_to_partial(&snapshot));
604    let last_line_bytes = output_guard.last_line_bytes();
605    output_guard.close_temp_file();
606    drop(output_guard);
607    Ok(CollectedCommandOutput {
608        exec_result,
609        snapshot,
610        last_line_bytes,
611    })
612}
613
614fn command_result(collected: CollectedCommandOutput) -> Result<AgentToolResult, ToolError> {
615    let CollectedCommandOutput {
616        exec_result,
617        snapshot,
618        last_line_bytes,
619    } = collected;
620    match exec_result {
621        Ok(exit_code) => {
622            let (text, details) = format_output(&snapshot, last_line_bytes, "(no output)");
623            if let Some(code) = exit_code
624                && code != 0
625            {
626                return Err(ToolError::new(append_status(
627                    &text,
628                    &format!("Command exited with code {code}"),
629                )));
630            }
631            Ok(AgentToolResult {
632                content: vec![ToolResultContent::Text(TextContent::new(text))],
633                details: details_to_value(details),
634                added_tool_names: None,
635                terminate: None,
636            })
637        }
638        Err(error) => command_error_result(&error, &snapshot, last_line_bytes),
639    }
640}
641
642fn command_error_result(
643    error: &ToolError,
644    snapshot: &OutputSnapshot,
645    last_line_bytes: usize,
646) -> Result<AgentToolResult, ToolError> {
647    let message = error.message().to_owned();
648    let (text, _) = format_output(snapshot, last_line_bytes, "");
649    if message == "aborted" {
650        return Err(ToolError::new(append_status(&text, "Command aborted")));
651    }
652    if let Some(secs) = message.strip_prefix("timeout:") {
653        return Err(ToolError::new(append_status(
654            &text,
655            &format!("Command timed out after {secs} seconds"),
656        )));
657    }
658    Err(ToolError::new(message))
659}
660
661/// Builds an [`Arc<dyn AgentTool>`] bash tool for `cwd`.
662#[must_use]
663pub fn create_bash_tool(cwd: impl Into<PathBuf>) -> Arc<dyn AgentTool> {
664    Arc::new(BashTool::new(cwd))
665}
666
667fn bash_description() -> String {
668    format!(
669        "Execute a bash 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.",
670        DEFAULT_MAX_BYTES / 1024
671    )
672}
673
674fn bash_parameters_schema() -> Value {
675    normalize_tool_schema(schemars::schema_for!(BashToolInput))
676}
677
678fn normalize_tool_schema(schema: schemars::Schema) -> Value {
679    let mut value = serde_json::to_value(schema).unwrap_or_else(|_| Value::Object(Map::new()));
680    if let Value::Object(map) = &mut value {
681        map.remove("$schema");
682        map.remove("title");
683        // TypeBox fixtures omit the root struct doc-comment description.
684        map.remove("description");
685        normalize_schema_node(map);
686    }
687    value
688}
689
690fn normalize_schema_node(map: &mut Map<String, Value>) {
691    map.remove("format");
692    // schemars represents `Option<number>` as `["number","null"]`; TypeBox uses
693    // a plain `"number"` with the key absent from `required`.
694    if let Some(Value::Array(types)) = map.get("type").cloned() {
695        let non_null: Vec<Value> = types
696            .into_iter()
697            .filter(|item| item.as_str() != Some("null"))
698            .collect();
699        if non_null.len() == 1 {
700            map.insert("type".to_owned(), non_null[0].clone());
701        }
702    }
703    let keys: Vec<String> = map.keys().cloned().collect();
704    for key in keys {
705        match map.get_mut(&key) {
706            Some(Value::Object(child)) => normalize_schema_node(child),
707            Some(Value::Array(items)) => {
708                for item in items {
709                    if let Value::Object(child) = item {
710                        normalize_schema_node(child);
711                    }
712                }
713            }
714            _ => {}
715        }
716    }
717}
718
719/// TypeScript `resolveTimeoutMs`.
720fn resolve_timeout_ms(timeout: Option<f64>) -> Result<Option<u64>, ToolError> {
721    let Some(timeout) = timeout else {
722        return Ok(None);
723    };
724    if !timeout.is_finite() || timeout <= 0.0 {
725        return Err(ToolError::new(
726            "Invalid timeout: must be a finite number of seconds",
727        ));
728    }
729    if timeout > MAX_TIMEOUT_SECONDS {
730        return Err(ToolError::new(format!(
731            "Invalid timeout: maximum is {MAX_TIMEOUT_SECONDS_DISPLAY} seconds"
732        )));
733    }
734    let milliseconds = (timeout * 1000.0).floor();
735    Ok(Some(
736        bounded_integer_f64_to_u64(milliseconds).min(MAX_TIMEOUT_MS),
737    ))
738}
739
740fn bounded_integer_f64_to_u64(value: f64) -> u64 {
741    const FRACTION_BITS: u32 = 52;
742    const FRACTION_BITS_I32: i32 = 52;
743    const EXPONENT_BIAS: i32 = 1023;
744
745    if value == 0.0 {
746        return 0;
747    }
748    let bits = value.to_bits();
749    let exponent_bits = (bits >> FRACTION_BITS) & 0x7ff;
750    let fraction = bits & ((1_u64 << FRACTION_BITS) - 1);
751    let significand = fraction | (1_u64 << FRACTION_BITS);
752    let exponent = i32::try_from(exponent_bits).unwrap_or(0) - EXPONENT_BIAS;
753    let denominator_shift = u32::try_from(FRACTION_BITS_I32 - exponent).unwrap_or(FRACTION_BITS);
754    significand >> denominator_shift
755}
756
757fn timeout_seconds_label(timeout: Option<f64>) -> String {
758    timeout.map_or_else(|| "0".to_owned(), |value| value.to_string())
759}
760
761struct ShellConfig {
762    shell: PathBuf,
763    args: Vec<String>,
764    command_from_stdin: bool,
765}
766
767fn resolve_shell_config(custom: Option<&Path>) -> Result<ShellConfig, ToolError> {
768    if let Some(path) = custom {
769        if path.exists() {
770            return Ok(bash_shell_config(path.to_path_buf()));
771        }
772        return Err(ToolError::new(format!(
773            "Custom shell path not found: {}",
774            path.display()
775        )));
776    }
777
778    #[cfg(windows)]
779    {
780        let mut candidates = Vec::new();
781        if let Ok(program_files) = std::env::var("ProgramFiles") {
782            candidates.push(PathBuf::from(program_files).join(r"Git\bin\bash.exe"));
783        }
784        if let Ok(program_files_x86) = std::env::var("ProgramFiles(x86)") {
785            candidates.push(PathBuf::from(program_files_x86).join(r"Git\bin\bash.exe"));
786        }
787        for path in &candidates {
788            if path.exists() {
789                return Ok(bash_shell_config(path.clone()));
790            }
791        }
792        if let Some(path) = find_on_path("bash.exe") {
793            return Ok(bash_shell_config(path));
794        }
795        return Err(ToolError::new(
796            "No bash shell found. Options:\n  1. Install Git for Windows: https://git-scm.com/download/win\n  2. Add your bash to PATH (Cygwin, MSYS2, etc.)\n  3. Set shellPath in settings.json".to_owned(),
797        ));
798    }
799
800    #[cfg(not(windows))]
801    {
802        if Path::new("/bin/bash").exists() {
803            return Ok(bash_shell_config(PathBuf::from("/bin/bash")));
804        }
805        if let Some(path) = find_on_path("bash") {
806            return Ok(bash_shell_config(path));
807        }
808        Ok(ShellConfig {
809            shell: PathBuf::from("sh"),
810            args: vec!["-c".to_owned()],
811            command_from_stdin: false,
812        })
813    }
814}
815
816fn bash_shell_config(shell: PathBuf) -> ShellConfig {
817    let legacy_wsl = shell
818        .to_string_lossy()
819        .replace('\\', "/")
820        .to_ascii_lowercase()
821        .contains("/windows/system32/bash.exe");
822    if legacy_wsl {
823        ShellConfig {
824            shell,
825            args: vec!["-s".to_owned()],
826            command_from_stdin: true,
827        }
828    } else {
829        ShellConfig {
830            shell,
831            args: vec!["-c".to_owned()],
832            command_from_stdin: false,
833        }
834    }
835}
836
837fn find_on_path(name: &str) -> Option<PathBuf> {
838    let path_var = std::env::var_os("PATH")?;
839    for dir in std::env::split_paths(&path_var) {
840        let candidate = dir.join(name);
841        if candidate.is_file() {
842            return Some(candidate);
843        }
844    }
845    None
846}
847
848fn current_env_map() -> HashMap<String, String> {
849    std::env::vars().collect()
850}
851
852fn resolve_spawn_context(
853    command: String,
854    cwd: PathBuf,
855    spawn_hook: Option<&BashSpawnHook>,
856) -> BashSpawnContext {
857    let base = BashSpawnContext {
858        command,
859        cwd,
860        env: current_env_map(),
861    };
862    match spawn_hook {
863        Some(hook) => hook(base),
864        None => base,
865    }
866}
867
868fn spawn_shell_command(
869    shell: &ShellConfig,
870    command: &str,
871    cwd: &Path,
872    env: &HashMap<String, String>,
873) -> Result<Child, ToolError> {
874    let mut cmd = Command::new(&shell.shell);
875    cmd.args(&shell.args);
876    if !shell.command_from_stdin {
877        cmd.arg(command);
878    }
879    cmd.current_dir(cwd)
880        .stdin(if shell.command_from_stdin {
881            Stdio::piped()
882        } else {
883            Stdio::null()
884        })
885        .stdout(Stdio::piped())
886        .stderr(Stdio::piped())
887        .kill_on_drop(true)
888        .envs(env.iter().map(|(k, v)| (k.as_str(), v.as_str())));
889
890    #[cfg(unix)]
891    {
892        // Detached process group so killpg reaps descendants (Node `detached`).
893        cmd.process_group(0);
894    }
895
896    #[cfg(windows)]
897    {
898        use std::os::windows::process::CommandExt as _;
899        const CREATE_NO_WINDOW: u32 = 0x0800_0000;
900        cmd.creation_flags(CREATE_NO_WINDOW);
901    }
902
903    let mut child = cmd
904        .spawn()
905        .map_err(|error| ToolError::new(format!("Failed to spawn shell: {error}")))?;
906
907    if shell.command_from_stdin
908        && let Some(mut stdin) = child.stdin.take()
909    {
910        let payload = command.to_owned();
911        tokio::spawn(async move {
912            let _ = stdin.write_all(payload.as_bytes()).await;
913            let _ = stdin.shutdown().await;
914        });
915    }
916
917    Ok(child)
918}
919
920fn spawn_stream_readers(
921    stdout: Option<impl AsyncRead + Unpin + Send + 'static>,
922    stderr: Option<impl AsyncRead + Unpin + Send + 'static>,
923    tx: mpsc::UnboundedSender<Vec<u8>>,
924) -> Vec<tokio::task::JoinHandle<()>> {
925    let mut handles = Vec::new();
926    if let Some(stdout) = stdout {
927        let tx = tx.clone();
928        handles.push(tokio::spawn(async move {
929            pump_reader(stdout, tx).await;
930        }));
931    }
932    if let Some(stderr) = stderr {
933        handles.push(tokio::spawn(async move {
934            pump_reader(stderr, tx).await;
935        }));
936    }
937    handles
938}
939
940async fn pump_reader<R>(mut reader: R, tx: mpsc::UnboundedSender<Vec<u8>>)
941where
942    R: AsyncRead + Unpin,
943{
944    let mut buf = vec![0_u8; 8192];
945    loop {
946        match reader.read(&mut buf).await {
947            Ok(0) | Err(_) => break,
948            Ok(n) => {
949                if tx.send(buf[..n].to_vec()).is_err() {
950                    break;
951                }
952            }
953        }
954    }
955}
956
957fn kill_process_tree(pid: u32) {
958    #[cfg(unix)]
959    {
960        use nix::sys::signal::{Signal, kill, killpg};
961        use nix::unistd::Pid;
962
963        let Ok(raw) = i32::try_from(pid) else {
964            return;
965        };
966        let group = Pid::from_raw(raw);
967        // Node: process.kill(-pid, SIGKILL) then fallback process.kill(pid).
968        if killpg(group, Signal::SIGKILL).is_err() {
969            let _ = kill(group, Signal::SIGKILL);
970        }
971    }
972
973    #[cfg(windows)]
974    {
975        let _ = std::process::Command::new("taskkill")
976            .args(["/F", "/T", "/PID", &pid.to_string()])
977            .stdin(Stdio::null())
978            .stdout(Stdio::null())
979            .stderr(Stdio::null())
980            .spawn();
981    }
982}
983
984struct UpdateThrottle {
985    last_update_at: Option<Instant>,
986    dirty: bool,
987    timer_armed: bool,
988}
989
990impl UpdateThrottle {
991    fn new() -> Self {
992        Self {
993            last_update_at: None,
994            dirty: false,
995            timer_armed: false,
996        }
997    }
998}
999
1000fn schedule_throttled_update(
1001    updates: ToolUpdates,
1002    output: Arc<Mutex<OutputAccumulator>>,
1003    state: Arc<Mutex<UpdateThrottle>>,
1004) {
1005    tokio::spawn(async move {
1006        {
1007            let mut guard = state.lock().await;
1008            guard.dirty = true;
1009            if guard.timer_armed {
1010                return;
1011            }
1012            guard.timer_armed = true;
1013        }
1014
1015        loop {
1016            let delay = {
1017                let guard = state.lock().await;
1018                match guard.last_update_at {
1019                    Some(last) => BASH_UPDATE_THROTTLE.saturating_sub(last.elapsed()),
1020                    None => Duration::ZERO,
1021                }
1022            };
1023            if !delay.is_zero() {
1024                tokio::time::sleep(delay).await;
1025            }
1026
1027            let should_emit = {
1028                let mut guard = state.lock().await;
1029                if guard.dirty {
1030                    guard.dirty = false;
1031                    guard.last_update_at = Some(Instant::now());
1032                    true
1033                } else {
1034                    false
1035                }
1036            };
1037            if should_emit {
1038                let mut output_guard = output.lock().await;
1039                if let Ok(snapshot) = output_guard.snapshot(true) {
1040                    updates.send(snapshot_to_partial(&snapshot));
1041                }
1042            }
1043
1044            let mut guard = state.lock().await;
1045            if guard.dirty {
1046                continue;
1047            }
1048            guard.timer_armed = false;
1049            break;
1050        }
1051    });
1052}
1053
1054fn snapshot_to_partial(snapshot: &OutputSnapshot) -> AgentToolResult {
1055    let mut details = Map::new();
1056    if snapshot.truncation.truncated
1057        && let Ok(value) = serde_json::to_value(&snapshot.truncation)
1058    {
1059        details.insert("truncation".to_owned(), value);
1060    }
1061    if let Some(path) = &snapshot.full_output_path {
1062        details.insert(
1063            "fullOutputPath".to_owned(),
1064            Value::String(path.to_string_lossy().into_owned()),
1065        );
1066    }
1067    AgentToolResult {
1068        content: vec![ToolResultContent::Text(TextContent::new(
1069            snapshot.content.clone(),
1070        ))],
1071        details: if details.is_empty() {
1072            Value::Null
1073        } else {
1074            Value::Object(details)
1075        },
1076        added_tool_names: None,
1077        terminate: None,
1078    }
1079}
1080
1081fn format_output(
1082    snapshot: &OutputSnapshot,
1083    last_line_bytes: usize,
1084    empty_text: &str,
1085) -> (String, Option<BashToolDetails>) {
1086    let truncation = &snapshot.truncation;
1087    let mut text = if snapshot.content.is_empty() {
1088        empty_text.to_owned()
1089    } else {
1090        snapshot.content.clone()
1091    };
1092    let mut details = None;
1093    if truncation.truncated {
1094        let full_output_path = snapshot
1095            .full_output_path
1096            .as_ref()
1097            .map(|path| path.to_string_lossy().into_owned());
1098        details = Some(BashToolDetails {
1099            truncation: Some(truncation.clone()),
1100            full_output_path: full_output_path.clone(),
1101        });
1102        let start_line = truncation
1103            .total_lines
1104            .saturating_sub(truncation.output_lines)
1105            + 1;
1106        let end_line = truncation.total_lines;
1107        let path_display = full_output_path.unwrap_or_default();
1108        if truncation.last_line_partial {
1109            let last_line_size = format_size(last_line_bytes as u64);
1110            let _ = write!(
1111                text,
1112                "\n\n[Showing last {} of line {end_line} (line is {last_line_size}). Full output: {path_display}]",
1113                format_size(truncation.output_bytes as u64)
1114            );
1115        } else if truncation.truncated_by == Some(TruncatedBy::Lines) {
1116            let _ = write!(
1117                text,
1118                "\n\n[Showing lines {start_line}-{end_line} of {}. Full output: {path_display}]",
1119                truncation.total_lines
1120            );
1121        } else {
1122            let _ = write!(
1123                text,
1124                "\n\n[Showing lines {start_line}-{end_line} of {} ({} limit). Full output: {path_display}]",
1125                truncation.total_lines,
1126                format_size(DEFAULT_MAX_BYTES as u64)
1127            );
1128        }
1129    }
1130    (text, details)
1131}
1132
1133fn append_status(text: &str, status: &str) -> String {
1134    if text.is_empty() {
1135        status.to_owned()
1136    } else {
1137        format!("{text}\n\n{status}")
1138    }
1139}
1140
1141fn details_to_value(details: Option<BashToolDetails>) -> Value {
1142    match details {
1143        Some(details) => serde_json::to_value(details).unwrap_or(Value::Null),
1144        None => Value::Null,
1145    }
1146}
1147
1148fn accumulator_error(error: &OutputAccumulatorError) -> ToolError {
1149    ToolError::new(error.to_string())
1150}
1151
1152#[cfg(test)]
1153mod tests {
1154    use super::*;
1155    use std::io;
1156    use std::sync::atomic::{AtomicUsize, Ordering};
1157
1158    use serde_json::json;
1159    use tempfile::tempdir;
1160
1161    fn fixture_schema() -> Result<Value, serde_json::Error> {
1162        let text = include_str!("../../../tests/fixtures/tool-schemas/bash.json");
1163        serde_json::from_str(text)
1164    }
1165
1166    fn json_map(value: Value) -> Result<Map<String, Value>, Box<dyn std::error::Error>> {
1167        match value {
1168            Value::Object(map) => Ok(map),
1169            _ => Err(io::Error::other("test arguments must be a JSON object").into()),
1170        }
1171    }
1172
1173    fn required<T>(
1174        value: Option<T>,
1175        message: &'static str,
1176    ) -> Result<T, Box<dyn std::error::Error>> {
1177        value.ok_or_else(|| io::Error::other(message).into())
1178    }
1179
1180    fn expected_error<T>(
1181        result: Result<T, ToolError>,
1182        message: &'static str,
1183    ) -> Result<ToolError, Box<dyn std::error::Error>> {
1184        match result {
1185            Err(error) => Ok(error),
1186            Ok(_) => Err(io::Error::other(message).into()),
1187        }
1188    }
1189
1190    fn text_of(result: &AgentToolResult) -> String {
1191        match result.content.first() {
1192            Some(ToolResultContent::Text(text)) => text.text.to_string(),
1193            _ => String::new(),
1194        }
1195    }
1196
1197    fn text_of_err(error: &ToolError) -> String {
1198        error.message().to_owned()
1199    }
1200
1201    #[test]
1202    fn schema_matches_typebox_fixture() -> Result<(), Box<dyn std::error::Error>> {
1203        let schema = BashTool::parameters_schema();
1204        assert_eq!(schema, fixture_schema()?);
1205        Ok(())
1206    }
1207
1208    #[test]
1209    fn timeout_validation_rejects_non_positive_and_too_large()
1210    -> Result<(), Box<dyn std::error::Error>> {
1211        assert_eq!(
1212            expected_error(resolve_timeout_ms(Some(0.0)), "zero accepted")?.message(),
1213            "Invalid timeout: must be a finite number of seconds"
1214        );
1215        assert_eq!(
1216            expected_error(resolve_timeout_ms(Some(-1.0)), "negative accepted")?.message(),
1217            "Invalid timeout: must be a finite number of seconds"
1218        );
1219        assert_eq!(
1220            expected_error(resolve_timeout_ms(Some(f64::NAN)), "NaN accepted")?.message(),
1221            "Invalid timeout: must be a finite number of seconds"
1222        );
1223        assert_eq!(
1224            expected_error(resolve_timeout_ms(Some(f64::INFINITY)), "infinity accepted",)?
1225                .message(),
1226            "Invalid timeout: must be a finite number of seconds"
1227        );
1228        let too_large = MAX_TIMEOUT_SECONDS + 0.001;
1229        assert_eq!(
1230            expected_error(resolve_timeout_ms(Some(too_large)), "large accepted")?.message(),
1231            format!("Invalid timeout: maximum is {MAX_TIMEOUT_SECONDS_DISPLAY} seconds")
1232        );
1233        assert_eq!(resolve_timeout_ms(Some(1.0))?, Some(1000));
1234        assert_eq!(resolve_timeout_ms(Some(0.000_999))?, Some(0));
1235        assert_eq!(resolve_timeout_ms(Some(0.001))?, Some(1));
1236        assert_eq!(
1237            resolve_timeout_ms(Some(MAX_TIMEOUT_SECONDS))?,
1238            Some(MAX_TIMEOUT_MS)
1239        );
1240        assert!(resolve_timeout_ms(None)?.is_none());
1241        Ok(())
1242    }
1243
1244    #[test]
1245    fn validate_arguments_rejects_bad_timeout() -> Result<(), Box<dyn std::error::Error>> {
1246        let tool = BashTool::new("/tmp");
1247        let err = expected_error(
1248            tool.validate_arguments(&json_map(json!({"command": "true", "timeout": 0}))?),
1249            "bad timeout accepted",
1250        )?;
1251        assert_eq!(
1252            err.message(),
1253            "Invalid timeout: must be a finite number of seconds"
1254        );
1255        Ok(())
1256    }
1257
1258    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1259    async fn empty_success_returns_no_output() -> Result<(), Box<dyn std::error::Error>> {
1260        let dir = tempdir()?;
1261        let tool = BashTool::new(dir.path());
1262        let result = tool
1263            .execute(
1264                "1",
1265                json_map(json!({"command": "true"}))?,
1266                CancellationToken::new(),
1267                ToolUpdates::noop(),
1268            )
1269            .await?;
1270        assert_eq!(text_of(&result), "(no output)");
1271        assert!(result.details.is_null());
1272        Ok(())
1273    }
1274
1275    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1276    async fn nonempty_success_returns_stdout() -> Result<(), Box<dyn std::error::Error>> {
1277        let dir = tempdir()?;
1278        let tool = BashTool::new(dir.path());
1279        let result = tool
1280            .execute(
1281                "1",
1282                json_map(json!({"command": "printf 'hello\n'"}))?,
1283                CancellationToken::new(),
1284                ToolUpdates::noop(),
1285            )
1286            .await?;
1287        assert_eq!(text_of(&result), "hello\n");
1288        assert!(result.details.is_null());
1289        Ok(())
1290    }
1291
1292    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1293    async fn nonzero_exit_appends_status() -> Result<(), Box<dyn std::error::Error>> {
1294        let dir = tempdir()?;
1295        let tool = BashTool::new(dir.path());
1296        let err = expected_error(
1297            tool.execute(
1298                "1",
1299                json_map(json!({"command": "printf 'fail\n'; exit 7"}))?,
1300                CancellationToken::new(),
1301                ToolUpdates::noop(),
1302            )
1303            .await,
1304            "nonzero exit succeeded",
1305        )?;
1306        assert_eq!(text_of_err(&err), "fail\n\n\nCommand exited with code 7");
1307        Ok(())
1308    }
1309
1310    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1311    async fn interleaves_stdout_and_stderr_by_arrival() -> Result<(), Box<dyn std::error::Error>> {
1312        let dir = tempdir()?;
1313        let tool = BashTool::new(dir.path());
1314        let result = tool
1315            .execute(
1316                "1",
1317                json_map(json!({
1318                    "command": "printf 'out1\n'; sleep 0.05; printf 'err1\n' 1>&2; sleep 0.05; printf 'out2\n'"
1319                }))?,
1320                CancellationToken::new(),
1321                ToolUpdates::noop(),
1322            )
1323            .await?;
1324        let text = text_of(&result);
1325        assert!(text.contains("out1"));
1326        assert!(text.contains("err1"));
1327        assert!(text.contains("out2"));
1328        let out1 = required(text.find("out1"), "out1 missing")?;
1329        let err1 = required(text.find("err1"), "err1 missing")?;
1330        let out2 = required(text.find("out2"), "out2 missing")?;
1331        assert!(out1 < err1 && err1 < out2, "text={text:?}");
1332        Ok(())
1333    }
1334
1335    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1336    async fn timeout_kills_descendants() -> Result<(), Box<dyn std::error::Error>> {
1337        let dir = tempdir()?;
1338        let marker = dir.path().join("child.pid");
1339        let marker_str = marker.display().to_string();
1340        let tool = BashTool::new(dir.path());
1341        let command = format!("sleep 60 & echo $! > '{marker_str}'; wait");
1342        let err = expected_error(
1343            tool.execute(
1344                "1",
1345                json_map(json!({"command": command, "timeout": 0.3}))?,
1346                CancellationToken::new(),
1347                ToolUpdates::noop(),
1348            )
1349            .await,
1350            "timed command succeeded",
1351        )?;
1352        assert!(
1353            text_of_err(&err).contains("Command timed out after 0.3 seconds"),
1354            "{}",
1355            text_of_err(&err)
1356        );
1357
1358        let mut child_pid = None;
1359        for _ in 0..50 {
1360            if marker.exists() {
1361                let raw = tokio::fs::read_to_string(&marker).await?;
1362                child_pid = raw.trim().parse::<u32>().ok();
1363                break;
1364            }
1365            tokio::time::sleep(Duration::from_millis(20)).await;
1366        }
1367        if let Some(pid) = child_pid {
1368            for _ in 0..50 {
1369                if !process_alive(pid) {
1370                    break;
1371                }
1372                tokio::time::sleep(Duration::from_millis(20)).await;
1373            }
1374            assert!(!process_alive(pid), "descendant {pid} still alive");
1375        }
1376        Ok(())
1377    }
1378
1379    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1380    async fn cancel_kills_descendants_and_wins_race() -> Result<(), Box<dyn std::error::Error>> {
1381        let dir = tempdir()?;
1382        let marker = dir.path().join("cancel.pid");
1383        let marker_str = marker.display().to_string();
1384        let tool = BashTool::new(dir.path());
1385        let cancel = CancellationToken::new();
1386        let cancel_task = cancel.clone();
1387        let command = format!("sleep 60 & echo $! > '{marker_str}'; wait");
1388
1389        let cancel_args = json_map(json!({"command": command, "timeout": 30.0}))?;
1390        let join = tokio::spawn(async move {
1391            tool.execute("1", cancel_args, cancel_task, ToolUpdates::noop())
1392                .await
1393        });
1394
1395        let mut child_pid = None;
1396        for _ in 0..100 {
1397            if marker.exists() {
1398                let raw = tokio::fs::read_to_string(&marker).await?;
1399                child_pid = raw.trim().parse::<u32>().ok();
1400                break;
1401            }
1402            tokio::time::sleep(Duration::from_millis(20)).await;
1403        }
1404        cancel.cancel();
1405        let err = expected_error(join.await?, "cancelled command succeeded")?;
1406        assert!(
1407            text_of_err(&err).contains("Command aborted"),
1408            "{}",
1409            text_of_err(&err)
1410        );
1411        if let Some(pid) = child_pid {
1412            for _ in 0..50 {
1413                if !process_alive(pid) {
1414                    break;
1415                }
1416                tokio::time::sleep(Duration::from_millis(20)).await;
1417            }
1418            assert!(!process_alive(pid), "descendant {pid} still alive");
1419        }
1420        Ok(())
1421    }
1422
1423    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1424    async fn partial_updates_are_throttled() -> Result<(), Box<dyn std::error::Error>> {
1425        let dir = tempdir()?;
1426        let tool = BashTool::new(dir.path());
1427        let seen = Arc::new(AtomicUsize::new(0));
1428        let seen_c = Arc::clone(&seen);
1429        let updates = ToolUpdates::new(move |_partial| {
1430            seen_c.fetch_add(1, Ordering::SeqCst);
1431        });
1432        let _ = tool
1433            .execute(
1434                "1",
1435                json_map(json!({
1436                    "command": "python3 - <<'PY'\nimport sys,time\nfor i in range(20):\n    sys.stdout.write(f'line-{i}\\n')\n    sys.stdout.flush()\n    time.sleep(0.02)\nPY"
1437                }))?,
1438                CancellationToken::new(),
1439                updates,
1440            )
1441            .await?;
1442        let count = seen.load(Ordering::SeqCst);
1443        assert!(count >= 2, "expected streaming updates, got {count}");
1444        assert!(count < 20, "updates were not throttled: {count}");
1445        Ok(())
1446    }
1447
1448    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1449    async fn line_truncation_notice_and_spill() -> Result<(), Box<dyn std::error::Error>> {
1450        let dir = tempdir()?;
1451        let tool = BashTool::new(dir.path());
1452        let result = tool
1453            .execute(
1454                "1",
1455                json_map(json!({
1456                    "command": "python3 - <<'PY'\nfor i in range(2100):\n    print(f'L{i}')\nPY"
1457                }))?,
1458                CancellationToken::new(),
1459                ToolUpdates::noop(),
1460            )
1461            .await?;
1462        let text = text_of(&result);
1463        assert!(
1464            text.contains("Showing lines ") && text.contains("Full output: "),
1465            "{text}"
1466        );
1467        let details = required(result.details.as_object(), "details object missing")?;
1468        let path = required(
1469            details.get("fullOutputPath").and_then(Value::as_str),
1470            "spill path missing",
1471        )?;
1472        assert!(
1473            Path::new(path)
1474                .file_name()
1475                .and_then(|name| name.to_str())
1476                .is_some_and(|name| {
1477                    name.starts_with("pi-bash-")
1478                        && Path::new(name)
1479                            .extension()
1480                            .is_some_and(|extension| extension.eq_ignore_ascii_case("log"))
1481                }),
1482            "path={path}"
1483        );
1484        assert!(Path::new(path).is_file(), "spill missing: {path}");
1485        let full = tokio::fs::read_to_string(path).await?;
1486        assert!(full.lines().count() >= 2100);
1487        Ok(())
1488    }
1489
1490    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1491    async fn byte_truncation_notice_and_spill() -> Result<(), Box<dyn std::error::Error>> {
1492        let dir = tempdir()?;
1493        let tool = BashTool::new(dir.path());
1494        let result = tool
1495            .execute(
1496                "1",
1497                json_map(json!({
1498                    "command": format!(
1499                        "python3 - <<'PY'\nprint('x'*{})\nPY",
1500                        DEFAULT_MAX_BYTES + 100
1501                    )
1502                }))?,
1503                CancellationToken::new(),
1504                ToolUpdates::noop(),
1505            )
1506            .await?;
1507        let text = text_of(&result);
1508        assert!(
1509            text.contains("limit). Full output: ") || text.contains("of line "),
1510            "{text}"
1511        );
1512        let details = required(result.details.as_object(), "details object missing")?;
1513        let path = required(
1514            details.get("fullOutputPath").and_then(Value::as_str),
1515            "spill path missing",
1516        )?;
1517        assert!(Path::new(path).is_file());
1518        Ok(())
1519    }
1520
1521    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1522    async fn untruncated_has_no_spill() -> Result<(), Box<dyn std::error::Error>> {
1523        let dir = tempdir()?;
1524        let tool = BashTool::new(dir.path());
1525        let result = tool
1526            .execute(
1527                "1",
1528                json_map(json!({"command": "printf 'small\n'"}))?,
1529                CancellationToken::new(),
1530                ToolUpdates::noop(),
1531            )
1532            .await?;
1533        assert_eq!(text_of(&result), "small\n");
1534        assert!(result.details.is_null());
1535        Ok(())
1536    }
1537
1538    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1539    async fn missing_cwd_errors() -> Result<(), Box<dyn std::error::Error>> {
1540        let dir = tempdir()?;
1541        let missing = dir.path().join("nope");
1542        let tool = BashTool::new(&missing);
1543        let err = expected_error(
1544            tool.execute(
1545                "1",
1546                json_map(json!({"command": "true"}))?,
1547                CancellationToken::new(),
1548                ToolUpdates::noop(),
1549            )
1550            .await,
1551            "missing cwd accepted",
1552        )?;
1553        assert!(
1554            text_of_err(&err).contains("Working directory does not exist:"),
1555            "{}",
1556            text_of_err(&err)
1557        );
1558        Ok(())
1559    }
1560
1561    fn process_alive(pid: u32) -> bool {
1562        #[cfg(unix)]
1563        {
1564            use nix::sys::signal::kill;
1565            use nix::unistd::Pid;
1566            i32::try_from(pid).is_ok_and(|raw| kill(Pid::from_raw(raw), None).is_ok())
1567        }
1568        #[cfg(not(unix))]
1569        {
1570            let _ = pid;
1571            false
1572        }
1573    }
1574}