Skip to main content

cli_agents/adapters/
mod.rs

1mod claude;
2mod codex;
3mod gemini;
4
5pub use claude::ClaudeAdapter;
6pub use codex::CodexAdapter;
7pub use gemini::GeminiAdapter;
8
9use crate::error::{Error, Result};
10use crate::events::StreamEvent;
11use crate::types::{CliName, RunOptions, RunResult};
12#[cfg(windows)]
13use process_wrap::tokio::JobObject;
14#[cfg(unix)]
15use process_wrap::tokio::ProcessGroup;
16use process_wrap::tokio::TokioChildWrapper;
17use process_wrap::tokio::TokioCommandWrap;
18use std::collections::HashMap;
19use tokio::io::{AsyncBufReadExt, BufReader};
20use tracing::{debug, warn};
21
22/// Trait implemented by each CLI adapter.
23pub trait CliAdapter: Send + Sync {
24    fn name(&self) -> CliName;
25
26    fn run(
27        &self,
28        opts: &RunOptions,
29        emit: &(dyn Fn(StreamEvent) + Send + Sync),
30        cancel: tokio_util::sync::CancellationToken,
31    ) -> impl std::future::Future<Output = crate::error::Result<RunResult>> + Send;
32}
33
34/// Get the adapter for a given CLI.
35pub(crate) fn get_adapter(cli: CliName) -> Box<dyn CliAdapterBoxed> {
36    match cli {
37        CliName::Claude => Box::new(ClaudeAdapter),
38        CliName::Codex => Box::new(CodexAdapter),
39        CliName::Gemini => Box::new(GeminiAdapter),
40    }
41}
42
43/// Object-safe version of [`CliAdapter`] for dynamic dispatch.
44///
45/// Needed because `CliAdapter::run` uses RPITIT (`impl Future`), which makes
46/// the trait non-object-safe. This wrapper boxes the future for `dyn` dispatch.
47/// The blanket impl below bridges the two automatically.
48#[allow(dead_code)]
49pub(crate) trait CliAdapterBoxed: Send + Sync {
50    fn name(&self) -> CliName;
51
52    fn run_boxed<'a>(
53        &'a self,
54        opts: &'a RunOptions,
55        emit: &'a (dyn Fn(StreamEvent) + Send + Sync),
56        cancel: tokio_util::sync::CancellationToken,
57    ) -> std::pin::Pin<
58        Box<dyn std::future::Future<Output = crate::error::Result<RunResult>> + Send + 'a>,
59    >;
60}
61
62impl<T: CliAdapter> CliAdapterBoxed for T {
63    fn name(&self) -> CliName {
64        CliAdapter::name(self)
65    }
66
67    fn run_boxed<'a>(
68        &'a self,
69        opts: &'a RunOptions,
70        emit: &'a (dyn Fn(StreamEvent) + Send + Sync),
71        cancel: tokio_util::sync::CancellationToken,
72    ) -> std::pin::Pin<
73        Box<dyn std::future::Future<Output = crate::error::Result<RunResult>> + Send + 'a>,
74    > {
75        Box::pin(self.run(opts, emit, cancel))
76    }
77}
78
79// ── Shared subprocess infrastructure ──
80
81/// Outcome of a spawned CLI process.
82pub(crate) enum SpawnOutcome {
83    /// Process reached a terminal state on its own (exited, or was signalled).
84    Done {
85        /// The process's exit status, or `None` when a SIGNAL ended it instead.
86        ///
87        /// A signalled process genuinely has no exit code. Substituting one
88        /// makes an out-of-memory kill indistinguishable from an agent that
89        /// cleanly decided it had failed, and [`RunResult::exit_code`] is an
90        /// `Option` precisely so a caller can tell the two apart.
91        exit_code: Option<i32>,
92        /// The signal that terminated the process, when one did. Unix only;
93        /// always `None` elsewhere.
94        signal: Option<i32>,
95        stderr: Option<String>,
96        /// How many stdout lines exceeded `max_bytes` and were dropped.
97        ///
98        /// A dropped line is a lost EVENT, not a lost run — adapters surface
99        /// this as a warning so the loss is never silent.
100        dropped_lines: u64,
101    },
102    /// Process was cancelled via the cancellation token.
103    Cancelled,
104}
105
106/// Parameters for [`spawn_and_stream`].
107pub(crate) struct SpawnParams<'a> {
108    pub cli_label: &'a str,
109    pub binary: &'a str,
110    pub args: &'a [String],
111    pub extra_env: &'a HashMap<String, String>,
112    /// Keys to remove from the inherited parent env before applying `extra_env`.
113    /// Used to prevent leaks like `ANTHROPIC_API_KEY` overriding subscription auth.
114    pub strip_env: &'a [&'static str],
115    pub cwd: &'a str,
116    pub max_bytes: usize,
117    pub cancel: &'a tokio_util::sync::CancellationToken,
118}
119
120/// Spawn a CLI subprocess and stream its stdout line-by-line.
121///
122/// Handles the boilerplate shared across all adapters: process spawning,
123/// stdout buffering with size limits, stderr collection, and cancellation.
124/// Does **not** clone the parent process environment — `Command` inherits it
125/// automatically; only `extra_env` entries are added.
126pub(crate) async fn spawn_and_stream(
127    params: SpawnParams<'_>,
128    mut on_line: impl FnMut(&str) + Send,
129) -> Result<SpawnOutcome> {
130    let SpawnParams {
131        cli_label,
132        binary,
133        args,
134        extra_env,
135        strip_env,
136        cwd,
137        max_bytes,
138        cancel,
139    } = params;
140    debug!(cli = cli_label, binary = %binary, args = ?args, "spawning CLI");
141
142    // ── The child owns a KILL GROUP, on every platform ──
143    //
144    // Cancelling a run has to take the whole tree, not just the process we
145    // spawned: `claude` is a launcher, and the work happens in node processes
146    // below it. Killing only the parent orphans those — they keep running, keep
147    // holding the model session, and keep writing to a pipe nobody reads.
148    //
149    // This used to be `pre_exec(setpgid)` plus `libc::killpg(SIGKILL)`, which is
150    // correct on unix and does not exist on Windows — where the equivalent is a
151    // Job Object, a completely different mechanism with the same purpose.
152    // `process-wrap` is that difference, already written and tested: the unix
153    // arm is the same process-group call, and the Windows arm assigns the child
154    // to a job that dies with it.
155    let mut wrap = TokioCommandWrap::with_new(binary, |cmd| {
156        cmd.args(args);
157        for key in strip_env {
158            cmd.env_remove(key);
159        }
160        cmd.envs(extra_env)
161            .current_dir(cwd)
162            .stdin(std::process::Stdio::null())
163            .stdout(std::process::Stdio::piped())
164            .stderr(std::process::Stdio::piped())
165            .kill_on_drop(true);
166
167        // ── AND IT OPENS NO CONSOLE WINDOW ──
168        //
169        // Windows gives every console-subsystem child its own console window
170        // unless the parent passes CREATE_NO_WINDOW at creation. `claude`,
171        // `codex` and `gemini` are all console programs, so a GUI application
172        // embedding this crate flashes a black terminal on every run — reported
173        // against a Tauri app, where it appears over the user's editor.
174        //
175        // A LIBRARY CANNOT LEAVE THIS TO ITS CALLER. The flag is only honoured
176        // when passed to CreateProcess, which happens inside this closure;
177        // nothing the caller holds afterwards can set it.
178        //
179        // Orthogonal to the JobObject below: that governs how the tree DIES,
180        // this governs whether it is ever VISIBLE.
181        // No `CommandExt` import: tokio's Command exposes `creation_flags`
182        // directly under cfg(windows), and importing the std trait as well is
183        // an unused-import warning.
184        #[cfg(windows)]
185        {
186            const CREATE_NO_WINDOW: u32 = 0x0800_0000;
187            cmd.creation_flags(CREATE_NO_WINDOW);
188        }
189    });
190    #[cfg(unix)]
191    wrap.wrap(ProcessGroup::leader());
192    #[cfg(windows)]
193    wrap.wrap(JobObject);
194
195    let mut child = wrap
196        .spawn()
197        .map_err(|e| Error::Process(format!("failed to spawn {cli_label}: {e}")))?;
198
199    let stdout = child.stdout().take().expect("stdout piped");
200    let stderr = child.stderr().take().expect("stderr piped");
201
202    // stderr is retained as a bounded TAIL. It exists so a failed run can be
203    // explained (`extract_error_message` reads it), and the parting words are
204    // at the end — a chatty CLI logging megabytes must not be held in full
205    // for that. Chunked reads, not lines: a single unterminated line cannot
206    // grow the buffer past the bound either.
207    let stderr_handle = tokio::spawn(async move {
208        use tokio::io::AsyncReadExt;
209        let mut stderr = stderr;
210        let mut buf: Vec<u8> = Vec::new();
211        let mut chunk = [0u8; 8192];
212        loop {
213            match stderr.read(&mut chunk).await {
214                Ok(0) | Err(_) => break,
215                Ok(n) => {
216                    buf.extend_from_slice(&chunk[..n]);
217                    if buf.len() > STDERR_TAIL_BYTES * 2 {
218                        buf.drain(..buf.len() - STDERR_TAIL_BYTES);
219                    }
220                }
221            }
222        }
223        String::from_utf8_lossy(&buf).into_owned()
224    });
225
226    // `max_bytes` bounds what a single LINE may retain — not cumulative
227    // throughput. Every line is handed to `on_line` and released, so the
228    // total streamed volume never lives in memory; a cumulative cap here
229    // used to KILL a healthy long run at the 10MB mark and discard the
230    // whole turn it was carrying. A line that will not fit is consumed to
231    // its newline, dropped, and counted; the run continues.
232    let mut reader = BufReader::new(stdout);
233    let mut line_buf: Vec<u8> = Vec::new();
234    let mut dropped_lines: u64 = 0;
235
236    loop {
237        line_buf.clear();
238        tokio::select! {
239            result = read_line_capped(&mut reader, &mut line_buf, max_bytes) => {
240                match result {
241                    Ok(CappedLine::Eof) => break,
242                    Ok(CappedLine::Line { dropped: true }) => {
243                        dropped_lines += 1;
244                        warn!(cli = cli_label, max_bytes, "dropped a stdout line larger than the retention cap");
245                    }
246                    Ok(CappedLine::Line { dropped: false }) => {
247                        on_line(String::from_utf8_lossy(&line_buf).trim());
248                    }
249                    Err(e) => {
250                        warn!(cli = cli_label, error = %e, "error reading stdout");
251                        break;
252                    }
253                }
254            }
255            _ = cancel.cancelled() => {
256                kill_process_group(&mut child).await;
257                return Ok(SpawnOutcome::Cancelled);
258            }
259        }
260    }
261
262    let status = Box::into_pin(child.wait()).await.map_err(Error::Io)?;
263    // `code()` is `None` for a signalled process. This used to be
264    // `.unwrap_or(1)`, which reported a SIGKILL as a clean `exit 1` and left
265    // callers with no way to recover the difference — the exact ambiguity that
266    // sent a downstream app hunting for an error message a killed process never
267    // wrote. Report what actually happened and let the caller decide.
268    let exit_code = status.code();
269    #[cfg(unix)]
270    let signal = std::os::unix::process::ExitStatusExt::signal(&status);
271    #[cfg(not(unix))]
272    let signal: Option<i32> = None;
273    let stderr_text = stderr_handle.await.unwrap_or_default();
274
275    Ok(SpawnOutcome::Done {
276        exit_code,
277        signal,
278        stderr: if stderr_text.is_empty() {
279            None
280        } else {
281            Some(stderr_text)
282        },
283        dropped_lines,
284    })
285}
286
287/// How much stderr is retained (as a tail — see the reader above).
288const STDERR_TAIL_BYTES: usize = 64 * 1024;
289
290/// Outcome of one capped line read.
291enum CappedLine {
292    /// A complete line is in the buffer — unless `dropped`, in which case the
293    /// line exceeded the cap, the buffer is empty, and the line is gone.
294    Line { dropped: bool },
295    /// End of stream.
296    Eof,
297}
298
299/// Read one `\n`-terminated line, retaining at most `cap` bytes of it.
300///
301/// A line that will not fit is not truncated-and-delivered — a cut JSONL
302/// event is garbage to every parser downstream — it is consumed to its
303/// newline, discarded, and reported as `dropped`. Memory stays bounded by
304/// `cap` plus the reader's own buffer, no matter what the child writes.
305async fn read_line_capped<R: tokio::io::AsyncBufRead + Unpin>(
306    reader: &mut R,
307    buf: &mut Vec<u8>,
308    cap: usize,
309) -> std::io::Result<CappedLine> {
310    let mut dropped = false;
311    loop {
312        let (consumed, line_complete) = {
313            let available = reader.fill_buf().await?;
314            if available.is_empty() {
315                // EOF — a final unterminated line still counts as a line.
316                return Ok(if buf.is_empty() && !dropped {
317                    CappedLine::Eof
318                } else {
319                    CappedLine::Line { dropped }
320                });
321            }
322            match available.iter().position(|&b| b == b'\n') {
323                Some(newline) => {
324                    if !dropped {
325                        if buf.len() + newline <= cap {
326                            buf.extend_from_slice(&available[..newline]);
327                        } else {
328                            dropped = true;
329                            buf.clear();
330                        }
331                    }
332                    (newline + 1, true)
333                }
334                None => {
335                    let n = available.len();
336                    if !dropped {
337                        if buf.len() + n <= cap {
338                            buf.extend_from_slice(available);
339                        } else {
340                            dropped = true;
341                            buf.clear();
342                        }
343                    }
344                    (n, false)
345                }
346            }
347        };
348        reader.consume(consumed);
349        if line_complete {
350            return Ok(CappedLine::Line { dropped });
351        }
352    }
353}
354
355/// The loss a dropped line represents must reach the consumer, not just the
356/// log — every adapter calls this after its spawn completes.
357pub(crate) fn warn_dropped_lines(dropped_lines: u64, max_bytes: usize, emit: &dyn Fn(StreamEvent)) {
358    if dropped_lines > 0 {
359        emit(StreamEvent::Error {
360            message: format!(
361                "{dropped_lines} output line(s) exceeded the {max_bytes}-byte retention cap and were dropped"
362            ),
363            severity: Some(crate::events::Severity::Warning),
364        });
365    }
366}
367
368/// A sentence for a process that died without writing one.
369///
370/// A signalled CLI usually produces NO stderr and no result event — there was
371/// no chance to. Without this the only fact reaching the user is a number, and
372/// the most common case by far (the OS reclaiming memory) reads as an
373/// unexplained failure.
374pub(crate) fn describe_signal(signal: Option<i32>) -> Option<String> {
375    let sig = signal?;
376    Some(match sig {
377        2 => "The agent was interrupted (SIGINT).".to_string(),
378        6 => "The agent aborted (SIGABRT).".to_string(),
379        9 => "The agent was killed (SIGKILL), most often by the system reclaiming memory."
380            .to_string(),
381        11 => "The agent crashed (SIGSEGV).".to_string(),
382        15 => "The agent was terminated (SIGTERM).".to_string(),
383        other => format!("The agent was terminated by signal {other}."),
384    })
385}
386
387/// Extract a user-friendly error message from CLI stderr.
388/// When an agent fails with no text output, this provides something
389/// meaningful to show the user instead of a blank response.
390pub(crate) fn extract_error_message(stderr: Option<&str>) -> Option<String> {
391    let stderr = stderr?;
392    // Find the most informative error line.
393    let msg = stderr
394        .lines()
395        .filter(|l| !l.is_empty())
396        .find(|l| {
397            let lower = l.to_lowercase();
398            lower.contains("error")
399                || lower.contains("limit")
400                || lower.contains("failed")
401                || lower.contains("denied")
402                || lower.contains("unauthorized")
403        })
404        .or_else(|| stderr.lines().rfind(|l| !l.is_empty()));
405    msg.map(|s| s.trim().to_string())
406}
407
408/// Kill the child AND everything it spawned.
409///
410/// `TokioChildWrapper::kill` dispatches to whichever group mechanism was wrapped
411/// on at spawn — the process group on unix, the Job Object on Windows — so the
412/// `#[cfg]` that used to live here is gone. It returns a boxed future, hence the
413/// pin.
414async fn kill_process_group(child: &mut Box<dyn TokioChildWrapper>) {
415    let _ = Box::into_pin(child.kill()).await;
416}
417
418#[cfg(test)]
419mod tests {
420    use super::*;
421
422    /// CANCELLING TAKES THE WHOLE TREE, not just the process we spawned.
423    ///
424    /// This is the contract `setpgid`/`killpg` existed to provide, and it had no
425    /// test — so the swap to `process-wrap` would have been unverifiable, and so
426    /// would any future change to it. It matters because `claude` is a
427    /// launcher: the work runs in node processes underneath. Killing only the
428    /// parent leaves those alive, holding a model session, writing to a pipe
429    /// nobody is reading.
430    ///
431    /// HOW IT PROVES IT WITHOUT TIMING GAMES: the shell writes a marker file,
432    /// spawns a grandchild that would DELETE that file after a delay, then
433    /// sleeps. Cancel immediately. If the group died, the grandchild never runs
434    /// and the marker survives. If only the parent died, the orphan wakes up and
435    /// removes it. The assertion is on a filesystem fact, not on a pid still
436    /// being enumerable, which is what makes it honest on both platforms.
437    ///
438    /// Unix-only for now: it needs a shell that can background a process, and
439    /// the Windows equivalent (`cmd /c start`) has different semantics worth
440    /// writing deliberately rather than transliterating. The Job Object path is
441    /// exercised by CI compiling this file for Windows; that it KILLS the tree
442    /// there is not yet proved. Marked plainly rather than assumed.
443    #[cfg(unix)]
444    #[tokio::test]
445    async fn cancelling_kills_the_grandchild_not_just_the_child() {
446        let dir = tempfile::tempdir().unwrap();
447        let marker = dir.path().join("survivor");
448        std::fs::write(&marker, "alive").unwrap();
449
450        // Grandchild removes the marker after 3s; parent then sleeps 10s.
451        let script = format!("(sleep 3; rm -f '{}') & sleep 10", marker.display());
452        let args = vec!["-c".to_string(), script];
453        let cancel = tokio_util::sync::CancellationToken::new();
454
455        let token = cancel.clone();
456        tokio::spawn(async move {
457            tokio::time::sleep(std::time::Duration::from_millis(300)).await;
458            token.cancel();
459        });
460
461        let outcome = spawn_and_stream(
462            SpawnParams {
463                cli_label: "test",
464                binary: "sh",
465                args: &args,
466                extra_env: &HashMap::new(),
467                strip_env: &[],
468                cwd: dir.path().to_str().unwrap(),
469                max_bytes: 1024,
470                cancel: &cancel,
471            },
472            |_: &str| {},
473        )
474        .await
475        .expect("spawn");
476
477        assert!(
478            matches!(outcome, SpawnOutcome::Cancelled),
479            "run was cancelled"
480        );
481
482        // Past when the grandchild would have deleted it, had it survived.
483        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
484        assert!(
485            marker.exists(),
486            "the grandchild outlived cancellation and deleted the marker — the kill did not reach the process group"
487        );
488    }
489
490    /// A process that ends by SIGNAL has no exit code, and says so.
491    ///
492    /// THE REGRESSION THIS PINS. `spawn_and_stream` used to finish with
493    /// `status.code().unwrap_or(1)`, so a killed process was reported as a
494    /// clean `exit 1`. Downstream that is unrecoverable: an out-of-memory kill
495    /// and an agent that decided it had failed become the same fact, and a
496    /// consumer looking for the reason searches a stderr the process never got
497    /// to write. `sh -c 'kill -9 $$'` reproduces it without timing games — the
498    /// shell signals itself, so the outcome is deterministic.
499    #[cfg(unix)]
500    #[tokio::test]
501    async fn a_signalled_process_reports_the_signal_not_a_fabricated_exit_code() {
502        let args = vec!["-c".to_string(), "kill -9 $$".to_string()];
503        let cancel = tokio_util::sync::CancellationToken::new();
504        let outcome = spawn_and_stream(
505            SpawnParams {
506                cli_label: "test",
507                binary: "sh",
508                args: &args,
509                extra_env: &HashMap::new(),
510                strip_env: &[],
511                cwd: ".",
512                max_bytes: 1024,
513                cancel: &cancel,
514            },
515            |_| {},
516        )
517        .await
518        .expect("spawn should succeed");
519
520        match outcome {
521            SpawnOutcome::Done {
522                exit_code, signal, ..
523            } => {
524                assert_eq!(exit_code, None, "a signalled process has no exit code");
525                assert_eq!(signal, Some(9), "SIGKILL should be reported as itself");
526            }
527            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
528        }
529    }
530
531    /// An ordinary non-zero exit still reports its code — the change above must
532    /// not turn every failure into `None`.
533    #[tokio::test]
534    async fn a_normal_exit_still_reports_its_code() {
535        let args = vec!["-c".to_string(), "exit 3".to_string()];
536        let cancel = tokio_util::sync::CancellationToken::new();
537        let outcome = spawn_and_stream(
538            SpawnParams {
539                cli_label: "test",
540                binary: "sh",
541                args: &args,
542                extra_env: &HashMap::new(),
543                strip_env: &[],
544                cwd: ".",
545                max_bytes: 1024,
546                cancel: &cancel,
547            },
548            |_| {},
549        )
550        .await
551        .expect("spawn should succeed");
552
553        match outcome {
554            SpawnOutcome::Done {
555                exit_code, signal, ..
556            } => {
557                assert_eq!(exit_code, Some(3));
558                assert_eq!(signal, None, "an ordinary exit was not signalled");
559            }
560            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
561        }
562    }
563
564    /// The user-facing sentence for the case that writes no stderr at all.
565    #[test]
566    fn describe_signal_names_the_common_kills() {
567        assert!(describe_signal(Some(9)).unwrap().contains("SIGKILL"));
568        assert!(describe_signal(Some(9)).unwrap().contains("memory"));
569        assert!(describe_signal(Some(15)).unwrap().contains("SIGTERM"));
570        assert!(describe_signal(Some(42)).unwrap().contains("42"));
571        assert_eq!(describe_signal(None), None);
572    }
573
574    /// A LONG RUN IS NOT AN ERROR. `max_bytes` used to count cumulative
575    /// throughput and KILL the process when the total crossed it — but every
576    /// line is handed to `on_line` and dropped, so the total was never held in
577    /// memory at all. A 30-minute agent run streaming tens of MB of tool
578    /// events died at the 10MB mark with its entire turn discarded, which is
579    /// how CueFrame's Director lost long turns in production. The cap bounds
580    /// what a single line may RETAIN; the run itself must complete.
581    #[cfg(unix)]
582    #[tokio::test]
583    async fn total_output_beyond_max_bytes_streams_through_and_completes() {
584        // 200 lines × ~100 bytes ≈ 20 KB through a 1 KB cap.
585        let script = "i=0; while [ $i -lt 200 ]; do printf '%0100d\\n' $i; i=$((i+1)); done";
586        let args = vec!["-c".to_string(), script.to_string()];
587        let cancel = tokio_util::sync::CancellationToken::new();
588        let mut lines = 0u32;
589        let outcome = spawn_and_stream(
590            SpawnParams {
591                cli_label: "test",
592                binary: "sh",
593                args: &args,
594                extra_env: &HashMap::new(),
595                strip_env: &[],
596                cwd: ".",
597                max_bytes: 1024,
598                cancel: &cancel,
599            },
600            |_| lines += 1,
601        )
602        .await
603        .expect("a large-but-line-bounded run must not be an error");
604
605        match outcome {
606            SpawnOutcome::Done { exit_code, .. } => assert_eq!(exit_code, Some(0)),
607            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
608        }
609        assert_eq!(lines, 200, "every line was streamed through");
610    }
611
612    /// One oversized line loses ITSELF, not the run. The line that cannot be
613    /// retained within `max_bytes` is dropped (a truncated JSON event would be
614    /// garbage anyway); the lines after it still arrive and the process still
615    /// reports its own exit.
616    #[cfg(unix)]
617    #[tokio::test]
618    async fn an_oversized_line_is_dropped_and_the_run_continues() {
619        let script = "echo before; printf '%05000d\\n' 7; echo after";
620        let args = vec!["-c".to_string(), script.to_string()];
621        let cancel = tokio_util::sync::CancellationToken::new();
622        let mut seen: Vec<String> = Vec::new();
623        let outcome = spawn_and_stream(
624            SpawnParams {
625                cli_label: "test",
626                binary: "sh",
627                args: &args,
628                extra_env: &HashMap::new(),
629                strip_env: &[],
630                cwd: ".",
631                max_bytes: 1024,
632                cancel: &cancel,
633            },
634            |l| seen.push(l.to_string()),
635        )
636        .await
637        .expect("an oversized line must not abort the run");
638
639        assert_eq!(seen, vec!["before".to_string(), "after".to_string()]);
640        match outcome {
641            SpawnOutcome::Done {
642                exit_code,
643                dropped_lines,
644                ..
645            } => {
646                assert_eq!(exit_code, Some(0));
647                assert_eq!(dropped_lines, 1, "the loss is counted, never silent");
648            }
649            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
650        }
651    }
652
653    /// stderr retention is a TAIL, not the whole stream. It exists so
654    /// `extract_error_message` has the CLI's parting words; a chatty process
655    /// logging megabytes to stderr must not be held in full for that.
656    #[cfg(unix)]
657    #[tokio::test]
658    async fn stderr_retains_a_bounded_tail() {
659        // ~1 MB of filler, then the line that matters, all on stderr.
660        let script = "i=0; while [ $i -lt 10000 ]; do printf '%0100d\\n' $i 1>&2; i=$((i+1)); done; \
661             echo 'Error: the last words' 1>&2; exit 1";
662        let args = vec!["-c".to_string(), script.to_string()];
663        let cancel = tokio_util::sync::CancellationToken::new();
664        let outcome = spawn_and_stream(
665            SpawnParams {
666                cli_label: "test",
667                binary: "sh",
668                args: &args,
669                extra_env: &HashMap::new(),
670                strip_env: &[],
671                cwd: ".",
672                max_bytes: 1024,
673                cancel: &cancel,
674            },
675            |_| {},
676        )
677        .await
678        .expect("spawn should succeed");
679
680        match outcome {
681            SpawnOutcome::Done { stderr, .. } => {
682                let stderr = stderr.expect("stderr was written");
683                assert!(
684                    stderr.len() <= 256 * 1024,
685                    "stderr retention must be bounded, got {} bytes",
686                    stderr.len()
687                );
688                assert!(
689                    stderr.contains("the last words"),
690                    "the tail is the part that explains the failure"
691                );
692            }
693            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
694        }
695    }
696}