mentra 0.27.0

An agent runtime for tool-using LLM applications
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
use std::{
    ffi::{OsStr, OsString},
    fmt, io,
    path::{Path, PathBuf},
    process::Stdio,
    time::Duration,
};

#[cfg(windows)]
use std::process::Command as StdCommand;

use tokio::{
    io::AsyncWriteExt,
    process::{Child, ChildStdin, ChildStdout, Command},
    task::JoinHandle,
    time::Instant,
};

use super::capture::{CapturedStream, read_capped};

/// How long a program that exited on time still gets for its pipes to drain.
///
/// One that answers with a millisecond to spare should not lose its output to
/// the scheduling latency between its exit and the last bytes arriving.
const DRAIN_GRACE: Duration = Duration::from_millis(250);

/// How long cleanup after a kill may take: reaping the child, then reading
/// what its now-closed pipes still hold.
const CLEANUP_GRACE: Duration = Duration::from_secs(2);

/// How a program ended.
///
/// Both variants carry whatever the program printed, bounded the same way: a
/// half-written answer is not an answer, but it is often the only evidence of
/// what went wrong, and discarding it is the caller's decision to make.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Completion {
    /// The program ran to completion inside its budget.
    Exited {
        /// The exit code, or `None` when a signal ended it — a failure like
        /// any other, and not something every platform gives a number for.
        code: Option<i32>,
        stdout: CapturedStream,
        stderr: CapturedStream,
    },
    /// The budget ran out first, and the process tree was killed.
    ///
    /// Reached either because the program was still running at the deadline or
    /// because its output was still arriving past it — a descendant holding
    /// the pipes open is the same missed deadline as a program that never
    /// exited.
    TimedOut {
        stdout: CapturedStream,
        stderr: CapturedStream,
    },
}

impl Completion {
    /// Whether the budget, rather than the program, decided when this ended.
    pub fn timed_out(&self) -> bool {
        matches!(self, Self::TimedOut { .. })
    }

    /// The exit code, or `None` for a signal or a timeout.
    pub fn code(&self) -> Option<i32> {
        match self {
            Self::Exited { code, .. } => *code,
            Self::TimedOut { .. } => None,
        }
    }

    /// What the program printed on stdout before it ended.
    pub fn stdout(&self) -> &CapturedStream {
        match self {
            Self::Exited { stdout, .. } | Self::TimedOut { stdout, .. } => stdout,
        }
    }

    /// What the program printed on stderr before it ended.
    pub fn stderr(&self) -> &CapturedStream {
        match self {
            Self::Exited { stderr, .. } | Self::TimedOut { stderr, .. } => stderr,
        }
    }
}

/// Runs another program under the process discipline mentra applies to its own
/// commands.
///
/// The guarantees a caller gets, all of them enforced by
/// [`run`](Self::run) rather than by convention:
///
/// - **The environment is exactly what was passed.** The child's environment is
///   cleared before the pairs given to [`env`](Self::env) / [`envs`](Self::envs)
///   are set, so nothing this process happens to be holding — a token, a proxy
///   setting, a `PATH` — reaches the program unless the caller listed it.
/// - **The process tree is grouped and killed as a unit.** On unix the child
///   is put in its own session with `setsid`, and the deadline kills the whole
///   group; a program that backgrounds work cannot leave it running. On Windows
///   the tree is killed with `taskkill /T /F`.
/// - **One budget covers spawning, running, and reading.** A descendant that
///   inherits the pipes cannot hold a caller past the deadline: output still
///   arriving then is a [`Completion::TimedOut`], not a wait.
/// - **Output is capped while it is read.** Neither stream can cost more than
///   the configured stdout/stderr cap, however much the program prints, and both
///   are still drained so the program is never blocked on a full pipe.
/// - **A dropped run kills the child.** The command is spawned with
///   `kill_on_drop`, so a cancelled future does not leak a process.
///
/// Both bounds are constructor arguments rather than options with defaults:
/// there is no way to spell an unbounded run of this type, which is the whole
/// of what "bounded" means here.
///
/// ```no_run
/// # async fn example() -> std::io::Result<()> {
/// use std::time::Duration;
/// use mentra::process::{BoundedCommand, Completion};
///
/// let completion = BoundedCommand::new("./hooks/guard.sh", Duration::from_secs(5), 64 * 1024)
///     .current_dir("/repo")
///     .env("BASIS_EVENT", "pre_tool_use")
///     .stdin(r#"{"tool":"shell"}"#)
///     .run()
///     .await?;
///
/// match completion {
///     Completion::Exited { code, stdout, .. } if code == Some(0) => {
///         println!("{}", stdout.to_string_lossy());
///     }
///     _ => println!("the hook did not answer"),
/// }
/// # Ok(())
/// # }
/// ```
pub struct BoundedCommand {
    program: OsString,
    args: Vec<OsString>,
    current_dir: Option<PathBuf>,
    env: Vec<(OsString, OsString)>,
    stdin: Option<Vec<u8>>,
    timeout: Duration,
    max_stdout_bytes: usize,
    max_stderr_bytes: usize,
}

/// The small environment baseline used by interactive host-owned processes.
///
/// `BoundedCommand` itself always starts from an empty environment and callers
/// choose which names to add. MCP stdio uses this shared baseline so its bare
/// command lookup and ordinary language runtimes keep working without
/// receiving ambient credentials or shell state.
#[cfg(not(windows))]
const BASELINE_ENVIRONMENT: &[&str] = &["PATH", "HOME", "TMPDIR", "TMP", "TEMP", "LANG", "LC_ALL"];

#[cfg(windows)]
const BASELINE_ENVIRONMENT: &[&str] = &["PATH", "PATHEXT", "SystemRoot", "COMSPEC", "TEMP", "TMP"];

pub(crate) fn baseline_environment() -> Vec<(OsString, OsString)> {
    BASELINE_ENVIRONMENT
        .iter()
        .filter_map(|name| std::env::var_os(name).map(|value| (OsString::from(name), value)))
        .collect()
}

/// A child started with the process discipline, kept alive for a bidirectional
/// protocol such as MCP rather than consumed by [`BoundedCommand::run`].
pub(crate) struct BoundedChild {
    child: Child,
    group: Option<u32>,
    stderr_task: Option<JoinHandle<io::Result<CapturedStream>>>,
    armed: bool,
}

impl BoundedChild {
    pub(crate) fn take_stdin(&mut self) -> Option<ChildStdin> {
        self.child.stdin.take()
    }

    pub(crate) fn take_stdout(&mut self) -> Option<ChildStdout> {
        self.child.stdout.take()
    }

    /// Terminates the process group and waits briefly for the direct child.
    pub(crate) async fn terminate(&mut self) -> io::Result<()> {
        if self.armed {
            kill_tree_and_reap(&mut self.child, self.group).await?;
            self.armed = false;
        }
        self.abort_stderr_reader();
        Ok(())
    }

    fn abort_stderr_reader(&mut self) {
        if let Some(task) = self.stderr_task.take() {
            task.abort();
        }
    }

    /// Whether the stderr reader is still installed for this child.
    ///
    /// Only exposed to crate tests: the observable contract is that a server
    /// cannot block on a full stderr pipe, not the task representation.
    #[cfg(test)]
    pub(crate) fn drains_stderr(&self) -> bool {
        self.stderr_task.is_some()
    }
}

impl Drop for BoundedChild {
    fn drop(&mut self) {
        if self.armed {
            // Drop cannot await, but the group signal is synchronous. The
            // Child's `kill_on_drop` flag covers the direct process as well;
            // this extra group signal is what prevents descendants surviving a
            // disconnected protocol client.
            let _ = kill_entire_process_tree(&mut self.child, self.group);
            self.armed = false;
        }
        self.abort_stderr_reader();
    }
}

impl fmt::Debug for BoundedCommand {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let env_names = self
            .env
            .iter()
            .map(|(key, _)| key.to_string_lossy().into_owned())
            .collect::<Vec<_>>();
        f.debug_struct("BoundedCommand")
            .field("program", &self.program)
            .field("arg_count", &self.args.len())
            .field("current_dir", &self.current_dir)
            .field("env_names", &env_names)
            .field("stdin_bytes", &self.stdin.as_ref().map(Vec::len))
            .field("timeout", &self.timeout)
            .field("stdout_max_bytes", &self.max_stdout_bytes)
            .field("stderr_max_bytes", &self.max_stderr_bytes)
            .finish()
    }
}

#[cfg(test)]
mod api_tests {
    use super::*;

    #[test]
    fn debug_redacts_values_and_payload_but_identifies_the_command() {
        let command = BoundedCommand::new("/bin/echo", Duration::from_secs(3), 128)
            .arg("--api-token")
            .arg("argv-secret")
            .env("API_TOKEN", "environment-secret")
            .stdin("stdin-secret")
            .max_stdout_bytes(256)
            .max_stderr_bytes(64);

        let rendered = format!("{command:?}");
        assert!(rendered.contains("/bin/echo"), "{rendered}");
        assert!(rendered.contains("API_TOKEN"), "{rendered}");
        assert!(rendered.contains("stdout_max_bytes: 256"), "{rendered}");
        assert!(rendered.contains("stderr_max_bytes: 64"), "{rendered}");
        assert!(!rendered.contains("environment-secret"), "{rendered}");
        assert!(!rendered.contains("argv-secret"), "{rendered}");
        assert!(!rendered.contains("stdin-secret"), "{rendered}");
    }
}

impl BoundedCommand {
    /// Runs `program` directly, with no shell between the caller and it.
    ///
    /// The argv vector is passed through as given, so an argument containing
    /// spaces, quotes or a `$` needs no escaping and means exactly itself.
    pub fn new(
        program: impl Into<OsString>,
        timeout: Duration,
        max_output_bytes_per_stream: usize,
    ) -> Self {
        Self {
            program: program.into(),
            args: Vec::new(),
            current_dir: None,
            env: Vec::new(),
            stdin: None,
            timeout,
            max_stdout_bytes: max_output_bytes_per_stream,
            max_stderr_bytes: max_output_bytes_per_stream,
        }
    }

    /// Runs `command` through the platform shell — `/bin/sh -c` on unix,
    /// `cmd.exe /C` on Windows.
    ///
    /// This is what a shell tool wants and what an argv caller does not: the
    /// command is one string the shell parses, so everything in it is the
    /// caller's to quote.
    pub fn shell(
        command: impl Into<OsString>,
        timeout: Duration,
        max_output_bytes_per_stream: usize,
    ) -> Self {
        let command = command.into();
        #[cfg(unix)]
        let (program, flag) = ("/bin/sh", "-c");
        #[cfg(windows)]
        let (program, flag) = ("cmd.exe", "/C");

        Self::new(program, timeout, max_output_bytes_per_stream)
            .arg(flag)
            .arg(command)
    }

    /// Appends one argument.
    pub fn arg(mut self, arg: impl Into<OsString>) -> Self {
        self.args.push(arg.into());
        self
    }

    /// Appends several arguments.
    pub fn args<I, A>(mut self, args: I) -> Self
    where
        I: IntoIterator<Item = A>,
        A: Into<OsString>,
    {
        self.args.extend(args.into_iter().map(Into::into));
        self
    }

    /// Runs the program in `dir`.
    ///
    /// A relative program path with a directory part — `./hooks/guard.sh` — is
    /// resolved against `dir`, so it means the same thing wherever the host
    /// process was started from. A bare name is left for `PATH` to answer.
    pub fn current_dir(mut self, dir: impl Into<PathBuf>) -> Self {
        self.current_dir = Some(dir.into());
        self
    }

    /// Adds one variable to the child's otherwise empty environment.
    pub fn env(mut self, key: impl Into<OsString>, value: impl Into<OsString>) -> Self {
        self.env.push((key.into(), value.into()));
        self
    }

    /// Adds several variables to the child's otherwise empty environment.
    pub fn envs<I, K, V>(mut self, vars: I) -> Self
    where
        I: IntoIterator<Item = (K, V)>,
        K: Into<OsString>,
        V: Into<OsString>,
    {
        self.env.extend(
            vars.into_iter()
                .map(|(key, value)| (key.into(), value.into())),
        );
        self
    }

    /// Writes `payload` to the program's stdin, then closes it.
    ///
    /// Without this the child gets a null stdin, which is what a program that
    /// reads nothing should see. The payload is written from its own task, so a
    /// program that answers without ever reading — `echo '{"allow":true}'` is a
    /// legitimate hook — cannot deadlock a caller whose payload outgrew the
    /// pipe buffer. The write ending in `EPIPE` is not an error: what the
    /// program printed and how it exited are the answer.
    pub fn stdin(mut self, payload: impl Into<Vec<u8>>) -> Self {
        self.stdin = Some(payload.into());
        self
    }

    /// Sets the maximum bytes retained from stdout.
    ///
    /// The constructor's `max_output_bytes_per_stream` remains the initial cap
    /// for both streams; use this additive builder when stdout needs its own
    /// budget.
    pub fn max_stdout_bytes(mut self, max_bytes: usize) -> Self {
        self.max_stdout_bytes = max_bytes;
        self
    }

    /// Sets the maximum bytes retained from stderr.
    ///
    /// The constructor's `max_output_bytes_per_stream` remains the initial cap
    /// for both streams; use this additive builder when stderr needs its own
    /// budget.
    pub fn max_stderr_bytes(mut self, max_bytes: usize) -> Self {
        self.max_stderr_bytes = max_bytes;
        self
    }

    /// Starts an interactive process with piped stdin/stdout and a bounded
    /// stderr drain. This is crate-internal because the public primitive's
    /// contract is the one-shot [`run`](Self::run) operation; protocol clients
    /// own their request deadlines separately.
    pub(crate) fn spawn_piped(&self) -> io::Result<BoundedChild> {
        let mut child = self.spawn_with_stdio(Stdio::piped(), Stdio::piped(), Stdio::piped())?;
        let group = child.id();
        let stderr = child.stderr.take().ok_or_else(|| missing_pipe("stderr"))?;
        let stderr_task = tokio::spawn(read_capped(stderr, self.max_stderr_bytes));
        Ok(BoundedChild {
            child,
            group,
            stderr_task: Some(stderr_task),
            armed: true,
        })
    }

    /// Spawns the program and waits for it, inside the budget.
    ///
    /// `Err` means the program could not be started or supervised at all. A
    /// program that ran and misbehaved — a non-zero exit, a flood of output, a
    /// missed deadline — is a [`Completion`], because what that means is the
    /// caller's to decide.
    pub async fn run(self) -> io::Result<Completion> {
        // Started before the spawn, because forking a process is part of what
        // the program is being given time for.
        let deadline = Instant::now() + self.timeout;
        let mut child = self.spawn()?;
        // Recorded now because a reaped child has no id left to ask for, and
        // the group kill below can outlive the program that led the group.
        let group = child.id();

        let stdin_task = match self.stdin {
            Some(payload) => {
                let mut pipe = child.stdin.take().ok_or_else(|| missing_pipe("stdin"))?;
                Some(tokio::spawn(async move {
                    // A broken pipe here is not an error: a program that
                    // answers without reading is legitimate, and this only
                    // offers the payload. Dropping the handle at the end of the
                    // task is what tells a program that does read where the
                    // payload stopped.
                    let _ = pipe.write_all(&payload).await;
                    let _ = pipe.shutdown().await;
                }))
            }
            None => None,
        };
        let stdout = child.stdout.take().ok_or_else(|| missing_pipe("stdout"))?;
        let stderr = child.stderr.take().ok_or_else(|| missing_pipe("stderr"))?;
        let stdout_cap = self.max_stdout_bytes;
        let stderr_cap = self.max_stderr_bytes;
        let mut streams = Drain {
            stdout: DrainingStream::new(tokio::spawn(read_capped(stdout, stdout_cap))),
            stderr: DrainingStream::new(tokio::spawn(read_capped(stderr, stderr_cap))),
        };

        let waited = tokio::time::timeout_at(deadline, child.wait()).await;
        let mut timed_out = waited.is_err();
        let code = match waited {
            Ok(status) => status?.code(),
            Err(_) => {
                kill_tree_and_reap(&mut child, group).await?;
                None
            }
        };

        // A program that exited on time still gets the rest of its budget, and
        // at least the grace, for output already in flight. Output still
        // arriving after that is a descendant holding the pipes: the same
        // missed deadline as a program that never exited, so the tree goes and
        // the completion says so.
        let drain_deadline = if timed_out {
            Instant::now() + CLEANUP_GRACE
        } else {
            deadline.max(Instant::now() + DRAIN_GRACE)
        };
        if !streams.drained_by(drain_deadline).await? {
            timed_out = true;
            kill_tree_and_reap(&mut child, group).await?;
            if !streams.drained_by(Instant::now() + CLEANUP_GRACE).await? {
                return Err(io::Error::new(
                    io::ErrorKind::TimedOut,
                    "timed out while draining command output",
                ));
            }
        }

        // The payload is no longer wanted by anyone: a program that read it has
        // exited, and one that did not was killed.
        if let Some(task) = stdin_task {
            task.abort();
        }

        let (stdout, stderr) = streams.into_captures();
        Ok(if timed_out {
            Completion::TimedOut { stdout, stderr }
        } else {
            Completion::Exited {
                code,
                stdout,
                stderr,
            }
        })
    }

    /// Starts the child with every process and environment rule this type
    /// promises applied.
    fn spawn(&self) -> io::Result<Child> {
        self.spawn_with_stdio(
            if self.stdin.is_some() {
                Stdio::piped()
            } else {
                Stdio::null()
            },
            Stdio::piped(),
            Stdio::piped(),
        )
    }

    fn spawn_with_stdio(&self, stdin: Stdio, stdout: Stdio, stderr: Stdio) -> io::Result<Child> {
        let mut process = Command::new(resolve_program(&self.program, self.current_dir.as_deref()));
        process
            .args(&self.args)
            .env_clear()
            .envs(self.env.iter().map(|(key, value)| (key, value)))
            .stdin(stdin)
            .stdout(stdout)
            .stderr(stderr)
            .kill_on_drop(true);
        if let Some(dir) = &self.current_dir {
            process.current_dir(dir);
        }

        #[cfg(unix)]
        {
            // Its own session, so the deadline can kill the whole group and not
            // just the program that happens to be holding the handle.
            unsafe {
                process.pre_exec(|| {
                    if libc::setsid() == -1 {
                        return Err(io::Error::last_os_error());
                    }
                    Ok(())
                });
            }
        }

        process.spawn()
    }
}

/// One output stream being read into a [`CapturedStream`].
///
/// Holds the finished capture once the reader task has produced it, so a drain
/// that has to be retried after a kill never polls a completed task again.
struct DrainingStream {
    task: JoinHandle<io::Result<CapturedStream>>,
    captured: Option<CapturedStream>,
}

impl DrainingStream {
    fn new(task: JoinHandle<io::Result<CapturedStream>>) -> Self {
        Self {
            task,
            captured: None,
        }
    }

    /// Whether the stream is fully read by `deadline`.
    ///
    /// A `false` return leaves the reader running: it is still holding the only
    /// copy of what arrived, and it will finish the moment the pipe closes.
    async fn drained_by(&mut self, deadline: Instant) -> io::Result<bool> {
        if self.captured.is_some() {
            return Ok(true);
        }
        match tokio::time::timeout_at(deadline, &mut self.task).await {
            Ok(joined) => {
                let captured = joined.map_err(|error| {
                    io::Error::other(format!("command output task failed: {error}"))
                })??;
                self.captured = Some(captured);
                Ok(true)
            }
            Err(_) => Ok(false),
        }
    }
}

struct Drain {
    stdout: DrainingStream,
    stderr: DrainingStream,
}

impl Drain {
    async fn drained_by(&mut self, deadline: Instant) -> io::Result<bool> {
        // Both, always: a stderr still arriving is as unfinished as a stdout,
        // and the second call is free once the first has landed.
        let stdout = self.stdout.drained_by(deadline).await?;
        let stderr = self.stderr.drained_by(deadline).await?;
        Ok(stdout && stderr)
    }

    /// The two captures. Only call after [`drained_by`](Self::drained_by)
    /// returned `true`.
    fn into_captures(self) -> (CapturedStream, CapturedStream) {
        let expect = "a drained stream has its capture";
        (
            self.stdout.captured.expect(expect),
            self.stderr.captured.expect(expect),
        )
    }
}

/// Kills the child's whole process tree and waits, briefly, for it to be reaped.
///
/// `group` is the id recorded at spawn, which is still needed when the program
/// itself has already exited and left the descendants this is here to end.
async fn kill_tree_and_reap(child: &mut Child, group: Option<u32>) -> io::Result<()> {
    kill_entire_process_tree(child, group)?;
    let _ = tokio::time::timeout(CLEANUP_GRACE, child.wait()).await;
    Ok(())
}

fn kill_entire_process_tree(child: &mut Child, group: Option<u32>) -> io::Result<()> {
    // `None` once the child has been waited on, which is exactly the case where
    // the descendants are still the problem.
    let live = child.id();

    #[cfg(unix)]
    {
        // Negative pid: the process group `setsid` gave the child, which is
        // every descendant that has not left it. The group outlives its leader
        // — an orphaned group keeps its id while any member is alive — so the
        // id recorded at spawn still names the right group after the program
        // that led it has been reaped.
        if let Some(pid) = live.or(group) {
            let result = unsafe { libc::kill(-(pid as i32), libc::SIGKILL) };
            if result == -1 {
                let error = io::Error::last_os_error();
                if error.raw_os_error() != Some(libc::ESRCH) {
                    return Err(error);
                }
            }
        }
    }

    #[cfg(windows)]
    {
        // Only while the handle is live: Windows has no orphaned-group handle
        // to aim at, and a pid the system may already have recycled is not a
        // thing to kill.
        let _ = group;
        if let Some(pid) = live {
            let status = StdCommand::new("taskkill")
                .args(["/PID", &pid.to_string(), "/T", "/F"])
                .status()?;
            if status.success() {
                return Ok(());
            }

            if child.try_wait()?.is_some() {
                return Ok(());
            }
        }
    }

    if live.is_none() {
        return Ok(());
    }

    child.start_kill()
}

fn missing_pipe(stream: &str) -> io::Error {
    io::Error::other(format!("failed to capture the child's {stream}"))
}

/// Where a relative program lives.
///
/// A path with a directory part is resolved against the working directory, so
/// `./hooks/guard.sh` means what it says regardless of where this process was
/// started — the platforms disagree about whether a relative program is
/// resolved before or after the child changes directory, and this removes the
/// question. A bare name is left alone for `PATH` to answer, which is what
/// someone writing `python3` expects.
fn resolve_program(program: &OsStr, current_dir: Option<&Path>) -> PathBuf {
    let path = Path::new(program);
    let has_directory = path
        .parent()
        .is_some_and(|parent| !parent.as_os_str().is_empty());

    match current_dir {
        Some(dir) if has_directory && path.is_relative() => dir.join(path),
        _ => path.to_path_buf(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    const CAP: usize = 64 * 1024;

    fn seconds(count: u64) -> Duration {
        Duration::from_secs(count)
    }

    #[cfg(unix)]
    fn exit_with_code_command() -> &'static str {
        "printf out; printf err >&2; exit 3"
    }

    #[cfg(windows)]
    fn exit_with_code_command() -> &'static str {
        "echo out& echo err 1>&2& exit 3"
    }

    #[cfg(unix)]
    fn minimal_shell_env() -> Vec<(String, String)> {
        vec![(
            "PATH".to_string(),
            std::env::var("PATH").expect("path available"),
        )]
    }

    #[cfg(windows)]
    fn minimal_shell_env() -> Vec<(String, String)> {
        ["PATH", "PATHEXT", "SystemRoot", "COMSPEC", "TEMP", "TMP"]
            .into_iter()
            .filter_map(|name| {
                std::env::var(name)
                    .ok()
                    .map(|value| (name.to_string(), value))
            })
            .collect()
    }

    #[tokio::test]
    async fn an_exit_code_and_both_streams_survive() {
        let completion = BoundedCommand::shell(exit_with_code_command(), seconds(5), CAP)
            .current_dir(std::env::temp_dir())
            .envs(minimal_shell_env())
            .run()
            .await
            .expect("the program is supervised");

        assert_eq!(completion.code(), Some(3), "{completion:?}");
        assert_eq!(completion.stdout().to_string_lossy().trim(), "out");
        assert_eq!(completion.stderr().to_string_lossy().trim(), "err");
        assert!(!completion.timed_out());
    }

    #[test]
    fn a_relative_program_is_resolved_against_the_working_directory() {
        assert_eq!(
            resolve_program(OsStr::new("./hooks/guard.sh"), Some(Path::new("/repo"))),
            PathBuf::from("/repo/./hooks/guard.sh")
        );
        assert_eq!(
            resolve_program(OsStr::new("python3"), Some(Path::new("/repo"))),
            PathBuf::from("python3"),
            "a bare name belongs to PATH"
        );
        assert_eq!(
            resolve_program(OsStr::new("./guard.sh"), None),
            PathBuf::from("./guard.sh"),
            "with no working directory there is nothing to resolve against"
        );
    }

    // Gated to unix: these fixtures are `/bin/sh` scripts, which is the
    // cheapest way to exercise a real process tree, a held pipe and a program
    // that ignores its stdin. The code under test is portable; the fixtures are
    // not, and inventing a Windows equivalent per case would test the fixture.
    #[cfg(unix)]
    mod unix {
        use super::*;

        /// Whether `pid` is gone, waiting up to `CLEANUP_GRACE` for the kernel
        /// to finish reaping it.
        async fn wait_until_dead(pid: i32) -> bool {
            let deadline = std::time::Instant::now() + CLEANUP_GRACE;
            loop {
                // Signal 0 asks the question without sending anything.
                if unsafe { libc::kill(pid, 0) } == -1 {
                    return true;
                }
                if std::time::Instant::now() >= deadline {
                    return false;
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        }

        #[tokio::test]
        async fn the_child_environment_holds_only_what_the_caller_passed() {
            // `/usr/bin/env` adds nothing of its own, so every name it prints
            // was either passed here or inherited — and inheriting is the bug.
            let completion = BoundedCommand::new("/usr/bin/env", seconds(5), CAP)
                .env("MENTRA_PASSED", "yes")
                .run()
                .await
                .expect("the program is supervised");

            let stdout = completion.stdout().to_string_lossy().into_owned();
            let listed: Vec<&str> = stdout
                .lines()
                .filter_map(|line| line.split_once('='))
                .map(|(name, _)| name)
                .collect();

            assert!(
                stdout.contains("MENTRA_PASSED=yes"),
                "the passed variable must arrive: {stdout}"
            );
            for (name, _) in std::env::vars() {
                assert!(
                    !listed.contains(&name.as_str()),
                    "`{name}` leaked from this process into the child: {stdout}"
                );
            }
        }

        #[tokio::test]
        async fn a_stdin_payload_reaches_the_program() {
            let completion = BoundedCommand::shell("cat", seconds(5), CAP)
                .envs(minimal_shell_env())
                .stdin("hello")
                .run()
                .await
                .expect("the program is supervised");

            assert_eq!(completion.code(), Some(0), "{completion:?}");
            assert_eq!(completion.stdout().to_string_lossy(), "hello");
        }

        #[tokio::test]
        async fn a_program_that_never_reads_stdin_still_answers() {
            // The deadlock this guards against needs a payload larger than the
            // pipe buffer; 256 KiB is comfortably past every platform's.
            let payload = "x".repeat(256 * 1024);

            let completion = BoundedCommand::shell("echo done", seconds(5), CAP)
                .envs(minimal_shell_env())
                .stdin(payload)
                .run()
                .await
                .expect("the program is supervised");

            assert_eq!(completion.code(), Some(0), "{completion:?}");
            assert_eq!(completion.stdout().to_string_lossy(), "done\n");
        }

        #[tokio::test]
        async fn stdout_is_capped_while_it_is_read() {
            // Far more output than the cap, printed as fast as the program can:
            // the bound has to hold during the read, not after it.
            let completion =
                BoundedCommand::shell("echo FIRST; seq 1 100000; echo LAST", seconds(30), 4096)
                    .envs(minimal_shell_env())
                    .run()
                    .await
                    .expect("the program is supervised");

            let stdout = completion.stdout();
            assert!(stdout.truncated(), "{completion:?}");
            assert!(stdout.len() <= 4096, "kept {} bytes", stdout.len());
            let text = stdout.to_string_lossy();
            assert!(text.starts_with("FIRST\n"), "{text}");
            assert!(text.ends_with("LAST\n"), "{text}");
            assert!(text.contains("bytes elided"), "{text}");
        }

        #[tokio::test]
        async fn stdout_and_stderr_caps_can_be_configured_independently() {
            let completion =
                BoundedCommand::shell("printf 1234567890; printf abcdefghij >&2", seconds(5), 4)
                    .envs(minimal_shell_env())
                    .max_stdout_bytes(9)
                    .max_stderr_bytes(3)
                    .run()
                    .await
                    .expect("the program is supervised");

            assert!(completion.stdout().truncated(), "{completion:?}");
            assert!(completion.stderr().truncated(), "{completion:?}");
            assert!(completion.stdout().len() <= 9, "{completion:?}");
            assert!(completion.stderr().len() <= 3, "{completion:?}");
        }

        #[tokio::test]
        async fn a_backgrounded_descendant_dies_with_the_process_group() {
            // The case a per-child kill misses: the shell outlives its budget
            // and has already started something that would outlive it too.
            let completion = BoundedCommand::shell(
                "sleep 60 & echo $!; sleep 60",
                Duration::from_millis(200),
                CAP,
            )
            .envs(minimal_shell_env())
            .run()
            .await
            .expect("the program is supervised");

            assert!(completion.timed_out(), "{completion:?}");
            let pid: i32 = completion
                .stdout()
                .to_string_lossy()
                .trim()
                .parse()
                .expect("the backgrounded pid was printed");
            assert!(
                wait_until_dead(pid).await,
                "the backgrounded descendant {pid} outlived the deadline"
            );
        }

        #[tokio::test]
        async fn a_descendant_holding_the_pipe_cannot_outlast_the_deadline() {
            // The program answers and exits at once, leaving a child holding
            // stdout open. Reading to EOF would wait for that child, so the
            // budget has to cover reading as well as waiting.
            let started = std::time::Instant::now();

            let completion =
                BoundedCommand::shell("sleep 60 & echo ready", Duration::from_millis(200), CAP)
                    .envs(minimal_shell_env())
                    .run()
                    .await
                    .expect("the program is supervised");

            assert!(completion.timed_out(), "{completion:?}");
            assert!(
                started.elapsed() < seconds(10),
                "the deadline, not the descendant, decides how long this takes: {:?}",
                started.elapsed()
            );
        }

        #[tokio::test]
        async fn a_missing_program_is_an_error_not_a_completion() {
            let error = BoundedCommand::new("/definitely/not/a/real/program", seconds(5), CAP)
                .run()
                .await
                .expect_err("cannot be started");

            assert_eq!(error.kind(), io::ErrorKind::NotFound);
        }
    }
}