Skip to main content

lean_ctx/shell/exec/
process.rs

1use std::io::Read;
2use std::process::{Child, Output};
3use std::sync::Arc;
4use std::sync::atomic::{AtomicBool, Ordering};
5
6/// Wait for a child process with output-size and time limits.
7/// Kills the process if either limit is exceeded, returning what was
8/// captured so far. Prevents unbounded memory growth on commands that
9/// produce massive output (e.g. `rg -i "pattern"` over a large tree).
10///
11/// `kill_group` (Unix): the child was spawned into its own process group
12/// (`process_group(0)`), so a timeout kill signals the whole group. Killing
13/// only the direct child (a shell) leaves orphaned grandchildren holding the
14/// stdout/stderr pipe write ends — the reader threads then never see EOF and
15/// the join below blocks forever, wedging the caller *despite* the timeout
16/// having fired (GH #720: an orphaned `rg` kept a Cursor shell session dead
17/// for hours).
18pub struct LimitedProcessOutput {
19    pub output: Output,
20    pub completeness: crate::core::context_admission::OutputCompleteness,
21}
22
23/// The output alone, for tests. Product callers take the completeness with
24/// it (`wait_with_limits_captured`): an incomplete capture is a decision input.
25#[cfg(test)]
26pub(crate) fn wait_with_limits(
27    child: Child,
28    max_bytes: usize,
29    timeout: std::time::Duration,
30    kill_group: bool,
31) -> Output {
32    wait_with_limits_captured(child, max_bytes, timeout, kill_group).output
33}
34
35pub fn wait_with_limits_captured(
36    mut child: Child,
37    max_bytes: usize,
38    timeout: std::time::Duration,
39    kill_group: bool,
40) -> LimitedProcessOutput {
41    const STDERR_LIMIT: usize = 512 * 1024;
42    /// How long the readers may take to see end-of-file after the child
43    /// exited before its own group is signalled.
44    const DRAIN_GRACE: std::time::Duration = std::time::Duration::from_millis(250);
45    /// How long the readers may take to finish once the child is gone.
46    const READER_BUDGET: std::time::Duration = std::time::Duration::from_secs(2);
47    /// How long an exited or killed child may take to become reapable.
48    const REAP_BUDGET: std::time::Duration = std::time::Duration::from_secs(2);
49
50    // Decided once, before anything is reaped: only a group the child leads
51    // itself is ever signalled. A child that inherited its group (the
52    // caller's, the host's) is killed alone.
53    let own_group = owned_group(&mut child, kill_group);
54
55    let stdout_pipe = child.stdout.take();
56    let stderr_pipe = child.stderr.take();
57    let start = std::time::Instant::now();
58    let truncated = Arc::new(AtomicBool::new(false));
59    let read_failed = Arc::new(AtomicBool::new(false));
60
61    // Each reader appends what it reads to a buffer this thread can read at
62    // any time: a reader that is abandoned below still hands over everything
63    // it had read by then.
64    let stdout_capture = Arc::new(StreamCapture::default());
65    let stderr_capture = Arc::new(StreamCapture::default());
66    let stdout_handle = std::thread::spawn({
67        let (capture, truncated, read_failed) = (
68            Arc::clone(&stdout_capture),
69            Arc::clone(&truncated),
70            Arc::clone(&read_failed),
71        );
72        move || read_stream(stdout_pipe, max_bytes, &capture, &truncated, &read_failed)
73    });
74    let stderr_handle = std::thread::spawn({
75        let (capture, truncated, read_failed) = (
76            Arc::clone(&stderr_capture),
77            Arc::clone(&truncated),
78            Arc::clone(&read_failed),
79        );
80        move || {
81            read_stream(
82                stderr_pipe,
83                STDERR_LIMIT,
84                &capture,
85                &truncated,
86                &read_failed,
87            );
88        }
89    });
90
91    let mut timed_out = false;
92    let mut wait_failed = false;
93    let mut kill_failed = false;
94    let mut forced_pipe_cleanup = false;
95    loop {
96        let hit_timeout = start.elapsed() > timeout;
97        if hit_timeout || truncated.load(Ordering::Relaxed) {
98            // Not reaped yet: the leader still pins its group id.
99            kill_failed |= !kill_child(&mut child, own_group);
100            timed_out = hit_timeout;
101            break;
102        }
103        match exited_unreaped(&mut child) {
104            Ok(true) => {
105                // The leader has exited but is deliberately not reaped here:
106                // as a zombie it keeps its pid, and with it the id of its own
107                // group, from being reused while descendants that still hold
108                // the pipes are signalled.
109                if own_group.is_some()
110                    && !readers_finish_within(&stdout_handle, &stderr_handle, DRAIN_GRACE)
111                {
112                    kill_failed |= !kill_child(&mut child, own_group);
113                    forced_pipe_cleanup = true;
114                }
115                break;
116            }
117            Err(_) => {
118                wait_failed = true;
119                kill_failed |= !kill_child(&mut child, own_group);
120                break;
121            }
122            Ok(false) => std::thread::sleep(std::time::Duration::from_millis(50)),
123        }
124    }
125
126    // The child has exited or was killed above; reaping it is bounded all the
127    // same. A child that cannot be reaped (a kill that did not take effect)
128    // must not hold this call: it is reported as an incomplete capture with a
129    // failure status instead.
130    let reaped = reap_within(&mut child, REAP_BUDGET);
131
132    // A descendant outside any group this call owns may keep a pipe open for
133    // as long as it likes. The readers are given a budget; a reader that does
134    // not finish is left behind — never joined, never waited for again — and
135    // the capture is reported as incomplete. What it had read up to this
136    // moment is still returned: on each stream a prefix of what was written
137    // to it, within that stream's limit.
138    let readers_finished = readers_finish_within(&stdout_handle, &stderr_handle, READER_BUDGET);
139    let reader_join_failed =
140        !readers_finished || stdout_handle.join().is_err() || stderr_handle.join().is_err();
141    let (mut stdout_buf, stdout_truncated) = stdout_capture.take();
142    let (mut stderr_buf, stderr_truncated) = stderr_capture.take();
143    let decode_lossy =
144        std::str::from_utf8(&stdout_buf).is_err() || std::str::from_utf8(&stderr_buf).is_err();
145    let reader_incomplete = wait_failed
146        || kill_failed
147        || reaped.is_none()
148        || forced_pipe_cleanup
149        || read_failed.load(Ordering::Relaxed)
150        || reader_join_failed;
151
152    if timed_out || stdout_truncated {
153        let notice = format!(
154            "\n[lean-ctx: output truncated at {} MB / {}s limit]\n",
155            max_bytes / (1024 * 1024),
156            timeout.as_secs()
157        );
158        stdout_buf.extend_from_slice(notice.as_bytes());
159    }
160    if stderr_truncated {
161        let notice = format!(
162            "\n[lean-ctx: stderr truncated at {} KB limit]\n",
163            STDERR_LIMIT / 1024
164        );
165        stderr_buf.extend_from_slice(notice.as_bytes());
166    }
167    if !readers_finished {
168        stdout_buf.extend_from_slice(
169            b"\n[lean-ctx: output not captured: a process outside this call kept it open]\n",
170        );
171    }
172
173    if reaped.is_none() {
174        stdout_buf.extend_from_slice(
175            b"\n[lean-ctx: the process could not be ended within the cleanup budget]\n",
176        );
177    }
178    let status = reaped.unwrap_or_else(synthetic_failure_status);
179
180    LimitedProcessOutput {
181        output: Output {
182            status,
183            stdout: stdout_buf,
184            stderr: stderr_buf,
185        },
186        completeness: crate::core::context_admission::OutputCompleteness {
187            truncated: stdout_truncated || stderr_truncated || truncated.load(Ordering::Relaxed),
188            timed_out,
189            reader_incomplete,
190            cancelled: false,
191            decode_lossy,
192        },
193    }
194}
195
196/// What one reader has read of its stream so far, shared with the waiting
197/// thread: the bytes (never more than the stream's limit) and whether the
198/// limit cut them short.
199#[derive(Default)]
200struct StreamCapture {
201    bytes: std::sync::Mutex<Vec<u8>>,
202    truncated: AtomicBool,
203}
204
205impl StreamCapture {
206    /// The bytes read up to now and whether the limit was reached. A reader
207    /// that is still running keeps its lock only for one append, so this
208    /// never waits on a pipe.
209    fn take(&self) -> (Vec<u8>, bool) {
210        let bytes = std::mem::take(
211            &mut *self
212                .bytes
213                .lock()
214                .unwrap_or_else(std::sync::PoisonError::into_inner),
215        );
216        (bytes, self.truncated.load(Ordering::Relaxed))
217    }
218}
219
220/// Reads `pipe` to its end or to `limit` bytes into `capture`, chunk by
221/// chunk, so that the capture holds at every moment a prefix of the stream.
222/// `any_truncated` tells the waiting thread to end the child.
223fn read_stream(
224    pipe: Option<impl Read>,
225    limit: usize,
226    capture: &StreamCapture,
227    any_truncated: &AtomicBool,
228    read_failed: &AtomicBool,
229) {
230    let Some(mut pipe) = pipe else {
231        read_failed.store(true, Ordering::Relaxed);
232        return;
233    };
234    let mut chunk = [0u8; 8192];
235    loop {
236        match pipe.read(&mut chunk) {
237            Ok(0) => return,
238            Ok(n) => {
239                let mut bytes = capture
240                    .bytes
241                    .lock()
242                    .unwrap_or_else(std::sync::PoisonError::into_inner);
243                if bytes.len() + n > limit {
244                    let remaining = limit.saturating_sub(bytes.len());
245                    bytes.extend_from_slice(&chunk[..remaining]);
246                    capture.truncated.store(true, Ordering::Relaxed);
247                    any_truncated.store(true, Ordering::Relaxed);
248                    return;
249                }
250                bytes.extend_from_slice(&chunk[..n]);
251            }
252            Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
253            Err(_) => {
254                read_failed.store(true, Ordering::Relaxed);
255                return;
256            }
257        }
258    }
259}
260
261/// The process group this call may signal: the child's own, and only when the
262/// child leads it (it was spawned with `process_group(0)`). Read while the
263/// child is certainly not reaped, so the answer is about this child.
264///
265/// A child that has already exited cannot be asked on every system (macOS
266/// reports no group for a zombie). Its unreaped pid still names it alone, and
267/// a group with that number is either the one it led or does not exist: no
268/// other process can have created a group under a pid that is still taken.
269/// So the number is safe to signal, and a call whose leader finished before
270/// the wait began still ends what it left behind.
271#[cfg(unix)]
272fn owned_group(child: &mut Child, kill_group: bool) -> Option<i32> {
273    if !kill_group {
274        return None;
275    }
276    // Only a group id that may be signalled is ever returned (never 0, 1,
277    // a wrapped negative id, or the group this process is in).
278    let pid = crate::ipc::process::group_target(child.id()).ok()?;
279    // SAFETY: getpgid only reads kernel process metadata; the pid is ours and
280    // unreaped (alive or zombie), so it cannot name another process.
281    let group = unsafe { libc::getpgid(pid) };
282    if group == pid {
283        return Some(pid);
284    }
285    (group == -1 && matches!(exited_unreaped(child), Ok(true))).then_some(pid)
286}
287
288#[cfg(not(unix))]
289fn owned_group(_: &mut Child, _: bool) -> Option<i32> {
290    None
291}
292
293/// Whether the child has exited, WITHOUT reaping it on Unix (`WNOWAIT`).
294#[cfg(unix)]
295pub(crate) fn exited_unreaped(child: &mut Child) -> std::io::Result<bool> {
296    // SAFETY: an all-zero siginfo_t is a valid out-parameter for waitid.
297    let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() };
298    // SAFETY: waitid on our own unreaped child; WNOWAIT leaves it waitable.
299    let result = unsafe {
300        libc::waitid(
301            libc::P_PID,
302            child.id() as libc::id_t,
303            &raw mut info,
304            libc::WEXITED | libc::WNOHANG | libc::WNOWAIT,
305        )
306    };
307    if result == -1 {
308        let error = std::io::Error::last_os_error();
309        return if error.kind() == std::io::ErrorKind::Interrupted {
310            Ok(false)
311        } else {
312            Err(error)
313        };
314    }
315    // SAFETY: waitid succeeded; with WNOHANG si_pid stays 0 while running.
316    Ok(unsafe { info.si_pid() } != 0)
317}
318
319#[cfg(not(unix))]
320fn exited_unreaped(child: &mut Child) -> std::io::Result<bool> {
321    child.try_wait().map(|status| status.is_some())
322}
323
324fn readers_finish_within<A, B>(
325    first: &std::thread::JoinHandle<A>,
326    second: &std::thread::JoinHandle<B>,
327    budget: std::time::Duration,
328) -> bool {
329    let deadline = std::time::Instant::now() + budget;
330    loop {
331        if first.is_finished() && second.is_finished() {
332            return true;
333        }
334        if std::time::Instant::now() >= deadline {
335            return false;
336        }
337        std::thread::sleep(std::time::Duration::from_millis(5));
338    }
339}
340
341/// Kill the child — and, when it leads its own process group, every
342/// descendant in that group (GH #720). SIGKILL to the group ends a shell's
343/// grandchildren so the captured pipes actually close. A group the child
344/// merely inherited is never signalled: that would end the caller and its
345/// unrelated siblings.
346///
347/// Returns `false` when the platform's tree kill did not confirm (Windows:
348/// `taskkill` failed, or did not finish within its own budget). The direct
349/// child is killed in every case.
350fn kill_child(child: &mut Child, own_group: Option<i32>) -> bool {
351    #[cfg(unix)]
352    if let Some(group) = own_group {
353        // SAFETY: plain syscall on a group id pinned by our unreaped child.
354        unsafe { libc::killpg(group, libc::SIGKILL) };
355    }
356    #[cfg(not(unix))]
357    let _ = own_group;
358    #[cfg(windows)]
359    let tree_killed = {
360        const TREE_KILL_BUDGET: std::time::Duration = std::time::Duration::from_secs(5);
361        match std::process::Command::new("taskkill")
362            .args(["/F", "/T", "/PID", &child.id().to_string()])
363            .stdin(std::process::Stdio::null())
364            .stdout(std::process::Stdio::null())
365            .stderr(std::process::Stdio::null())
366            .spawn()
367        {
368            Ok(mut tree_kill) => {
369                if let Some(status) = reap_within(&mut tree_kill, TREE_KILL_BUDGET) {
370                    status.success()
371                } else {
372                    let _ = tree_kill.kill();
373                    let _ = reap_within(&mut tree_kill, TREE_KILL_BUDGET);
374                    false
375                }
376            }
377            Err(_) => false,
378        }
379    };
380    #[cfg(not(windows))]
381    let tree_killed = true;
382    let _ = child.kill();
383    tree_killed
384}
385
386/// Reap a child that has exited or was just killed, within `budget`.
387pub fn reap_within(
388    child: &mut Child,
389    budget: std::time::Duration,
390) -> Option<std::process::ExitStatus> {
391    let deadline = std::time::Instant::now() + budget;
392    loop {
393        match child.try_wait() {
394            Ok(Some(status)) => return Some(status),
395            Ok(None) if std::time::Instant::now() < deadline => {
396                std::thread::sleep(std::time::Duration::from_millis(5));
397            }
398            Ok(None) | Err(_) => return None,
399        }
400    }
401}
402
403/// A synthetic failed `ExitStatus`, used only when `Child::wait()` itself
404/// errors (e.g. the process was already reaped by another waiter) and there
405/// is no real status to report. The previous fallback shelled out to
406/// `Command::new("false").status()` to manufacture one, which panicked via
407/// `.expect()` wherever no `false` binary exists on `PATH` — Windows, and
408/// minimal/scratch containers. `ExitStatusExt::from_raw` builds the status
409/// value directly, with no subprocess involved, so it can't fail.
410#[cfg(unix)]
411fn synthetic_failure_status() -> std::process::ExitStatus {
412    use std::os::unix::process::ExitStatusExt;
413    // Raw wait(2) status encoding: low 7 bits 0 signals a normal exit
414    // (`WIFEXITED`), the next byte up is the exit code (`WEXITSTATUS`) — so
415    // `1 << 8` decodes as "exited normally with code 1".
416    std::process::ExitStatus::from_raw(1 << 8)
417}
418
419#[cfg(not(unix))]
420fn synthetic_failure_status() -> std::process::ExitStatus {
421    use std::os::windows::process::ExitStatusExt;
422    std::process::ExitStatus::from_raw(1)
423}
424
425#[cfg(test)]
426mod tests {
427    const OUTPUT_HELPER_ENV: &str = "LEAN_CTX_TEST_OUTPUT_HELPER";
428    const OUTPUT_HELPER_TEST: &str = "shell::exec::process::tests::wait_with_limits_output_helper";
429
430    #[test]
431    #[ignore = "subprocess helper for wait_with_limits tests"]
432    fn wait_with_limits_output_helper() {
433        use std::io::Write as _;
434
435        match std::env::var(OUTPUT_HELPER_ENV).as_deref() {
436            Ok("stdout-small") => {
437                std::io::stdout().write_all(b"hello\n").unwrap();
438            }
439            // A producer that never ends on its own. Once the reader drops the
440            // pipe, keep the process alive until the parent kills it: exiting
441            // here would race that kill against the instrumented binary's
442            // at-exit coverage profile write and leave a truncated .profraw
443            // that breaks the Coverage job's report.
444            Ok("stdout-forever") => {
445                let chunk = [b'a'; 8192];
446                let mut stream = std::io::stdout().lock();
447                while stream.write_all(&chunk).is_ok() {}
448                std::thread::sleep(std::time::Duration::from_mins(1));
449            }
450            Ok("stderr-forever") => {
451                let chunk = [b'a'; 8192];
452                let mut stream = std::io::stderr().lock();
453                while stream.write_all(&chunk).is_ok() {}
454                std::thread::sleep(std::time::Duration::from_mins(1));
455            }
456            Ok("sleep") => {
457                std::thread::sleep(std::time::Duration::from_mins(1));
458            }
459            #[cfg(unix)]
460            Ok("guardian") => guardian(),
461            Ok(other) => panic!("unknown output helper mode: {other}"),
462            Err(error) => panic!("missing {OUTPUT_HELPER_ENV}: {error}"),
463        }
464    }
465
466    fn spawn_output_helper(mode: &str) -> std::process::Child {
467        std::process::Command::new(std::env::current_exe().unwrap())
468            .args([OUTPUT_HELPER_TEST, "--exact", "--ignored", "--nocapture"])
469            .env(OUTPUT_HELPER_ENV, mode)
470            .stdout(std::process::Stdio::piped())
471            .stderr(std::process::Stdio::piped())
472            .spawn()
473            .unwrap()
474    }
475
476    #[test]
477    fn wait_with_limits_captures_output() {
478        let child = spawn_output_helper("stdout-small");
479
480        let output = super::wait_with_limits(child, 1024, std::time::Duration::from_secs(5), false);
481        let stdout = String::from_utf8_lossy(&output.stdout);
482        assert!(
483            stdout.contains("hello"),
484            "expected 'hello' in output: {stdout}"
485        );
486        assert!(output.status.success());
487    }
488
489    #[test]
490    fn wait_with_limits_truncates_large_output() {
491        let child = spawn_output_helper("stdout-forever");
492
493        let output =
494            super::wait_with_limits(child, 1024, std::time::Duration::from_secs(10), false);
495        let stdout = String::from_utf8_lossy(&output.stdout);
496        assert!(
497            stdout.contains("[lean-ctx: output truncated"),
498            "expected truncation notice, got len={}: ...{}",
499            stdout.len(),
500            &stdout[stdout.len().saturating_sub(80)..]
501        );
502    }
503
504    #[test]
505    fn synthetic_failure_status_is_a_failure_without_spawning_anything() {
506        let status = super::synthetic_failure_status();
507        assert!(!status.success());
508        #[cfg(unix)]
509        {
510            use std::os::unix::process::ExitStatusExt;
511            assert_eq!(status.code(), Some(1));
512            assert_eq!(status.signal(), None);
513        }
514    }
515
516    #[test]
517    fn wait_with_limits_truncates_large_stderr() {
518        let child = spawn_output_helper("stderr-forever");
519
520        let output = super::wait_with_limits(
521            child,
522            1024 * 1024,
523            std::time::Duration::from_secs(10),
524            false,
525        );
526        let stderr = String::from_utf8_lossy(&output.stderr);
527        assert!(
528            stderr.contains("[lean-ctx: stderr truncated"),
529            "expected stderr truncation notice, got len={}: ...{}",
530            stderr.len(),
531            &stderr[stderr.len().saturating_sub(80)..]
532        );
533    }
534
535    #[test]
536    fn wait_with_limits_kills_promptly_on_truncation() {
537        let child = spawn_output_helper("stdout-forever");
538
539        let start = std::time::Instant::now();
540        let output =
541            super::wait_with_limits(child, 4096, std::time::Duration::from_secs(20), false);
542        let elapsed = start.elapsed();
543
544        assert!(
545            elapsed < std::time::Duration::from_secs(3),
546            "truncation should kill promptly, took {elapsed:?} (timeout was 20s)"
547        );
548        let stdout = String::from_utf8_lossy(&output.stdout);
549        assert!(stdout.contains("[lean-ctx: output truncated"));
550    }
551
552    #[test]
553    fn wait_with_limits_timeout_kills_process() {
554        let child = spawn_output_helper("sleep");
555
556        let start = std::time::Instant::now();
557        let output =
558            super::wait_with_limits(child, 1024, std::time::Duration::from_millis(200), false);
559        let elapsed = start.elapsed();
560
561        assert!(
562            elapsed < std::time::Duration::from_secs(3),
563            "timeout should kill quickly, took {elapsed:?}"
564        );
565        let stdout = String::from_utf8_lossy(&output.stdout);
566        assert!(stdout.contains("[lean-ctx: output truncated"));
567    }
568
569    /// GH #720: killing only the direct child (a shell) on timeout leaves its
570    /// grandchildren alive holding the stdout pipe — the reader threads never
571    /// see EOF and `wait_with_limits` blocks forever even though the timeout
572    /// fired. With the child in its own process group and a group kill, the
573    /// whole tree dies and the call returns promptly.
574    #[cfg(unix)]
575    #[test]
576    fn wait_with_limits_group_kill_reaps_grandchildren() {
577        use std::os::unix::process::CommandExt as _;
578        // The shell spawns a grandchild that inherits stdout and sleeps far
579        // beyond the timeout; the shell itself also sleeps so the timeout path
580        // (not natural exit) is exercised.
581        let mut cmd = std::process::Command::new("sh");
582        cmd.args(["-c", "sleep 30 & sleep 30"])
583            .stdin(std::process::Stdio::null())
584            .stdout(std::process::Stdio::piped())
585            .stderr(std::process::Stdio::piped());
586        cmd.process_group(0);
587        let child = cmd.spawn().unwrap();
588        let pgid = child.id() as libc::pid_t;
589
590        let start = std::time::Instant::now();
591        let _ = super::wait_with_limits(child, 1024, std::time::Duration::from_millis(200), true);
592        let elapsed = start.elapsed();
593
594        assert!(
595            elapsed < std::time::Duration::from_secs(5),
596            "group kill must unblock the reader threads, took {elapsed:?}"
597        );
598        // The whole group must be gone (ESRCH), not just the direct child.
599        // A brief grace period lets the kernel finish reaping.
600        let mut group_gone = false;
601        for _ in 0..50 {
602            // SAFETY: signal 0 only probes for existence.
603            if unsafe { libc::killpg(pgid, 0) } == -1 {
604                group_gone = true;
605                break;
606            }
607            std::thread::sleep(std::time::Duration::from_millis(20));
608        }
609        assert!(group_gone, "process group {pgid} must be fully reaped");
610    }
611
612    #[cfg(unix)]
613    fn alive(pid: u32) -> bool {
614        crate::ipc::process::is_alive(pid)
615    }
616
617    /// Body of the guardian process (its own process group, created by the
618    /// test below). Everything that could be signalled by mistake lives in
619    /// this group: the guardian itself and a sentinel that belongs to no call.
620    #[cfg(unix)]
621    fn guardian() {
622        use std::os::unix::process::CommandExt as _;
623        use std::process::{Command, Stdio};
624        let piped = |script: &str| {
625            let mut command = Command::new("sh");
626            command
627                .args(["-c", script])
628                .stdin(Stdio::null())
629                .stdout(Stdio::piped())
630                .stderr(Stdio::piped());
631            command
632        };
633        let mut sentinel = Command::new("sleep")
634            .arg("60")
635            .stdin(Stdio::null())
636            .stdout(Stdio::null())
637            .stderr(Stdio::null())
638            .spawn()
639            .unwrap();
640        // SAFETY: getpgrp has no preconditions.
641        let own = unsafe { libc::getpgrp() };
642        assert_eq!(own, std::process::id() as libc::pid_t, "guardian leads");
643        // Whatever happens below, the fixture ends its own group — itself
644        // included — while it still leads it, so nothing outlives the test
645        // and no group id is signalled after it could have been reused.
646        struct EndOwnGroup(libc::pid_t);
647        impl Drop for EndOwnGroup {
648            fn drop(&mut self) {
649                use std::io::Write as _;
650                let _ = std::io::stdout().flush();
651                // SAFETY: this process leads the group and is alive.
652                unsafe { libc::killpg(self.0, libc::SIGKILL) };
653            }
654        }
655        let _end = EndOwnGroup(own);
656        // SAFETY: reads metadata of our live child.
657        assert_eq!(unsafe { libc::getpgid(sentinel.id() as libc::pid_t) }, own);
658
659        // 1. A call that INHERITED this group runs into its timeout while a
660        // grandchild holds its pipes. Only the call's direct child is killed;
661        // sentinel and guardian live; the capture is bounded and incomplete.
662        let child = piped("sleep 60 & sleep 60").spawn().unwrap();
663        let start = std::time::Instant::now();
664        let captured = super::wait_with_limits_captured(
665            child,
666            1024,
667            std::time::Duration::from_millis(200),
668            true,
669        );
670        assert!(start.elapsed() < std::time::Duration::from_secs(8));
671        assert!(captured.completeness.timed_out);
672        assert!(captured.completeness.reader_incomplete);
673        assert!(
674            alive(sentinel.id()),
675            "inherited group must not be signalled"
676        );
677
678        // 2. The same with the output limit instead of the timeout.
679        let child = piped("sleep 60 & yes").spawn().unwrap();
680        let captured =
681            super::wait_with_limits_captured(child, 1024, std::time::Duration::from_secs(30), true);
682        assert!(captured.completeness.truncated);
683        assert!(
684            alive(sentinel.id()),
685            "inherited group must not be signalled"
686        );
687
688        // 3. A call with its OWN group whose leader exits at once while its
689        // child keeps the pipes: the group is found although the leader is
690        // gone, the whole call tree ends, nothing else does.
691        let mut command = piped("sleep 60 & echo started");
692        command.process_group(0);
693        let child = command.spawn().unwrap();
694        let group = child.id() as libc::pid_t;
695        let start = std::time::Instant::now();
696        let captured =
697            super::wait_with_limits_captured(child, 1024, std::time::Duration::from_secs(30), true);
698        assert!(
699            start.elapsed() < std::time::Duration::from_secs(5),
700            "leader gone, pipe holder alive: must not wait for it"
701        );
702        assert!(!captured.completeness.timed_out);
703        assert!(captured.completeness.reader_incomplete);
704        assert!(String::from_utf8_lossy(&captured.output.stdout).contains("started"));
705        let mut group_gone = false;
706        for _ in 0..100 {
707            // SAFETY: signal 0 only probes for existence.
708            if unsafe { libc::killpg(group, 0) } == -1 {
709                group_gone = true;
710                break;
711            }
712            std::thread::sleep(std::time::Duration::from_millis(20));
713        }
714        assert!(group_gone, "the call's own group must be gone");
715        assert!(
716            alive(sentinel.id()),
717            "a foreign group must not be signalled"
718        );
719
720        // 4. A command writes a line and exits, and leaves a process outside
721        // any group this call owns holding its pipes (the call inherited
722        // this group, so nothing is signalled). The readers are abandoned
723        // after their budget; what they had read is returned, in front of
724        // the notice, and the capture is flagged incomplete.
725        let child = piped("echo written; echo problem >&2; sleep 60 &")
726            .spawn()
727            .unwrap();
728        let start = std::time::Instant::now();
729        let captured =
730            super::wait_with_limits_captured(child, 1024, std::time::Duration::from_secs(30), true);
731        assert!(start.elapsed() < std::time::Duration::from_secs(8));
732        assert!(captured.output.status.success());
733        assert!(captured.completeness.reader_incomplete);
734        assert!(!captured.completeness.timed_out && !captured.completeness.truncated);
735        assert_eq!(
736            String::from_utf8_lossy(&captured.output.stdout),
737            "written\n\n[lean-ctx: output not captured: a process outside this call kept it open]\n"
738        );
739        assert_eq!(captured.output.stderr, b"problem\n");
740        assert!(
741            alive(sentinel.id()),
742            "inherited group must not be signalled"
743        );
744
745        // 5. Control: a command that leaves nothing is returned as it always
746        // was — its bytes and nothing else, complete.
747        let child = piped("echo written; echo problem >&2").spawn().unwrap();
748        let captured =
749            super::wait_with_limits_captured(child, 1024, std::time::Duration::from_secs(30), true);
750        assert_eq!(captured.output.stdout, b"written\n");
751        assert_eq!(captured.output.stderr, b"problem\n");
752        assert!(captured.completeness.is_complete());
753
754        // The sentinel survived every case; end and reap it ourselves.
755        sentinel.kill().unwrap();
756        sentinel.wait().unwrap();
757        println!("GUARDIAN-COMPLETE");
758    }
759
760    /// Q13: a timeout or output limit never signals a process group the call
761    /// did not create, and a leader that exited before its pipe-holding child
762    /// does not hide the call's own group. Runs under a guardian in its own
763    /// group, so a regression can only kill the fixture, never the test run.
764    #[cfg(unix)]
765    #[test]
766    fn limits_signal_only_a_group_the_call_owns() {
767        use std::os::unix::process::CommandExt as _;
768        let mut command = std::process::Command::new(std::env::current_exe().unwrap());
769        command
770            .args([OUTPUT_HELPER_TEST, "--exact", "--ignored", "--nocapture"])
771            .env(OUTPUT_HELPER_ENV, "guardian")
772            .stdin(std::process::Stdio::null())
773            .stdout(std::process::Stdio::piped())
774            .stderr(std::process::Stdio::piped());
775        command.process_group(0);
776        let child = command.spawn().unwrap();
777        // The guardian ends its own group as its last act, so its exit status
778        // is a kill signal by design; the marker line is the verdict.
779        let captured = super::wait_with_limits_captured(
780            child,
781            64 * 1024,
782            std::time::Duration::from_mins(1),
783            true,
784        );
785        let stdout = String::from_utf8_lossy(&captured.output.stdout);
786        let stderr = String::from_utf8_lossy(&captured.output.stderr);
787        assert!(
788            stdout.contains("GUARDIAN-COMPLETE"),
789            "guardian status {:?}\n{stdout}\n{stderr}",
790            captured.output.status
791        );
792    }
793
794    /// #806: piped stdin must be forwarded to the child via Stdio::piped()
795    /// and relayed, not nulled. Tests the relay pattern: write data to child
796    /// stdin, close it (EOF), child reads and exits.
797    #[cfg(unix)]
798    #[test]
799    fn stdin_relay_forwards_piped_data() {
800        use std::io::Write;
801        use std::os::unix::process::CommandExt as _;
802
803        let mut cmd = std::process::Command::new("cat");
804        cmd.stdin(std::process::Stdio::piped())
805            .stdout(std::process::Stdio::piped())
806            .stderr(std::process::Stdio::piped());
807        cmd.process_group(0);
808        let mut child = cmd.spawn().expect("failed to spawn cat");
809
810        let mut child_stdin = child.stdin.take().unwrap();
811        std::thread::spawn(move || {
812            child_stdin.write_all(b"hello from pipe\n").unwrap();
813            drop(child_stdin);
814        });
815
816        let output = child.wait_with_output().expect("wait failed");
817        let stdout = String::from_utf8_lossy(&output.stdout);
818        assert!(
819            stdout.contains("hello from pipe"),
820            "#806: piped stdin must reach the child, got: {stdout}"
821        );
822    }
823
824    /// #806: commands that don't read stdin must still work normally
825    /// when no data is piped (relay thread sees immediate EOF from parent).
826    #[cfg(unix)]
827    #[test]
828    fn stdin_relay_no_data_does_not_hang() {
829        let start = std::time::Instant::now();
830        let mut cmd = std::process::Command::new("sh");
831        cmd.args(["-c", "echo ok"])
832            .stdin(std::process::Stdio::piped())
833            .stdout(std::process::Stdio::piped())
834            .stderr(std::process::Stdio::piped());
835        use std::os::unix::process::CommandExt as _;
836        cmd.process_group(0);
837        let mut child = cmd.spawn().unwrap();
838        // Close stdin immediately (simulates relay with empty parent pipe)
839        drop(child.stdin.take());
840        let output = child.wait_with_output().unwrap();
841        let elapsed = start.elapsed();
842        assert!(
843            elapsed < std::time::Duration::from_secs(5),
844            "must not hang when stdin is closed immediately, took {elapsed:?}"
845        );
846        let stdout = String::from_utf8_lossy(&output.stdout);
847        assert!(stdout.contains("ok"), "command output missing: {stdout}");
848    }
849}