bamboo_tools/tools/
exec_command.rs

1//! Yielding shell execution whose results are pulled through ordinary tools.
2use async_trait::async_trait;
3use bamboo_agent_core::{Tool, ToolCtx, ToolError, ToolOutcome, ToolResult};
4use serde::{Deserialize, Serialize};
5use serde_json::json;
6use std::path::{Path, PathBuf};
7use std::sync::atomic::{AtomicBool, Ordering};
8use std::sync::{Arc, OnceLock};
9use tokio::io::{AsyncRead, AsyncReadExt};
10use tokio::sync::{watch, Mutex, Notify, OwnedSemaphorePermit, Semaphore};
11use tokio::time::{timeout, Duration, Instant};
12
13use super::{bash_runtime, workspace_state};
14
15const MAX_RETAINED_SESSIONS: usize = 64;
16const MAX_CAPTURE_BYTES: usize = 256 * 1024;
17const MAX_OUTPUT_TOKENS: usize = 8192;
18const DEFAULT_OUTPUT_TOKENS: usize = 2000;
19const MAX_COMMAND_BYTES: usize = 64 * 1024;
20// Advisory only: silence never changes process lifetime or terminal status.
21const QUIET_POLL_ADVISORY_AFTER: Duration = Duration::from_secs(90);
22const QUIET_POLL_ADVISORY_READOUTS: usize = 3;
23const QUIET_POLL_DIAGNOSIS_READOUTS: usize = 6;
24const QUIET_POLL_ADVISORY: &str = "No new output across repeated polls. Do other useful work or check a command-owned progress signal before polling again. Quiet output or an existing report does not prove completion. Continue waiting when appropriate, or report the concrete blocker; retain this handle for the final status.";
25
26#[derive(Debug, Clone)]
27struct CapturedOutput {
28    bytes: Vec<u8>,
29    omitted_bytes: usize,
30    completion: Option<(&'static str, Option<i32>)>,
31    error: Option<String>,
32    started_at: Instant,
33    last_output_at: Instant,
34    consecutive_empty_readouts: usize,
35}
36
37impl Default for CapturedOutput {
38    fn default() -> Self {
39        let now = Instant::now();
40        Self {
41            bytes: Vec::new(),
42            omitted_bytes: 0,
43            completion: None,
44            error: None,
45            started_at: now,
46            last_output_at: now,
47            consecutive_empty_readouts: 0,
48        }
49    }
50}
51
52/// Per-handle pull cursor and bounded capture. Process ownership remains in
53/// `ManagedBashChild`; this state can outlive a drained local resource scope.
54#[derive(Debug)]
55pub(crate) struct PolledShellState {
56    output: Mutex<CapturedOutput>,
57    completed: Notify,
58    // A reaped parent may still have descendants holding its output pipes.
59    // This observation does not publish a terminal capture result.
60    process_exited: AtomicBool,
61    background_monitor: AtomicBool,
62    #[cfg(test)]
63    fail_cleanup_after_reap: AtomicBool,
64    capture_stop: watch::Sender<bool>,
65    pub(crate) interaction: Mutex<()>,
66    _permit: OwnedSemaphorePermit,
67}
68
69impl PolledShellState {
70    fn new() -> Result<Arc<Self>, ToolError> {
71        static CAPACITY: OnceLock<Arc<Semaphore>> = OnceLock::new();
72        let permit = CAPACITY
73            .get_or_init(|| Arc::new(Semaphore::new(MAX_RETAINED_SESSIONS)))
74            .clone()
75            .try_acquire_owned()
76            .map_err(|_| ToolError::Execution("Exec session capacity exhausted; finish or stop an existing session before starting another command".into()))?;
77        Ok(Arc::new(Self {
78            output: Mutex::new(CapturedOutput::default()),
79            completed: Notify::new(),
80            process_exited: AtomicBool::new(false),
81            background_monitor: AtomicBool::new(false),
82            #[cfg(test)]
83            fail_cleanup_after_reap: AtomicBool::new(false),
84            capture_stop: watch::channel(false).0,
85            interaction: Mutex::new(()),
86            _permit: permit,
87        }))
88    }
89
90    async fn push(&self, bytes: &[u8]) {
91        if bytes.is_empty() {
92            return;
93        }
94        let mut output = self.output.lock().await;
95        output.last_output_at = Instant::now();
96        output.consecutive_empty_readouts = 0;
97        output.bytes.extend_from_slice(bytes);
98        let mut excess = output.bytes.len().saturating_sub(MAX_CAPTURE_BYTES);
99        if excess > 0 {
100            while excess < output.bytes.len() && output.bytes[excess] & 0b1100_0000 == 0b1000_0000 {
101                excess += 1;
102            }
103            output.bytes.drain(..excess);
104            output.omitted_bytes = output.omitted_bytes.saturating_add(excess);
105        }
106    }
107
108    pub(crate) async fn record_error(&self, error: String) {
109        self.output.lock().await.error.get_or_insert(error);
110    }
111
112    pub(crate) async fn record_cleanup_error(&self, error: String) {
113        // A real wait/group-drain failure takes priority over an earlier
114        // capture timeout so the retained continuation explains failed cleanup.
115        self.output.lock().await.error = Some(error);
116    }
117
118    #[cfg(test)]
119    pub(super) fn test_cleanup_after_reap(&self) -> std::io::Result<()> {
120        if self.fail_cleanup_after_reap.swap(false, Ordering::AcqRel) {
121            Err(std::io::Error::other("controlled post-reap cleanup error"))
122        } else {
123            Ok(())
124        }
125    }
126
127    pub(crate) fn stop_capture(&self) {
128        self.capture_stop.send_replace(true);
129    }
130
131    pub(crate) async fn finish(&self, status: &'static str, exit_code: Option<i32>) {
132        let mut output = self.output.lock().await;
133        let status = if output.error.is_some() {
134            "error"
135        } else {
136            status
137        };
138        output.completion = Some((status, exit_code));
139        drop(output);
140        self.completed.notify_waiters();
141    }
142
143    pub(super) fn mark_process_exited(&self) {
144        self.process_exited.store(true, Ordering::Release);
145    }
146
147    pub(super) fn process_exited(&self) -> bool {
148        self.process_exited.load(Ordering::Acquire)
149    }
150
151    pub(super) async fn wait(&self, yield_time_ms: u64) {
152        let notified = self.completed.notified();
153        tokio::pin!(notified);
154        // Register before observing completion: even an immediate child exit
155        // cannot slip between the state check and the notification wait.
156        notified.as_mut().enable();
157        let running = self.output.lock().await.completion.is_none();
158        if running {
159            let _ = timeout(Duration::from_millis(yield_time_ms), notified).await;
160        }
161    }
162
163    pub(crate) async fn is_complete(&self) -> bool {
164        self.output.lock().await.completion.is_some()
165    }
166
167    pub(super) async fn reset_empty_readouts(&self) {
168        self.output.lock().await.consecutive_empty_readouts = 0;
169    }
170
171    async fn take(&self, shell_id: &str, max_output_tokens: Option<usize>) -> ExecSessionResult {
172        let mut output = self.output.lock().await;
173        Self::take_locked(
174            &mut output,
175            shell_id,
176            max_output_tokens,
177            self.process_exited(),
178        )
179    }
180
181    fn take_locked(
182        output: &mut CapturedOutput,
183        shell_id: &str,
184        max_output_tokens: Option<usize>,
185        exit_observed: bool,
186    ) -> ExecSessionResult {
187        // Count captured bytes, including bytes omitted by either cap, rather
188        // than the rendered string. A tiny response cap can hide real output.
189        let had_output = !output.bytes.is_empty() || output.omitted_bytes > 0;
190        let polling = output.completion.is_none().then(|| {
191            output.consecutive_empty_readouts = if had_output {
192                0
193            } else {
194                output.consecutive_empty_readouts.saturating_add(1)
195            };
196            let now = Instant::now();
197            let idle = now.duration_since(output.last_output_at);
198            ExecSessionProgress {
199                terminal_result_ttl_ms: bash_runtime::COMPLETED_SESSION_TTL_SECS * 1000,
200                state: if idle >= QUIET_POLL_ADVISORY_AFTER
201                    && output.consecutive_empty_readouts >= QUIET_POLL_DIAGNOSIS_READOUTS
202                {
203                    "needs_diagnosis"
204                } else {
205                    "waiting"
206                },
207                elapsed_ms: duration_ms(now.duration_since(output.started_at)),
208                output_idle_ms: duration_ms(idle),
209                consecutive_empty_readouts: output.consecutive_empty_readouts,
210                exit_observed,
211                advisory: (idle >= QUIET_POLL_ADVISORY_AFTER
212                    && output.consecutive_empty_readouts >= QUIET_POLL_ADVISORY_READOUTS)
213                    .then_some(QUIET_POLL_ADVISORY),
214            }
215        });
216        let bytes = std::mem::take(&mut output.bytes);
217        let decoded = String::from_utf8_lossy(&bytes);
218        let budget = max_output_tokens
219            .unwrap_or(DEFAULT_OUTPUT_TOKENS)
220            .clamp(1, MAX_OUTPUT_TOKENS)
221            .saturating_mul(4);
222        let end = json_output_prefix_end(&decoded, budget);
223        let omitted = std::mem::take(&mut output.omitted_bytes)
224            .saturating_add(decoded.len().saturating_sub(end));
225        let (status, exit_code) = output.completion.unwrap_or(("running", None));
226        ExecSessionResult {
227            session_id: output.completion.is_none().then(|| shell_id.to_owned()),
228            status,
229            output: decoded[..end].to_owned(),
230            exit_code,
231            output_omitted_bytes: omitted,
232            error: output.error.clone(),
233            polling,
234        }
235    }
236}
237
238/// A reserved, owner-checked observation of an existing exec command. It does
239/// not own the process or advance its output cursor. At most one live monitor
240/// can hold a command; dropping an unstarted or cancelled monitor releases it.
241pub struct BackgroundExecJob {
242    shell: Arc<bash_runtime::ShellSession>,
243}
244
245impl BackgroundExecJob {
246    pub fn id(&self) -> &str {
247        &self.shell.id
248    }
249
250    pub fn session_id(&self) -> &str {
251        self.shell.session_id.as_deref().expect("validated owner")
252    }
253
254    /// Wait for the same fully drained terminal capture used by write_stdin.
255    /// The bounded snapshot leaves the ordinary pull result and TTL untouched.
256    pub async fn completion(&self) -> bamboo_agent_core::BashCompletionInfo {
257        let state = self.shell.polled.as_ref().expect("validated exec capture");
258        loop {
259            state.wait(30_000).await;
260            let output = state.output.lock().await;
261            if output.completion.is_none() {
262                continue;
263            }
264            let mut snapshot = output.clone();
265            drop(output);
266            let result = PolledShellState::take_locked(
267                &mut snapshot,
268                &self.shell.id,
269                Some(1024),
270                state.process_exited(),
271            );
272            return bamboo_agent_core::BashCompletionInfo {
273                session_id: self.session_id().to_owned(),
274                bash_id: self.id().to_owned(),
275                command: self.shell.command.clone(),
276                exit_code: result.exit_code,
277                status: result.status.to_owned(),
278                output_tail: serde_json::to_string(&result).expect("exec result is serializable"),
279            };
280        }
281    }
282}
283
284impl Drop for BackgroundExecJob {
285    fn drop(&mut self) {
286        self.shell
287            .polled
288            .as_ref()
289            .expect("validated exec capture")
290            .background_monitor
291            .store(false, Ordering::Release);
292    }
293}
294
295/// Reserve only an exact handle from this owner's pull-session registry.
296/// Legacy Bash shells and another owner's handles cannot be handed off.
297pub fn reserve_background_job(owner: &str, handle: &str) -> Option<BackgroundExecJob> {
298    let shell = bash_runtime::get_shell(handle)?;
299    if shell.session_id.as_deref() != Some(owner) {
300        return None;
301    }
302    shell
303        .polled
304        .as_ref()?
305        .background_monitor
306        .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
307        .ok()?;
308    Some(BackgroundExecJob { shell })
309}
310
311/// Bound the serialized `output` string, including its JSON quotes and escapes.
312/// Protocol fields such as status and session_id remain outside this body budget.
313fn json_output_prefix_end(output: &str, budget: usize) -> usize {
314    let mut encoded_bytes = 2; // The surrounding JSON string quotes.
315    let mut end = 0;
316    for (offset, character) in output.char_indices() {
317        // Match serde_json's string escaping without allocating an encoded copy.
318        let character_bytes = match character {
319            '"' | '\\' | '\n' | '\r' | '\t' | '\u{8}' | '\u{c}' => 2,
320            '\u{0}'..='\u{1f}' => 6,
321            _ => character.len_utf8(),
322        };
323        if encoded_bytes + character_bytes > budget {
324            break;
325        }
326        encoded_bytes += character_bytes;
327        end = offset + character.len_utf8();
328    }
329    end
330}
331
332/// Fixed-size reads capture prompts without a newline and keep an unterminated
333/// line from growing a BufReader allocation without a bound.
334pub(crate) async fn pump_polled_stream(
335    mut reader: impl AsyncRead + Unpin,
336    state: Arc<PolledShellState>,
337) {
338    let mut buffer = [0; 8192];
339    let mut pending = Vec::new();
340    let mut stop = state.capture_stop.subscribe();
341    loop {
342        let stopped = *stop.borrow();
343        if stopped {
344            if !pending.is_empty() {
345                state
346                    .push(String::from_utf8_lossy(&pending).as_bytes())
347                    .await;
348            }
349            break;
350        }
351        let read = tokio::select! {
352            biased;
353            _ = stop.changed() => continue,
354            result = reader.read(&mut buffer) => result,
355        };
356        match read {
357            Ok(0) => {
358                if !pending.is_empty() {
359                    state
360                        .push(String::from_utf8_lossy(&pending).as_bytes())
361                        .await;
362                }
363                break;
364            }
365            Ok(count) => {
366                pending.extend_from_slice(&buffer[..count]);
367                let mut consumed = 0;
368                let mut decoded = String::new();
369                while consumed < pending.len() {
370                    match std::str::from_utf8(&pending[consumed..]) {
371                        Ok(text) => {
372                            decoded.push_str(text);
373                            consumed = pending.len();
374                        }
375                        Err(error) => {
376                            let valid_end = consumed + error.valid_up_to();
377                            decoded.push_str(
378                                std::str::from_utf8(&pending[consumed..valid_end]).unwrap(),
379                            );
380                            consumed = valid_end;
381                            if let Some(invalid_len) = error.error_len() {
382                                decoded.push('\u{fffd}');
383                                consumed += invalid_len;
384                            } else {
385                                break;
386                            }
387                        }
388                    }
389                }
390                pending.drain(..consumed);
391                state.push(decoded.as_bytes()).await;
392            }
393            Err(error) => {
394                state
395                    .record_error(format!("Shell output read failed: {error}"))
396                    .await;
397                break;
398            }
399        }
400    }
401}
402
403#[derive(Debug, Serialize)]
404struct ExecSessionResult {
405    #[serde(skip_serializing_if = "Option::is_none")]
406    session_id: Option<String>,
407    status: &'static str,
408    output: String,
409    #[serde(skip_serializing_if = "Option::is_none")]
410    exit_code: Option<i32>,
411    #[serde(skip_serializing_if = "is_zero")]
412    output_omitted_bytes: usize,
413    #[serde(skip_serializing_if = "Option::is_none")]
414    error: Option<String>,
415    #[serde(skip_serializing_if = "Option::is_none")]
416    polling: Option<ExecSessionProgress>,
417}
418
419/// Capture observations, never a liveness probe or evidence of a hung process.
420#[derive(Debug, Serialize)]
421struct ExecSessionProgress {
422    terminal_result_ttl_ms: u64,
423    state: &'static str,
424    elapsed_ms: u64,
425    output_idle_ms: u64,
426    consecutive_empty_readouts: usize,
427    // False means no exit has been observed, not that liveness was verified.
428    exit_observed: bool,
429    #[serde(skip_serializing_if = "Option::is_none")]
430    advisory: Option<&'static str>,
431}
432
433fn duration_ms(duration: Duration) -> u64 {
434    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
435}
436
437fn is_zero(value: &usize) -> bool {
438    *value == 0
439}
440
441pub(crate) fn yield_duration(requested: Option<u64>) -> u64 {
442    requested.unwrap_or(10_000).clamp(250, 30_000)
443}
444
445pub(crate) fn owner_session(ctx: &ToolCtx) -> Result<&str, ToolError> {
446    ctx.session_id().filter(|id| !id.is_empty()).ok_or_else(|| {
447        ToolError::InvalidArguments("Exec sessions require an owning session".into())
448    })
449}
450
451pub(crate) async fn poll_result(
452    shell: &bash_runtime::ShellSession,
453    yield_time_ms: u64,
454    max_output_tokens: Option<usize>,
455) -> ToolOutcome {
456    let state = shell.polled.as_ref().expect("validated exec session");
457    state.wait(yield_time_ms).await;
458    let result = state.take(&shell.id, max_output_tokens).await;
459    if result.session_id.is_none() {
460        bash_runtime::remove_shell(&shell.id);
461    }
462    result_outcome(&result)
463}
464
465fn result_outcome(result: &ExecSessionResult) -> ToolOutcome {
466    ToolOutcome::Completed(ToolResult {
467        // A command's nonzero exit is an observed outcome, not tool transport
468        // failure. Models inspect exit_code/status to decide their next step.
469        success: true,
470        result: serde_json::to_string(result).expect("exec result is serializable"),
471        display_preference: Some("Collapsible".into()),
472        images: Vec::new(),
473    })
474}
475
476/// Exists only during the first invocation. The completion owner, not Drop,
477/// performs and certifies cleanup before discarding an unpublished entry.
478struct InitialDelivery<'a> {
479    shell: &'a bash_runtime::ShellSession,
480    delivered: bool,
481}
482
483impl Drop for InitialDelivery<'_> {
484    fn drop(&mut self) {
485        if !self.delivered {
486            self.shell.cancel_initial();
487        }
488    }
489}
490
491async fn poll_initial_result(
492    shell: &bash_runtime::ShellSession,
493    yield_time_ms: u64,
494    max_output_tokens: Option<usize>,
495) -> Result<ToolOutcome, ToolError> {
496    let mut delivery = InitialDelivery {
497        shell,
498        delivered: false,
499    };
500    let state = shell.polled.as_ref().expect("validated initial exec");
501    if let Some(owner) = &shell.initial_handoff {
502        tokio::select! {
503            _ = state.wait(yield_time_ms) => {},
504            _ = owner.capture_ready() => {},
505        }
506    } else {
507        state.wait(yield_time_ms).await;
508    }
509    let result = take_initial_result(shell, max_output_tokens, &mut delivery, || {}).await?;
510    if result.session_id.is_none() {
511        bash_runtime::remove_shell(&shell.id);
512    }
513    Ok(result_outcome(&result))
514}
515
516/// The last await acquires capture, then ownership and its one output snapshot
517/// commit synchronously under that same lock. A writer cannot finish draining
518/// between a stale snapshot and terminal delivery. The observer is a private
519/// scheduling seam; it does not supply ownership readiness or an exit status.
520async fn take_initial_result(
521    shell: &bash_runtime::ShellSession,
522    max_output_tokens: Option<usize>,
523    delivery: &mut InitialDelivery<'_>,
524    before_commit: impl FnOnce(),
525) -> Result<ExecSessionResult, ToolError> {
526    let state = shell.polled.as_ref().expect("validated initial exec");
527    let mut output = state.output.lock().await;
528    before_commit();
529    // Delivery errors must leave the output cursor untouched for cleanup.
530    let terminal = shell.deliver_initial().map_err(ToolError::Execution)?;
531    let mut result = PolledShellState::take_locked(
532        &mut output,
533        &shell.id,
534        max_output_tokens,
535        state.process_exited(),
536    );
537    if let Some(status) = terminal {
538        result.session_id = None;
539        result.polling = None;
540        result.exit_code = status.code();
541        result.status = if result.error.is_some() {
542            "error"
543        } else if status.code().is_none() {
544            "killed"
545        } else {
546            "completed"
547        };
548    } else if shell.initial_handoff.is_some() {
549        // A not-yet-drained initial owner still has a retrievable result handle.
550        result.session_id = Some(shell.id.clone());
551        result.status = "running";
552        result.exit_code = None;
553    }
554    delivery.delivered = true;
555    drop(output);
556    Ok(result)
557}
558
559#[derive(Deserialize)]
560#[serde(deny_unknown_fields)]
561struct ExecCommandArgs {
562    cmd: String,
563    workdir: Option<String>,
564    yield_time_ms: Option<u64>,
565    max_output_tokens: Option<usize>,
566    #[serde(default)]
567    interactive: bool,
568}
569
570fn resolve_cwd(owner: &str, workdir: Option<&str>) -> Result<PathBuf, ToolError> {
571    let workspace = workspace_state::workspace_or_process_cwd(Some(owner));
572    let workspace = workspace.canonicalize().map_err(|error| {
573        ToolError::InvalidArguments(format!(
574            "Invalid session workspace '{}': {error}",
575            workspace.display()
576        ))
577    })?;
578    let requested = match workdir {
579        Some(value) if value.trim().is_empty() => {
580            return Err(ToolError::InvalidArguments(
581                "'workdir' cannot be empty".into(),
582            ));
583        }
584        Some(value) => Path::new(value.trim()).to_path_buf(),
585        None => workspace.clone(),
586    };
587    let resolved = if requested.is_absolute() {
588        requested
589    } else {
590        workspace.join(requested)
591    };
592    let canonical = resolved.canonicalize().map_err(|error| {
593        ToolError::InvalidArguments(format!("Invalid workdir '{}': {error}", resolved.display()))
594    })?;
595    if !canonical.is_dir() {
596        return Err(ToolError::InvalidArguments(
597            "workdir must be a directory".into(),
598        ));
599    }
600    if !canonical.starts_with(&workspace) {
601        return Err(ToolError::InvalidArguments(
602            "workdir must stay within the current session workspace".into(),
603        ));
604    }
605    Ok(canonical)
606}
607
608#[derive(Default)]
609pub struct ExecCommandTool;
610
611impl ExecCommandTool {
612    pub fn new() -> Self {
613        Self
614    }
615}
616
617#[async_trait]
618impl Tool for ExecCommandTool {
619    fn name(&self) -> &str {
620        "exec_command"
621    }
622
623    fn description(&self) -> &str {
624        "Execute a shell command and wait up to yield_time_ms. A running command returns a session_id and polling progress; continue other work, then call write_stdin with that ID and empty chars to retrieve incremental output and its final exit code. Follow a quiet polling advisory before repeating empty polls; silence does not prove completion. Completion is delivered through these tool results. interactive=true keeps a stdin pipe open for later input; this is a pipe, not a terminal/PTY. Output is bounded and reports omitted bytes."
625    }
626
627    fn parameters_schema(&self) -> serde_json::Value {
628        json!({"type":"object","properties":{
629            "cmd":{"type":"string","description":"Shell command to execute"},
630            "workdir":{"type":"string","description":"Working directory within the current session workspace, relative or absolute; paths and symlinks outside the workspace are rejected"},
631            "yield_time_ms":{"type":"integer","description":"Wait before yielding a running session; default 10000, clamped to 250..30000 ms"},
632            "max_output_tokens":{"type":"integer","description":"Output string budget, default 2000 and capped at 8192; estimated as four JSON-encoded bytes per token, including string quotes and escapes. Status/session/exit metadata is additional"},
633            "interactive":{"type":"boolean","description":"Keep a piped stdin open for write_stdin; default false sends immediate EOF"}
634        },"required":["cmd"],"additionalProperties":false})
635    }
636
637    async fn invoke(
638        &self,
639        args: serde_json::Value,
640        ctx: ToolCtx,
641    ) -> Result<ToolOutcome, ToolError> {
642        let parsed: ExecCommandArgs = serde_json::from_value(args).map_err(|error| {
643            ToolError::InvalidArguments(format!("Invalid exec_command args: {error}"))
644        })?;
645        let owner = owner_session(&ctx)?;
646        if parsed.cmd.trim().is_empty() || parsed.cmd.len() > MAX_COMMAND_BYTES {
647            return Err(ToolError::InvalidArguments(
648                "cmd must be nonempty and at most 65536 bytes".into(),
649            ));
650        }
651        let cwd = resolve_cwd(owner, parsed.workdir.as_deref())?;
652        let state = PolledShellState::new()?;
653        let shell = bash_runtime::spawn_polled(
654            &parsed.cmd,
655            &cwd,
656            None,
657            owner.to_owned(),
658            parsed.interactive,
659            state,
660        )
661        .await
662        .map_err(ToolError::Execution)?;
663        poll_initial_result(
664            &shell,
665            yield_duration(parsed.yield_time_ms),
666            parsed.max_output_tokens,
667        )
668        .await
669    }
670}
671
672#[cfg(test)]
673#[path = "exec_sessions_tests.rs"]
674mod tests;
675
676#[cfg(all(test, unix))]
677#[path = "initial_exec_cleanup_tests.rs"]
678mod initial_cleanup_tests;
679
680#[cfg(test)]
681#[path = "background_exec_tests.rs"]
682mod background_tests;
bamboo_tools/tools/exec_command.rs

bamboo_tools/tools/
exec_command.rs
