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}