rig-tools 0.45.0

Coding tools for Rig agents: read, write, edit, search and shell, and the file and process helpers they share.
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
//! Child processes started in a session of their own, such as `shell`
//! commands, so they never read the terminal and are killed as a group, and
//! the [`Launcher`] that says how every tool starts them. A [`Process`] it
//! [spawns](Launcher::spawn) takes its input, hands its output over line
//! by line, and is stopped gently or killed with its whole group.

use std::ffi::{OsStr, OsString};
use std::io::{self, Read, Write as _};
use std::path::PathBuf;
use std::process::{Child, Command, ExitStatus, Output, Stdio};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
use std::thread;
use std::time::{Duration, Instant};

/// How tools start processes: under a wrapper program (a sandbox such as
/// `bwrap … --`, `nice -n 10`, a container's `exec`), in a directory, with
/// variables set and removed. Every command it builds is [`detach`]ed. A
/// host that hands one launcher to every tool that starts processes
/// ([`Shell`](crate::Shell), a background job, a formatter) puts them all
/// under the same policy.
///
/// ```
/// use rig_tools::process::Launcher;
///
/// let launcher = Launcher::default()
///     .wrapped(["nice", "-n", "10"])
///     .in_dir("/srv/project")
///     .with_env("CI", "1")
///     .without_env(["AGENT_TOKEN"]);
/// let mut command = launcher.command("sh");
/// command.args(["-c", "make test"]);
/// assert_eq!(command.get_program(), "nice");
/// let args: Vec<_> = command.get_args().collect();
/// assert_eq!(args, ["-n", "10", "sh", "-c", "make test"]);
/// assert_eq!(command.get_current_dir(), Some(std::path::Path::new("/srv/project")));
/// ```
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Launcher {
    /// The program and arguments every command runs under; empty runs the
    /// command itself.
    pub prefix: Vec<OsString>,
    /// The directory commands start in; `None` is this process's.
    pub cwd: Option<PathBuf>,
    /// Variables set for every command.
    pub env: Vec<(OsString, OsString)>,
    /// Variables no command inherits, such as those that make a program
    /// act as the agent that runs it.
    pub unset_env: Vec<OsString>,
}

impl Launcher {
    /// The same launcher, running every command under `prefix`.
    pub fn wrapped(mut self, prefix: impl IntoIterator<Item = impl Into<OsString>>) -> Self {
        self.prefix = prefix.into_iter().map(Into::into).collect();
        self
    }

    /// The same launcher, starting commands in `dir`.
    pub fn in_dir(mut self, dir: impl Into<PathBuf>) -> Self {
        self.cwd = Some(dir.into());
        self
    }

    /// The same launcher, setting `name` to `value` for every command.
    pub fn with_env(mut self, name: impl Into<OsString>, value: impl Into<OsString>) -> Self {
        self.env.push((name.into(), value.into()));
        self
    }

    /// The same launcher, keeping `names` from every command.
    pub fn without_env(mut self, names: impl IntoIterator<Item = impl Into<OsString>>) -> Self {
        self.unset_env.extend(names.into_iter().map(Into::into));
        self
    }

    /// A detached command that runs `program` under this launcher's policy;
    /// add the program's arguments to it.
    pub fn command(&self, program: impl AsRef<OsStr>) -> Command {
        let mut command = match self.prefix.split_first() {
            Some((wrapper, arguments)) => {
                let mut command = Command::new(wrapper);
                command.args(arguments).arg(program);
                command
            }
            None => Command::new(program),
        };
        if let Some(dir) = &self.cwd {
            command.current_dir(dir);
        }
        for name in &self.unset_env {
            command.env_remove(name);
        }
        command.envs(self.env.iter().map(|(name, value)| (name, value)));
        detach(&mut command);
        command
    }

    /// Starts `program` with `args` under this launcher's policy, as
    /// [`command`](Self::command) builds it, with stdin and output as `how`
    /// says.
    ///
    /// ```
    /// # #[cfg(unix)] {
    /// use rig_tools::process::{Launcher, OutputLine, Spawn};
    ///
    /// let how = Spawn { stdin: Some(b"hello\n".to_vec()), ..Spawn::default() };
    /// let mut process = Launcher::default().spawn("cat", ["-"], how)?;
    /// let status = process.wait_for(std::time::Duration::from_secs(5))?;
    /// assert!(status.is_some_and(|status| status.success()));
    /// let lines: Vec<OutputLine> = process.lines().iter().collect();
    /// assert_eq!(lines, [OutputLine::Out("hello\n".to_owned())]);
    /// # }
    /// # Ok::<(), std::io::Error>(())
    /// ```
    pub fn spawn(
        &self,
        program: impl AsRef<OsStr>,
        args: impl IntoIterator<Item = impl AsRef<OsStr>>,
        how: Spawn,
    ) -> io::Result<Process> {
        let mut command = self.command(program);
        command.args(args).stdin(match how.stdin {
            Some(_) => Stdio::piped(),
            None => Stdio::null(),
        });
        let (sender, lines) = channel();
        let mut pipes: Vec<(Box<dyn Read + Send>, Tag)> = Vec::new();
        if how.merge_stderr {
            // One pipe for both keeps their order.
            let (reader, writer) = io::pipe()?;
            command.stdout(writer.try_clone()?).stderr(writer);
            pipes.push((Box::new(reader), OutputLine::Out));
        } else {
            command.stdout(Stdio::piped()).stderr(Stdio::piped());
        }
        let spawned = command.spawn();
        // The parent's copies of a shared pipe's write end must close, or
        // reading never ends.
        drop(command);
        let child = spawned?;
        let group = Group(Arc::new(AtomicU32::new(child.id())));
        // From here on, an error drops the process, which kills it.
        let mut process = Process {
            child,
            lines,
            group,
            remote: None,
            forced: false,
        };
        let (stdout, stderr) = (process.child.stdout.take(), process.child.stderr.take());
        #[cfg(unix)]
        let (stdout, stderr) = match (stdout, stderr) {
            (Some(stdout), Some(stderr)) => {
                let sender = sender.clone();
                thread::Builder::new()
                    .name("rig-process-output".to_owned())
                    .spawn(move || forward_both(stdout, stderr, &sender))?;
                (None, None)
            }
            apart => apart,
        };
        if let Some(stdout) = stdout {
            pipes.push((Box::new(stdout), OutputLine::Out));
        }
        if let Some(stderr) = stderr {
            pipes.push((Box::new(stderr), OutputLine::Err));
        }
        for (pipe, tag) in pipes {
            let sender = sender.clone();
            thread::Builder::new()
                .name("rig-process-output".to_owned())
                .spawn(move || forward(pipe, tag, &sender))?;
        }
        if let (Some(mut stdin), Some(bytes)) = (process.child.stdin.take(), how.stdin) {
            // Written on a thread of its own, so a child that reads slowly,
            // or not at all, never blocks the caller; closed once written.
            thread::Builder::new()
                .name("rig-process-input".to_owned())
                .spawn(move || {
                    stdin.write_all(&bytes).ok();
                })?;
        }
        Ok(process)
    }
}

/// How [`Launcher::spawn`] connects a process.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Spawn {
    /// What the process reads on stdin, which is closed after it; `None`
    /// gives it an empty stdin.
    pub stdin: Option<Vec<u8>>,
    /// Whether stderr goes into stdout's pipe, so the two keep their order
    /// and every line arrives as [`OutputLine::Out`].
    pub merge_stderr: bool,
}

/// A piece of a process's output, as [`Process::lines`] hands it over: a
/// line with its `\n`, or the end of the output without one. A line longer
/// than 64 KB comes in pieces. Bytes that are not UTF-8 are replaced.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum OutputLine {
    /// From stdout.
    Out(String),
    /// From stderr.
    Err(String),
}

impl OutputLine {
    /// The text, whichever stream it came from.
    pub fn text(&self) -> &str {
        match self {
            Self::Out(text) | Self::Err(text) => text,
        }
    }
}

type Tag = fn(String) -> OutputLine;

/// How often a waiting [`Process`] is checked.
const POLL: Duration = Duration::from_millis(10);

/// How long [`Process::run`] reads on after the process exited, for output
/// something it left running still writes, and again after a kill.
const PIPE_GRACE: Duration = Duration::from_millis(100);

/// How long a process asked to stop (`SIGTERM`) has to end before it is
/// killed: time for a transport that hangs up the command it runs on
/// another machine when it is stopped to do so.
pub const STOP_GRACE: Duration = Duration::from_secs(2);

/// The longest piece of a line [`forward`] hands over at once.
const MAX_PIECE: usize = 64 * 1024;

/// Reads `pipe` until every writer closed it, or the receiver is gone,
/// and sends what it reads on line by line.
///
/// When the receiver stops listening while a process that left the group
/// still holds the pipe, this thread stays blocked in `read` until that
/// process writes again or exits; its next write then finds the receiver
/// gone, the read end is dropped, and the process gets `EPIPE` or
/// `SIGPIPE`, as after any closed pipe.
fn forward(mut pipe: impl Read, tag: Tag, lines: &Sender<OutputLine>) {
    let mut pending = Vec::new();
    let mut buffer = [0; 8192];
    loop {
        let read = match pipe.read(&mut buffer) {
            Ok(0) => break,
            Ok(read) => read,
            Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
            Err(_) => break,
        };
        pending.extend_from_slice(buffer.get(..read).unwrap_or_default());
        let mut start = 0;
        while let Some(end) = pending.get(start..).and_then(piece_end) {
            let piece = pending.get(start..start + end).unwrap_or_default();
            if lines
                .send(tag(String::from_utf8_lossy(piece).into_owned()))
                .is_err()
            {
                return;
            }
            start += end;
        }
        pending.drain(..start);
    }
    if !pending.is_empty() {
        lines
            .send(tag(String::from_utf8_lossy(&pending).into_owned()))
            .ok();
    }
}

/// Reads a process's stdout and stderr on one thread, whichever has data
/// first, stdout first when both have, and sends each on line by line as
/// [`forward`] does: so the lines of the two keep the order they were
/// written in, as far as one reader can tell, and a line one writes in
/// pieces is never split by the other's.
#[cfg(unix)]
fn forward_both(
    stdout: std::process::ChildStdout,
    stderr: std::process::ChildStderr,
    lines: &Sender<OutputLine>,
) {
    use std::os::fd::AsRawFd;

    let fds = [stdout.as_raw_fd(), stderr.as_raw_fd()];
    let mut pipes: [(Box<dyn Read>, Tag, Vec<u8>, bool); 2] = [
        (Box::new(stdout), OutputLine::Out, Vec::new(), true),
        (Box::new(stderr), OutputLine::Err, Vec::new(), true),
    ];
    let mut buffer = [0; 8192];
    while pipes.iter().any(|(.., open)| *open) {
        let mut polled = fds.map(|fd| libc::pollfd {
            fd,
            events: libc::POLLIN,
            revents: 0,
        });
        for (poll, (.., open)) in polled.iter_mut().zip(&pipes) {
            if !*open {
                poll.fd = -1;
            }
        }
        // SAFETY: `polled` is an array of two initialized `pollfd`s that
        // outlives the call, and its length is passed with it.
        let ready = unsafe { libc::poll(polled.as_mut_ptr(), 2, -1) };
        if ready < 0 {
            if io::Error::last_os_error().kind() == io::ErrorKind::Interrupted {
                continue;
            }
            break;
        }
        for (poll, (pipe, tag, pending, open)) in polled.iter().zip(&mut pipes) {
            if !*open || poll.revents == 0 {
                continue;
            }
            let read = match pipe.read(&mut buffer) {
                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
                Ok(0) | Err(_) => None,
                Ok(read) => Some(read),
            };
            let Some(read) = read else {
                *open = false;
                if !pending.is_empty() {
                    let text = String::from_utf8_lossy(pending).into_owned();
                    pending.clear();
                    if lines.send(tag(text)).is_err() {
                        return;
                    }
                }
                continue;
            };
            pending.extend_from_slice(buffer.get(..read).unwrap_or_default());
            let mut start = 0;
            while let Some(end) = pending.get(start..).and_then(piece_end) {
                let piece = pending.get(start..start + end).unwrap_or_default();
                if lines
                    .send(tag(String::from_utf8_lossy(piece).into_owned()))
                    .is_err()
                {
                    return;
                }
                start += end;
            }
            pending.drain(..start);
        }
    }
}

/// Where the first piece of `bytes` ends: after its first `\n`, else after
/// [`MAX_PIECE`] bytes, not inside a character; `None` while it is shorter.
fn piece_end(bytes: &[u8]) -> Option<usize> {
    if let Some(at) = bytes.iter().position(|&byte| byte == b'\n') {
        return Some(at + 1);
    }
    let head = bytes.get(..MAX_PIECE)?;
    match std::str::from_utf8(head) {
        Err(error) if error.error_len().is_none() && error.valid_up_to() > 0 => {
            Some(error.valid_up_to())
        }
        _ => Some(head.len()),
    }
}

/// A process [`Launcher::spawn`] started, the leader of a process group of
/// its own. Dropping it while it runs kills the group and reaps it.
pub struct Process {
    child: Child,
    lines: Receiver<OutputLine>,
    group: Group,
    /// The machine it is the local end of a transport to, if it is one.
    remote: Option<String>,
    /// Whether the last stop had to kill it.
    forced: bool,
}

impl Process {
    /// The process's id, which is also its group's.
    pub fn id(&self) -> u32 {
        self.child.id()
    }

    /// The same process, marked as the local end of a transport that runs
    /// the command on the machine `host`: stopping it stops the transport,
    /// and what that does there is the transport's to say.
    pub fn through(mut self, host: impl Into<String>) -> Self {
        self.remote = Some(host.into());
        self
    }

    /// The machine it runs the command on through a transport, if it was
    /// marked so ([`through`](Self::through)).
    pub fn remote(&self) -> Option<&str> {
        self.remote.as_deref()
    }

    /// Whether the last [`stop`](Self::stop) had to kill it: it was still
    /// running its grace period after `SIGTERM`.
    pub fn forced(&self) -> bool {
        self.forced
    }

    /// Its output, line by line, as it comes. Lines wait here until they
    /// are received; the channel disconnects once every process holding
    /// the output closed it.
    pub fn lines(&self) -> &Receiver<OutputLine> {
        &self.lines
    }

    /// A handle that stops its group from another thread, such as the one
    /// dropping the call that waits for it.
    pub fn group(&self) -> Group {
        self.group.clone()
    }

    /// How it ended, if it has.
    pub fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
        let status = self.child.try_wait();
        if !matches!(status, Ok(None)) {
            // Reaped, or about to be: its id may be reused.
            self.group.0.store(0, Ordering::SeqCst);
        }
        status
    }

    /// Waits at most `timeout` for it to end, and says how it ended if it
    /// did. Its output keeps arriving at [`lines`](Self::lines) meanwhile.
    /// A timeout too long for the clock, such as `Duration::MAX`, waits
    /// until it ends.
    pub fn wait_for(&mut self, timeout: Duration) -> io::Result<Option<ExitStatus>> {
        let deadline = Instant::now().checked_add(timeout);
        // A process whose output just closed is usually a moment from
        // ending, so the first naps are short.
        let mut nap = Duration::from_millis(1);
        loop {
            if let Some(status) = self.try_wait()? {
                return Ok(Some(status));
            }
            let left = deadline.map_or(POLL, |deadline| {
                deadline.saturating_duration_since(Instant::now())
            });
            if left.is_zero() {
                return Ok(None);
            }
            thread::sleep(left.min(nap));
            nap = (nap * 2).min(POLL);
        }
    }

    /// Runs it to its end, at most `timeout` (`None` waits as long as it
    /// takes), and returns what it printed on stdout and on stderr (with
    /// [`Spawn::merge_stderr`], all of it on stdout) and how it ended. A
    /// blocking call: run it on a thread of its own, such as inside
    /// [`blocking`](crate::blocking). Past `timeout` the error is
    /// [`io::ErrorKind::TimedOut`]; otherwise as [`run`](Self::run).
    ///
    /// ```
    /// # #[cfg(unix)] {
    /// use rig_tools::process::{Launcher, Spawn};
    ///
    /// let process = Launcher::default().spawn("sh", ["-c", "echo hi"], Spawn::default())?;
    /// let output = process.output(Some(std::time::Duration::from_secs(5)))?;
    /// assert_eq!(output.stdout, b"hi\n");
    /// assert!(output.status.success());
    /// # }
    /// # Ok::<(), std::io::Error>(())
    /// ```
    pub fn output(mut self, timeout: Option<Duration>) -> io::Result<Output> {
        let deadline = timeout.and_then(|timeout| Instant::now().checked_add(timeout));
        let (mut stdout, mut stderr) = (Vec::new(), Vec::new());
        let ended = self.run(deadline, None, |line| {
            match line {
                OutputLine::Out(text) => stdout.extend(text.into_bytes()),
                OutputLine::Err(text) => stderr.extend(text.into_bytes()),
            }
            true
        })?;
        match ended {
            Ended::Exited(status) => Ok(Output {
                status,
                stdout,
                stderr,
            }),
            Ended::TimedOut | Ended::Quiet | Ended::Stopped => Err(io::Error::new(
                io::ErrorKind::TimedOut,
                "the process ran past its timeout and was killed",
            )),
        }
    }

    /// Waits for it to end, at most until `deadline` (`None` waits as long
    /// as it takes) and, with `quiet`, while it printed something within
    /// that long, handing each piece of its output to `sink`; `sink`
    /// returning `false` stops the wait. A blocking call.
    ///
    /// After an exit, something it left running that still holds its
    /// output gets a moment (100 ms) to finish, then the rest of its group
    /// is killed; past the deadline or stopped, the group is
    /// [stopped](Self::stop) with [`STOP_GRACE`]. After a kill, reading
    /// goes on for that moment: a process that left the group survives the
    /// kill and may hold the pipe for good.
    pub fn run(
        &mut self,
        deadline: Option<Instant>,
        quiet: Option<Duration>,
        mut sink: impl FnMut(OutputLine) -> bool,
    ) -> io::Result<Ended> {
        let mut more = true;
        let mut heard = Instant::now();
        let ended = loop {
            if let Some(status) = self.try_wait()? {
                break Ended::Exited(status);
            }
            if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
                break Ended::TimedOut;
            }
            if quiet.is_some_and(|quiet| heard.elapsed() >= quiet) {
                break Ended::Quiet;
            }
            if !more {
                break Ended::Stopped;
            }
            match self.lines.recv_timeout(POLL) {
                Ok(line) => {
                    heard = Instant::now();
                    more = sink(line);
                }
                Err(RecvTimeoutError::Timeout) => {}
                // It closed its output but still runs.
                Err(RecvTimeoutError::Disconnected) => {
                    self.wait_for(POLL)?;
                }
            }
        };
        if !matches!(ended, Ended::Exited(_)) {
            self.stop(STOP_GRACE).ok();
            self.read_rest(&mut sink);
        } else if !self.read_rest(&mut sink) {
            self.kill().ok();
            self.read_rest(&mut sink);
        }
        Ok(ended)
    }

    /// Hands output to `sink` until the pipe's end, for at most
    /// [`PIPE_GRACE`]; whether the end came.
    fn read_rest(&self, sink: &mut impl FnMut(OutputLine) -> bool) -> bool {
        let deadline = Instant::now() + PIPE_GRACE;
        loop {
            match self
                .lines
                .recv_timeout(deadline.saturating_duration_since(Instant::now()))
            {
                Ok(line) => {
                    sink(line);
                }
                Err(RecvTimeoutError::Disconnected) => return true,
                Err(RecvTimeoutError::Timeout) => return false,
            }
        }
    }

    /// Asks its group to stop (`SIGTERM`), waits at most `grace` for it to
    /// end, then [kills](Self::kill) the group. Elsewhere than unix it is
    /// killed at once.
    pub fn stop(&mut self, grace: Duration) -> io::Result<ExitStatus> {
        self.forced = false;
        if self.try_wait()?.is_none() {
            self.group.terminate();
            self.forced = self.wait_for(grace)?.is_none();
        }
        self.kill()
    }

    /// Kills its group (`SIGKILL`), also what is left of it after the
    /// process itself ended, and says how the process ended. Kill within
    /// moments of the end: once the group is empty its id is free again.
    pub fn kill(&mut self) -> io::Result<ExitStatus> {
        self.group.0.store(0, Ordering::SeqCst);
        #[cfg(unix)]
        kill_group_of(self.child.id());
        #[cfg(not(unix))]
        if self.child.try_wait()?.is_none() {
            self.child.kill()?;
        }
        self.child.wait()
    }
}

/// How [`Process::run`]'s wait ended.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Ended {
    /// It exited.
    Exited(ExitStatus),
    /// It ran past the deadline, and its group was stopped.
    TimedOut,
    /// It printed nothing for the `quiet` time, and its group was stopped.
    Quiet,
    /// The sink stopped the wait, and its group was stopped.
    Stopped,
}

impl Drop for Process {
    fn drop(&mut self) {
        if self.group.0.load(Ordering::SeqCst) != 0 {
            self.kill().ok();
        }
    }
}

/// A [`Process`]'s group, to stop from another thread. It acts only while
/// the process runs: once it was reaped, its id may name another process.
#[derive(Clone, Debug)]
pub struct Group(Arc<AtomicU32>);

impl Group {
    /// Kills the group (`SIGKILL`); elsewhere than unix, nothing.
    pub fn kill(&self) {
        self.signal(Signal::Kill);
    }

    /// Asks the group to stop (`SIGTERM`); elsewhere than unix, nothing.
    pub fn terminate(&self) {
        self.signal(Signal::Terminate);
    }

    fn signal(&self, signal: Signal) {
        let leader = self.0.load(Ordering::SeqCst);
        #[cfg(unix)]
        if leader != 0 {
            signal_group(leader, signal);
        }
        #[cfg(not(unix))]
        let _ = (leader, signal);
    }
}

/// Stops a [`Process`]'s group when dropped: kept in a future that waits
/// for the process on another thread ([`blocking`](crate::blocking)), it
/// asks the group to stop (`SIGTERM`) the moment the future is dropped,
/// such as when a call is interrupted or the app quits, and kills it
/// [`STOP_GRACE`] later if it still runs. The thread waiting for the
/// process must go on reaping it meanwhile, as [`Process::run`] does.
pub struct KillOnDrop(pub Group);

impl Drop for KillOnDrop {
    fn drop(&mut self) {
        self.0.terminate();
        let group = self.0.clone();
        let later = thread::Builder::new()
            .name("rig-process-stop".to_owned())
            .spawn(move || {
                thread::sleep(STOP_GRACE);
                group.kill();
            });
        if later.is_err() {
            self.0.kill();
        }
    }
}

/// The signals a [`Group`] sends.
#[derive(Clone, Copy)]
enum Signal {
    Terminate,
    Kill,
}

/// Starts `command` in a new session, which also makes it the leader of a
/// new process group. With no controlling terminal, a program that opens
/// `/dev/tty` for a prompt (ssh, git, sudo) fails at once instead of
/// drawing over the view and stopping on a terminal read. Git is also told
/// not to prompt.
pub fn detach(command: &mut Command) {
    command.env("GIT_TERMINAL_PROMPT", "0");
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        // SAFETY: the closure runs in the forked child before `exec` and
        // only calls `setsid`, which is async-signal-safe.
        unsafe {
            command.pre_exec(|| {
                if libc::setsid() == -1 {
                    Err(std::io::Error::last_os_error())
                } else {
                    Ok(())
                }
            });
        }
    }
}

/// Kills the process group `child` leads, created by [`detach`];
/// elsewhere, the child alone.
#[cfg(unix)]
pub fn kill_group(child: &mut Child) {
    kill_group_of(child.id());
}

/// Kills the process group led by the process `leader`. Its id names that
/// group while the leader is not reaped or any member lives: the kernel
/// does not reuse the id of a process group that exists. Once the leader
/// was reaped and the group emptied, the id is free again, so callers kill
/// within moments of the reap.
#[cfg(unix)]
pub fn kill_group_of(leader: u32) {
    signal_group(leader, Signal::Kill);
}

/// Sends `signal` to the process group led by the process `leader`.
#[cfg(unix)]
fn signal_group(leader: u32, signal: Signal) {
    let signal = match signal {
        Signal::Terminate => libc::SIGTERM,
        Signal::Kill => libc::SIGKILL,
    };
    if let Ok(group) = i32::try_from(leader) {
        // SAFETY: `kill` takes plain integers; a negative pid names the
        // process group the child leads, created by `detach`.
        unsafe {
            libc::kill(-group, signal);
        }
    }
}

#[cfg(not(unix))]
pub fn kill_group(child: &mut Child) {
    child.kill().ok();
}

#[cfg(test)]
mod tests;