processkit — documentation

processkit is a child-process toolkit for Rust in two layers:

┌─────────────────────────────────────────────────────────────────┐
│  Runner layer (async, tokio)                                    │
│  Command · RunningProcess · Pipeline · Supervisor · CliClient   │
│  capture / streaming / interactive stdin / readiness probes     │
│  testing seam: ProcessRunner → ScriptedRunner / RecordReplay…   │
├─────────────────────────────────────────────────────────────────┤
│  Group layer (kill-on-drop containment)                         │
│  ProcessGroup: spawn / adopt / signal / suspend / members /     │
│  stats / limits / shutdown                                      │
├─────────────────────────────────────────────────────────────────┤
│  OS mechanisms                                                  │
│  Windows Job Object · Linux cgroup v2 · POSIX process group     │
└─────────────────────────────────────────────────────────────────┘

Every Command run gets containment for free: the one-shot helpers spawn into a fresh private group that dies with the run, so a panicking caller never leaks a process tree. The layers are also usable independently — a raw ProcessGroup can contain children you spawn yourself, and the runner's test doubles never touch the OS at all.

Guides

New to the crate? Start with the Cookbook — short task-to-snippet recipes for everything the crate does — then read Running commands end to end (it's the vocabulary every other guide builds on). Reach for the rest as the need arises, and keep Platform support handy before you ship: it collects every per-OS caveat in one place.

Feature flags

Every flag is additive and only gates visibility — the kill-on-drop tree guarantee is unconditional in every configuration.

[dependencies]
processkit = { version = "…", features = ["limits"] }

The 60-second tour

use processkit::{Command, ProcessGroup, Stdin};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // One-shot: capture everything. A non-zero exit is data, not an Err.
    let head = Command::new("git").args(["rev-parse", "HEAD"]).output_string().await?;
    println!("HEAD = {}", head.stdout().trim());

    // Success-checking: non-zero exit / timeout / signal-kill become typed errors.
    let version = Command::new("cargo").arg("--version").run().await?;

    // Stdin, timeout, streaming, pipelines, supervision … see the guides.
    let sorted = Command::new("sort")
        .stdin(Stdin::from_string("b\na\n"))
        .timeout(std::time::Duration::from_secs(5))
        .run()
        .await?;

    // Containment: anything spawned through a group dies with it.
    let group = ProcessGroup::new()?;
    let _server = group.start(&Command::new("dev-server")).await?;
    drop(group); // the server — and everything *it* spawned — is reaped

    let _ = (version, sorted);
    Ok(())
}

API reference

The rustdoc on docs.rs is the authoritative per-item reference; these guides are the narrative layer on top — they explain how the pieces compose, with the platform fine print collected in Platform support.

These examples are compiler-checked. Every fenced Rust block across these guides and the root README.md is compiled (and, unless annotated no_run or ignore, actually run) as an ordinary doctest by cargo test --all-features (as CI does) — a signature change that stops matching a guide's snippet fails CI instead of silently lying to a reader. The hidden harness only builds under --all-features, so a plain cargo test with the default features does not exercise this check. See src/doc_examples.rs for the (test-only, hidden) harness.

Other languages

Not on Rust? processkit-py wraps this crate's core in a Python (PyO3/asyncio) API — this crate stays the single source of truth for the containment/runner logic underneath.

Python wrapper

Redirecting to the Python wrapper documentation...

.NET version

Redirecting to the .NET version documentation...

Cookbook

‹ docs index

Task-oriented recipes: find the thing you're trying to do, copy the snippet, follow the link when you need the fine print. Every snippet assumes a tokio runtime and use processkit::Command; unless shown otherwise.

• Run a command and get its output
• Inspect a failure instead of erroring
• Ask a yes/no question
• Accept non-zero exit codes as success
• Bound a run with a timeout
• Let a tool clean up on timeout
• Show a useful error message
• Check a tool is installed without running it
• Feed the child's stdin
• Stream output as it arrives
• Talk to an interactive child
• Pipe commands without a shell
• Start a server and wait until it's ready
• Tear down several children as a unit
• React to whichever child exits first
• Sandbox an untrusted tool
• Keep a crash-prone service running
• Retry a flaky command
• Cancel runs on shutdown
• Measure what a run cost
• Contain a process you didn't spawn
• Test code that runs processes — without processes
• Test streaming code — without processes
• Wrap a CLI tool behind a typed API

Run a command and get its output

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let head = Command::new("git").args(["rev-parse", "HEAD"]).run().await?;
    Ok(())
}

run() requires a zero exit and returns stdout with trailing whitespace trimmed; a non-zero exit, spawn failure, or timeout is a typed Error. For a one-liner without the builder: processkit::run("git", ["rev-parse", "HEAD"]).

Fine print: Running commands → consuming verbs.

Inspect a failure instead of erroring

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let result = Command::new("git").args(["merge", "topic"]).output_string().await?;
    if !result.is_success() {
        eprintln!("merge exited {:?}: {}", result.code(), result.stderr());
    }
    Ok(())
}

output_string() (and output_bytes() for raw bytes) treats the exit code as data — Err means the run couldn't happen at all. Call result.ensure_success()? later to convert a stored failure into the same typed error run() would have produced.

Fine print: Running commands → results and errors.

Ask a yes/no question

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let dirty = !Command::new("git").args(["diff", "--quiet"]).probe().await?;
    Ok(())
}

probe() maps exit 0 → true, exit 1 → false, and anything else to an error — the git diff --quiet / grep -q convention without manual code matching.

Accept non-zero exit codes as success

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // `grep` exits 1 when it finds no match — not a failure for this call.
    let found = Command::new("grep")
        .args(["needle", "haystack.txt"])
        .ok_codes([0, 1])
        .output_string()
        .await?;
    let matched = found.code() == Some(0); // 0 = matched, 1 = no match (both "success")
    Ok(())
}

ok_codes widens what the checking verbs (run/run_unit) and is_success/ensure_success treat as success — for tools whose non-zero exit is a normal result (grep 1 = no match, diff 1 = differs, rsync's code families). It does not change exit_code (always the raw code) or probe (always the 0/1 convention). An empty set is ignored, so the default stays exit 0.

Bound a run with a timeout

use processkit::Command;
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let result = Command::new("slow-tool")
        .timeout(Duration::from_secs(30))
        .output_string()
        .await?;
    if result.timed_out() {
        eprintln!("gave up after 30s; partial output: {}", result.stdout());
    }
    Ok(())
}

At the deadline the whole tree is killed. On the capture verbs the timeout is captured (timed_out(), partial output kept); on the success-checking verbs (run, exit_code) it surfaces as Error::Timeout.

Let a tool clean up on timeout

use processkit::Command;
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let result = Command::new("dev-server")
        .timeout(Duration::from_secs(30))
        .timeout_grace(Duration::from_secs(5)) // SIGTERM, wait up to 5s, then SIGKILL
        .output_string()
        .await?;
    Ok(())
}

timeout_grace turns the hard deadline kill into a graceful one: SIGTERM (or the signal from timeout_signal, with the process-control feature), up to the grace window to exit, then SIGKILL. A signal-handling child exits early; timed_out() stays true. Windows has no signal tier — the deadline kills atomically.

Fine print: Timeouts → graceful timeout.

Fine print: Timeouts, retries & cancellation.

Show a useful error message

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    if let Err(e) = Command::new("git").args(["merge", "topic"]).run().await {
        eprintln!("merge failed: {}", e.diagnostic().unwrap_or("(no output)"));
    }
    Ok(())
}

Error::diagnostic() picks the most explanatory captured text — stderr, falling back to stdout (git writes CONFLICT … there) — so callers don't re-implement the same heuristic.

Check a tool is installed without running it

use processkit::Command;

fn main() {
    // The crate-level shortcut for a bare tool:
    match processkit::which("git") {
        Ok(path) => println!("git is at {}", path.display()),
        Err(e) if e.is_not_found() => eprintln!("git is not installed"),
        Err(e) => eprintln!("could not resolve git: {e}"),
    }

    // On a builder it honors that command's `prefer_local` / env, resolving
    // exactly what a real run would launch:
    let _ = Command::new("eslint")
        .prefer_local("./node_modules/.bin")
        .resolve_program();
}

which / Command::resolve_program() locate a program without spawning it — a side-effect-free doctor / preflight check ("is the tool installed?") for a friendly up-front error. Resolution reuses the crate's own launch-path logic (the same PATH/PATHEXT/execute-bit and prefer_local handling a real run uses), so a hit is exactly what would be launched and a miss is exactly the Error::NotFound (is_not_found()) a run would raise. It is synchronous — no tokio runtime needed. A wrapped tool's client offers the same via CliClient::resolve_program().

Fine print: Running commands → preflight.

Feed the child's stdin

use processkit::Command;
use processkit::Stdin;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // A string you already have:
    let sorted = Command::new("sort")
        .stdin(Stdin::from_string("banana\napple\n"))
        .run()
        .await?;

    // …or any async source: a reader (file, socket) or a stream of lines.
    let from_file = Stdin::from_reader(tokio::fs::File::open("input.txt").await?);
    let from_chan = Stdin::from_lines(tokio_stream::iter(vec!["one".to_owned()]));
    let _ = (sorted, from_file, from_chan);
    Ok(())
}

One-shot sources (from_reader/from_lines) feed a single run; re-running the same Command afterwards fails loud (an Error::Io at launch, D10) instead of silently seeing empty stdin. For a conversation, see the next recipe but one.

Fine print: Running commands → standard input.

Stream output as it arrives

use processkit::Command;
use processkit::Finished;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    use processkit::prelude::StreamExt; // re-exported; provides `.next()`

    let mut run = Command::new("cargo").args(["build", "--verbose"]).start().await?;
    let mut lines = run.stdout_lines()?;
    while let Some(line) = lines.next().await {
        println!("build: {line}");
    }
    let Finished { outcome, stderr, .. } = run.finish().await?; // outcome + buffered stderr
    Ok(())
}

No waiting for exit, no full-output buffering; stderr is drained in the background so the child can't block. A timeout on the command bounds the stream itself. Prefer a callback? .on_stdout_line(|l| …) runs one per line while any capture verb drives the run.

Fine print: Streaming & interactive I/O.

Talk to an interactive child

use processkit::Command;
use processkit::prelude::StreamExt;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut run = Command::new("bc").keep_stdin_open().start().await?;
    let mut stdin = run.take_stdin().expect("stdin was kept open");
    stdin.write_line("2 + 2").await?;
    stdin.finish().await?; // EOF — bc exits

    let mut answers = run.stdout_lines()?;
    while let Some(answer) = answers.next().await {
        println!("{answer}");
    }
    Ok(())
}

keep_stdin_open() hands you an async writer instead of closing stdin at spawn; interleave writes with reads for request/response tools. Its writer methods return std::io::Result (idiomatic for a writer) — convert with .map_err(processkit::Error::Io)? in a processkit::Result function, or use Box<dyn std::error::Error>.

Fine print: Streaming & interactive I/O → interactive stdin.

Pipe commands without a shell

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let authors = Command::new("git").args(["log", "--format=%an"])
        .pipe(Command::new("sort"))
        .pipe(Command::new("uniq").arg("-c"))
        .output_string()
        .await?;
    Ok(())
}

Native pipes — no shell string, no quoting, no injection surface. The outcome is pipefail: stdout comes from the last stage, the reported failure from the first stage that didn't exit cleanly. All stages share one kill-on-drop group. The | operator is equivalent sugar: (a | b | c).output_string().

For a consumer that legitimately stops reading early — the | head -1 shape, where the producer's broken-pipe death (its next write fails once the downstream closes, or SIGPIPE where the OS delivers it) is expected — mark the producer unchecked_in_pipe() so that death doesn't fail the chain:

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let first = (Command::new("seq").args(["1", "1000000"]).unchecked_in_pipe()
        | Command::new("head").args(["-n", "1"]))
        .run()
        .await?;
    Ok(())
}

Fine print: Pipelines → unchecked stages.

Start a server and wait until it's ready

use processkit::Command;
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let mut server = Command::new("my-server").args(["--port", "8080"]).start().await?;

    // Pick the probe that matches how the server announces readiness:
    server.wait_for_line(|l| l.contains("listening"), Duration::from_secs(10)).await?;
    // server.wait_for_port("127.0.0.1:8080".parse().unwrap(), Duration::from_secs(10)).await?;
    // server.wait_for(|| async { http_health().await }, Duration::from_secs(10)).await?;

    // …use the server; dropping `server` kills its whole tree.
    Ok(())
}

A probe that can't succeed fails fast with Error::NotReady and never kills the child — you decide what happens next. No more sleep(2) and hoping.

Fine print: Streaming & interactive I/O → readiness probes.

Tear down several children as a unit

use processkit::Command;
use processkit::ProcessGroup;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::new()?;
    let _db = group.start(&Command::new("dev-db")).await?;
    let _api = group.start(&Command::new("dev-api")).await?;

    // Either: graceful — SIGTERM, bounded wait, optional SIGKILL escalation…
    group.shutdown().await?;
    // …or just drop(group): hard kill-on-drop of everything, grandchildren included.
    Ok(())
}

The group is the unit of fate: a panic or early return anywhere reaps every member. Configure the grace window via ProcessGroupOptions.

Fine print: Process groups.

React to whichever child exits first

use processkit::Command;
use processkit::{ProcessGroup, wait_any};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::new()?;
    let mut a = group.start(&Command::new("worker-a")).await?;
    let mut b = group.start(&Command::new("worker-b")).await?;

    let (idx, outcome) = wait_any(&mut [&mut a, &mut b]).await?;
    println!("worker #{idx} exited first with {outcome:?}");
    // `a` and `b` are only borrowed — the loser is still usable here.
    Ok(())
}

Fine print: Streaming & interactive I/O → racing children.

Sandbox an untrusted tool

use processkit::Command;
use processkit::{ProcessGroup, ProcessGroupOptions};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // Cap the whole tree (requires the `limits` feature; Windows Job / Linux cgroup):
    let group = ProcessGroup::with_options(
        ProcessGroupOptions::default()
            .max_memory(512 * 1024 * 1024)
            .max_processes(64)
            .cpu_quota(0.5),
    )?;

    let result = group
        .start(
            &Command::new("untrusted-tool")
                .inherit_env(["PATH"]) // allow-list: everything else is cleared
                .timeout(std::time::Duration::from_secs(60)),
        )
        .await?
        .output_string()
        .await?;
    Ok(())
}

Unenforceable limits are a hard Error::ResourceLimit, never a silently unbounded group. On Unix, add .uid(…)/.gid(…) to drop privileges (note the cgroup-mechanism caveat in the guide).

Fine print: Process groups → resource limits · Running commands → privileges.

Keep a crash-prone service running

use processkit::Command;
use processkit::{RestartPolicy, Supervisor};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let outcome = Supervisor::new(Command::new("my-service"))
        .restart(RestartPolicy::OnCrash)
        .max_restarts(5)
        .backoff(Duration::from_millis(200), 2.0)
        .storm_pause(Duration::from_secs(15)) // crash-loop guard (off by default)
        .run()
        .await?;
    println!(
        "stopped after {} restarts ({} storm pauses): {:?}",
        outcome.restarts, outcome.storm_pauses, outcome.stopped
    );
    Ok(())
}

Exponential backoff with jitter by default; stop_when(…) ends supervision on a condition; .with_runner(&group) keeps every incarnation inside one shared kill-on-drop group. storm_pause arms the failure-storm guard: failures feed a decaying score, and past the threshold the supervisor takes one collective pause instead of hammering restarts — "fails rarely" and "crash-looping" stop being the same case.

Fine print: Supervision, failure storms.

Retry a flaky command

use processkit::Command;
use processkit::Error;
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let fetched = Command::new("git")
        .args(["fetch", "--quiet"])
        .timeout(Duration::from_secs(10))
        .retry(3, Duration::from_millis(200), |e| {
            matches!(e, Error::Timeout { .. })
                || e.diagnostic().is_some_and(|m| m.contains("Could not resolve host"))
        })
        .run()
        .await?;
    Ok(())
}

The classifier sees the typed error and decides whether this failure is worth another attempt; each attempt is a fresh process. retry replays a run to success — for keeping a process alive, use a Supervisor (previous recipe).

Fine print: Timeouts, retries & cancellation → retry.

Cancel runs on shutdown

use processkit::Command;
use processkit::CancellationToken;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let token = CancellationToken::new();

    let job = tokio::spawn({
        let token = token.child_token();
        async move { Command::new("long-job").cancel_on(token).run().await }
    });

    // On Ctrl-C / shutdown signal / sibling failure:
    token.cancel(); // kills the tree; the run resolves to Error::Cancelled
    let outcome = job.await; // Err(Error::Cancelled { .. }) inside
    Ok(())
}

Cancellation is always an error (the run was abandoned, there is no result), beats a simultaneous timeout, and is terminal for retry and Supervisor alike.

For a typed wrapper whose commands never cross your code, set the token once on the client — every command it builds carries it:

#![allow(unused)]
fn main() {
use processkit::{CancellationToken, CliClient};

let token = CancellationToken::new();
let gh = CliClient::new("gh").default_cancel_on(token.child_token());
// token.cancel() → every in-flight command of THIS client dies.
}

Fine print: Timeouts, retries & cancellation → cancellation, client-level default.

Measure what a run cost

use processkit::Command;
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // One run, summarized (requires the opt-in `stats` feature):
    let profile = Command::new("crunch").start().await?.profile(Duration::from_millis(100)).await?;
    println!("outcome={:?} took={:?} peak_rss={:?} avg_cpu_cores={:?}",
        profile.outcome, profile.duration, profile.peak_memory_bytes, profile.avg_cpu_cores());
    Ok(())
}

For a live series over a whole group, group.sample_stats(every) yields a Stream of snapshots. CPU/memory need a real container (Windows Job / Linux cgroup); elsewhere you still get process counts.

Fine print: Process groups → stats · Streaming → profiling.

Contain a process you didn't spawn

use processkit::{Error, ProcessGroup};

fn main() -> processkit::Result<()> {
    // `tokio`/`std` calls return `io::Error`, which the crate does NOT auto-convert
    // into `processkit::Error` (there is no blanket `From<io::Error>`, by design) —
    // map it explicitly, or use a `Box<dyn std::error::Error>` / `anyhow` return in
    // your own code so both error types `?` freely.
    let child = tokio::process::Command::new("legacy-launcher")
        .spawn()
        .map_err(Error::Io)?;

    let group = ProcessGroup::new()?; // `adopt` is part of `process-control` (default-on)
    group.adopt(&child)?;            // from now on the group's teardown covers it
    Ok(())
}

Adoption is best-effort by mechanism — on Windows/cgroup the whole running tree joins; on the POSIX process-group backends an exec'd child is contained individually (its future forks too, where it could be re-grouped). The guide spells out exactly what each mechanism can promise.

Fine print: Process groups → adopt · Platform support.

Test code that runs processes — without processes

use processkit::Command;
use processkit::testing::{Reply, ScriptedRunner};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // Your code takes any `R: ProcessRunner`; in tests, hand it a script.
    // Rules match on a prefix of the *program name followed by its arguments*
    // (the first element is the program):
    let runner = ScriptedRunner::new()
        .on(["git", "rev-parse"], Reply::ok("abc123\n"))
        .on(["git", "push"], Reply::fail(128, "remote: permission denied"))
        .fallback(Reply::ok(""));

    // my_deploy(&runner).await? — no subprocess, fully deterministic.
    Ok(())
}

RecordingRunner wraps any runner and captures every Invocation for assertions; MockRunner (feature mock) gives mockall expectations; and the record feature's RecordReplayRunner records real runs into a JSON cassette once and replays them hermetically in CI.

Fine print: Testing your code.

Test streaming code — without processes

use processkit::{Command, Outcome, ProcessRunner, Finished};
use processkit::testing::{Reply, ScriptedRunner};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let runner = ScriptedRunner::new()
        .on(["gh", "run", "watch"], Reply::lines(["queued", "in_progress", "completed"])
            .with_line_delay(Duration::from_millis(50))); // paced delivery

    let mut run = runner.start(&Command::new("gh").args(["run", "watch", "123"])).await?;
    run.wait_for_line(|l| l.contains("completed"), Duration::from_secs(5)).await?;
    let Finished { outcome, .. } = run.finish().await?;
    assert_eq!(outcome, Outcome::Exited(0));
    Ok(())
}

A scripted start() feeds the canned lines through the same pump machinery a real child uses, so stdout_lines, the readiness probes, and finish behave identically — and with_line_delay is deterministic under #[tokio::test(start_paused = true)]. Canned output also replays through on_stdout_line/on_stderr_line handlers on the bulk verbs, so progress-reporting paths test hermetically too.

Fine print: Testing → scripted streaming.

Wrap a CLI tool behind a typed API

#![allow(unused)]
fn main() {
use processkit::{cli_client, ProcessRunner, Result};

cli_client!(pub struct Git => "git");

impl<R: ProcessRunner> Git<R> {
    pub async fn current_branch(&self) -> Result<String> {
        // A verb takes the args directly (D7); pass a built `command(..)` only
        // when you need to customize it (per-call timeout, stdin, …).
        self.core.run(["branch", "--show-current"]).await
    }
    pub async fn is_clean(&self) -> Result<bool> {
        self.core.probe(["diff", "--quiet"]).await
    }
}
}

The generated struct carries a runner and per-client defaults (default_timeout, default_env); your methods are just argument lists and parsers — and because the runner is injectable, the whole wrapper is testable with the previous recipe's ScriptedRunner.

Fine print: Testing your code → CliClient.

Running commands

‹ docs index

Command is the entry point of the runner layer: a builder describing what to run and how, plus a family of consuming verbs that decide what you get back. Every one-shot verb spawns the child into a fresh, private kill-on-drop process group, so an early return, panic, or dropped future can never leak a process tree.

• Program, arguments, working directory
• Resolving a locally-installed tool: prefer_local
• Environment
• Standard input
• Output handling
• Timeouts and retries
• Privileges and spawn flags
• Consuming verbs
• Results and errors

Program, arguments, working directory

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let out = Command::new("git")
        .arg("log")                          // one at a time…
        .args(["--oneline", "-n", "10"])     // …or in bulk
        .current_dir("/path/to/repo")        // run there
        .run()
        .await?;
    Ok(())
}

Arguments are passed as an array — there is no shell between you and the child, so there is no quoting, no word-splitting, and no injection surface. (When you actually want a | b | c, use a pipeline, which connects the stages in-process instead of invoking a shell.)

The program name reaches the OS verbatim — two deliberate non-goals (conveniences some libraries layer on, e.g. duct): a bare name is resolved on PATH by the OS, never rewritten to ./name; and current_dir does not re-anchor a relative program path against the new directory — whether Command::new("./tool").current_dir(dir) resolves tool relative to dir is the platform's behavior (Unix: yes; Windows: the parent's directory may win). Pass absolute program paths when combining the two.

For quick one-liners the free functions skip the builder:

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let version = processkit::run("cargo", ["--version"]).await?;       // trimmed stdout, success required
    let result  = processkit::output_string("git", ["status", "-s"]).await?;   // full ProcessResult
    Ok(())
}

Resolving a locally-installed tool: prefer_local

prefer_local adds a directory to check before the system PATH when resolving a bare-name program for this one run — for a project's own node_modules/.bin, a target/debug build, or a vendored toolchain, without hand-rolling a PATH override:

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let out = Command::new("eslint")
        .prefer_local("./node_modules/.bin")
        .arg("src/")
        .output_string()
        .await?;
    Ok(())
}

Resolution order. Repeated calls accumulate, in priority order: the directory from the first call is probed first, then the second, and so on, with the system PATH tried last as the final fallback:

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    Command::new("tool")
        .prefer_local("./vendor/bin")   // checked first
        .prefer_local("./target/debug") // checked second
        .run()                          // then the system PATH
        .await?;
    Ok(())
}

Resolution reuses the exact same PATHEXT-aware lookup as the PATH search (the same internal probe_dir helper — not a separate implementation), so a .exe/.cmd/.bat on Windows is found under a prefer_local directory exactly as it would be on PATH.

Only a bare name is affected. If the program passed to Command::new is a path — absolute, or relative with a separator ("./tool", "../bin/x") — prefer_local has no effect at all: the existing contract that such a program is never looked up on PATH (or here) is unchanged.

Interaction with PATH/inherit_env/env. prefer_local only changes where the parent looks to resolve the program for this one launch. It does not rewrite or extend the PATH the child sees in its own environment — that is governed entirely by env/inherit_env/env_clear, as usual. When the program is found under a prefer_local directory, the child is simply spawned via that resolved absolute path instead of the bare name; a grandchild the program itself spawns does not inherit this reach — only the one program named in this Command benefits.

Interaction with current_dir. A relative prefer_local directory (as in the examples above) is probed against the process's actual current directory, never against whatever is set via current_dir on the same Command. The resolved match is then always turned into an absolute path before being handed to the OS, so it can't later be reinterpreted against the child's working directory once current_dir is set — unlike a relative-path program passed straight to Command::new, which is subject to that footgun (see Program, arguments, working directory above).

Diagnostics. If resolution fails everywhere, Error::NotFound's searched field includes the prefer_local directories too — first, in priority order, ahead of the PATH directories — so the diagnostic never hides that they were checked.

Preflight: resolve a program without running it

Sometimes you want to know whether an external tool is available before you run it — a doctor check at startup, a friendly "is git installed?" error up front — with no side effects. resolve_program locates a command's program and returns its absolute path without spawning anything:

use processkit::Command;

fn main() -> processkit::Result<()> {
    // `which` is the crate-level shortcut for a bare tool.
    let git = processkit::which("git")?;   // Ok(/usr/bin/git) or Err(NotFound)
    println!("git lives at {}", git.display());

    // On a builder it honors that command's own `prefer_local` and env, so it
    // resolves exactly what a real run of that command would launch.
    let eslint = Command::new("eslint")
        .prefer_local("./node_modules/.bin")
        .resolve_program()?;
    println!("eslint lives at {}", eslint.display());
    Ok(())
}

No divergence from a real run. Resolution reuses the crate's own launch-path logic — the same PATH/PATHEXT/execute-bit resolution and prefer_local handling a spawn performs, not a second copy — so a resolve_program hit is exactly the executable a run would launch, and a miss is exactly the Error::NotFound (with the same searched diagnostic and is_not_found() classification) a run would raise. A command that relocates the child's PATH (env/env_remove of PATH, env_clear, inherit_env) is resolved against that effective child PATH, so preflight still matches the spawn.

It is a synchronous, cheap filesystem probe (a few stats) — no async runtime is required, and no process is ever started. Contrast probe(), which runs the tool to read its exit code; resolve_program only locates it.

fn main() {
    match processkit::which("definitely-not-installed") {
        Ok(path) => println!("found: {}", path.display()),
        Err(e) if e.is_not_found() => eprintln!("tool not installed"),
        Err(e) => eprintln!("resolution error: {e}"),
    }
}

For a tool wrapped behind a CliClient, CliClient::resolve_program() does the same for the client's program, honoring its env defaults.

Environment

Four builders compose, applied in a fixed order at spawn:

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    Command::new("worker")
        .env("RUST_LOG", "debug")        // set one variable
        .env_remove("GIT_DIR")           // unset one inherited variable
        .run().await?;

    // Allow-list mode: clear everything, copy only the named parent variables.
    Command::new("sandboxed-tool")
        .inherit_env(["PATH", "HOME", "LANG"])
        .env("MODE", "ci")               // explicit env/env_remove still apply on top
        .run().await?;

    // Scorched earth: the child starts with an empty environment.
    Command::new("hermetic-tool").env_clear().run().await?;
    Ok(())
}

inherit_env is the sandboxing middle ground: it implies env_clear, then copies the listed variables from the parent at each spawn (so a retry sees fresh values), and repeated calls accumulate names. A name the parent doesn't have is skipped, not set to empty.

Standard input

By default stdin is closed at spawn — the child reads EOF immediately and can never hang waiting for input. Everything else is opt-in via stdin(Stdin::…):

use processkit::{Command, Stdin};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let sorted = Command::new("sort")
        .stdin(Stdin::from_iter_lines(["banana", "apple", "cherry"]))
        .run()
        .await?;
    assert_eq!(sorted, "apple\nbanana\ncherry");
    Ok(())
}

The payload is written on a background task (so a large input can't deadlock against the child's output) and the pipe is dropped afterwards to signal EOF. The two one-shot sources are consumed by their first run: a retried or cloned command reusing them fails loud the second time — re-running a consumed from_reader/from_lines source is an Error::Io (InvalidInput) at launch (D10), not a silent empty stdin. Prefer the reusable sources when a command may run more than once.

For conversational, request/response stdin — write a line, read the answer, repeat — use keep_stdin_open() and the streaming API instead: see Streaming & interactive I/O.

Inheriting the parent's stdin: inherit_stdin()

inherit_stdin() hands the child the parent's own standard input — it reads directly from whatever this process's stdin is (a terminal, a file, a pipe) rather than from a crate-managed pipe. It is the stdin counterpart of stdout(StdioMode::Inherit) / stderr(StdioMode::Inherit): the child shares the parent stream instead of the crate mediating it.

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // `git commit` opens $EDITOR on the parent's terminal; the child talks to the
    // real tty directly. stdout/stderr are still captured as usual.
    Command::new("git").arg("commit").inherit_stdin().run().await?;
    Ok(())
}

Reach for it when a child must talk to the real terminal — git commit opening $EDITOR, a tool prompting for a password or a yes/no — or to forward the parent's piped stdin straight through. Until a pseudo-terminal exists (a future direction, not yet provided) this covers the common non-tty-negotiating interactive cases without the crate having to pump bytes. Because the child reads the parent's stdin directly, the crate neither feeds nor captures that input, and take_stdin() returns None (as for a non-keep_stdin_open run). Capturing and streaming the child's output is unaffected.

Why a dedicated verb rather than a Stdin::Inherit source or a mode enum. For stdout/stderr the three StdioMode variants map cleanly onto one setter, but stdin's "piped" case is not modeless — it needs a payload (which source? what bytes?), already expressed by stdin(Stdin::…), and its "null" case is Stdin::empty(). Folding inheritance into that same stdin(Stdin) field would make "inherit and a source" collapse to silent last-write-wins, impossible to flag. A separate inherit_stdin() keeps the two intents in distinct fields so an incompatible pairing is a detectable, rejectable error instead.

Accordingly, inherit_stdin() is mutually exclusive with either way the crate would otherwise drive stdin — a configured stdin(Stdin::…) source (including an explicit Stdin::empty()) or keep_stdin_open()'s interactive pipe. Setting inherit_stdin() together with one of those is a contradiction (feed the child a source and let it read the terminal?), so it is refused at the launch boundary with a typed Error::Io (InvalidInput) — the same failure mode as re-running a consumed one-shot source — rather than silently letting one win. Drop the other stdin knob to resolve it. The refusal is enforced on the same launch seam the hermetic test doubles route through, so a ScriptedRunner rejects the conflict exactly as a live run does.

Output handling

Encodings

Output is decoded line by line, UTF-8 by default (invalid bytes become U+FFFD, never an error). Legacy-encoding tools can override per stream:

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let out = Command::new("legacy-tool")
        .encoding(encoding_rs::SHIFT_JIS)          // both streams…
        // .stdout_encoding(…) / .stderr_encoding(…) // …or each its own
        .output_string()
        .await?;
    Ok(())
}

(processkit::prelude::Encoding re-exports encoding_rs::Encoding, so any of its encodings works — the single-byte and ASCII-compatible multibyte ones (WINDOWS_1252, GBK, SHIFT_JIS, …) and the non-ASCII-compatible ones (UTF_16LE/UTF_16BE): output is fed through one persistent decoder and split on decoded newlines, so a 0x0A byte inside a UTF-16 code unit is not mistaken for a line break. A leading byte-order mark of the chosen encoding is stripped once at the stream start.)

Buffer policies — bounding memory on chatty children

Captured lines are held in memory; a multi-gigabyte log would normally grow the buffer to match. output_buffer bounds retention (the pipe is always fully drained, so the child never blocks):

use processkit::{Command, OutputBufferPolicy, OverflowMode};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let tail = Command::new("verbose-build")
        .output_buffer(OutputBufferPolicy::bounded(1_000)) // keep the newest 1000 lines
        .output_string()
        .await?;

    // …or keep the head instead of the tail:
    let head_policy = OutputBufferPolicy::bounded(1_000).with_overflow(OverflowMode::DropNewest);
    Ok(())
}

DropOldest (the default) keeps a rolling tail; DropNewest freezes the head. bounded(0) retains nothing — useful when a line handler (below) is the real consumer. Under a line cap, dropped or not, every line still feeds the handlers and the line counters.

The line cap alone does not bound memory — one enormous newline-free "line" (base64 -w0) is held whole. Add with_max_bytes to cap the retained bytes too (either ceiling, or both); the byte cap also bounds the pump's in-flight assembly buffer, so a never-terminated flood can't exhaust memory. One consequence: a line whose own length exceeds the byte cap can't be assembled, so it is dropped whole — counted, but not delivered to a per-line handler or stdout_tee (don't set a byte cap if a tee must see arbitrarily long lines):

#![allow(unused)]
fn main() {
use processkit::{Command, OutputBufferPolicy};
let policy = OutputBufferPolicy::unbounded().with_max_bytes(8 << 20); // 8 MiB ring
let strict = OutputBufferPolicy::fail_loud(10_000).with_max_bytes(8 << 20); // error on either
}

fail_loud makes the ceiling error instead of dropping: the run fails with Error::OutputTooLarge once the cumulative output (lines or bytes) crosses the cap — even when a streaming consumer is draining lines as they arrive. It bounds memory, not wall-time, so pair it with timeout against a flooding child.

Even under a drop policy (DropOldest/DropNewest), the checking verbs that hand back stdout as if complete — run, parse, try_parse — refuse silently-truncated output (B12): if the policy dropped lines they fail with Error::OutputTooLarge rather than feed a parser a truncated tail. The lenient capture verbs (output_string / output_bytes) are unaffected — they return the partial result with truncated() set for you to inspect.

Line handlers — tee output as it arrives

on_stdout_line / on_stderr_line run a callback on each decoded line in addition to capture or streaming — logging, progress bars, metrics:

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let result = Command::new("cargo")
        .args(["build", "--release"])
        .on_stderr_line(|line| eprintln!("[build] {line}"))
        .output_string()
        .await?;
    Ok(())
}

The handler runs on the read pump — keep it cheap. The contract is forgiving and precisely specified:

• A panicking handler does not poison the run. The panic is caught, the handler is disabled for the rest of the run (surfaced as a tracing warn when that feature is on), and pumping continues — the final result still carries every line. You can safely re-export this callback seam to your own users without auditing their closures.
• Ordering: invocations are FIFO within a stream; there is no ordering between stdout and stderr handlers (two independent pumps). On the consuming verbs, all handler calls happen-before the awaited future resolves — finalize a progress bar the moment the call returns. (One documented exception: a leaked pipe held open past the child's death is cut off after a bounded teardown grace.)
• Handlers are hermetically testable: ScriptedRunner replays canned output through them — see Testing → scripting replies.

For a ready-made tee to an async sink — a file, socket, or any [tokio::io::AsyncWrite] — reach for stdout_tee / stderr_tee instead of hand-writing a handler. Each decoded line is written to the sink (plus a \n) as it is produced, awaited on the pump so a slow sink applies backpressure (the pump slows, the pipe fills, the child blocks) rather than blocking the runtime; a write error disables the tee with a tracing warn instead of being swallowed. It runs independently of on_stdout_line — set both and both fire per line.

Timeouts and retries

use processkit::{Command, Error};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let out = Command::new("flaky-network-tool")
        .timeout(Duration::from_secs(30))                 // kill the tree at the deadline
        .retry(3, Duration::from_millis(200), |e| {       // up to 3 attempts total
            matches!(e, Error::Timeout { .. })            // …but only retry timeouts
        })
        .run()
        .await?;
    Ok(())
}

• timeout kills the whole process tree at the deadline. On the capturing verbs the expiry is captured (ProcessResult::timed_out), on the success-checking verbs it raises Error::Timeout — the full decision table lives in Timeouts, retries & cancellation.
• retry applies to the success-checking verbs only — run, run_unit, exit_code, probe, checked, parse, and try_parse (seven in all; each runs through the retry loop). The classifier sees the typed error and decides. The non-erroring output_string/output_bytes paths never retry, and neither does first_line (its stream search is single-attempt).

Privileges and spawn flags

Spawn-time controls for sandboxing and service launch:

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // Unix: drop privileges (uid + gid + supplementary groups) and detach.
    Command::new("worker")
        .gid(1000)            // applied before uid (a gid change needs privilege)
        .groups([1000])       // replace the inherited (often root's) supplementary groups
        .uid(1000)            // dropped last
        .setsid()             // new session: survives the controlling terminal
        .run().await?;

    // Windows: no console window flashing up from a GUI app.
    Command::new("helper").create_no_window().run().await?;

    // Hardening: take the direct child down even if THIS process is SIGKILLed
    // (Drop never runs). Windows has this for free; Linux arms PDEATHSIG.
    Command::new("worker").kill_on_parent_death().start().await?;
    Ok(())
}

uid / gid / groups / setsid are POSIX-only — on Windows the run fails with Error::Unsupported rather than silently skipping a privilege drop. A correct drop sets all three of uid/gid/groups: dropping the uid alone leaves the child holding the parent's (often root's) supplementary groups. create_no_window is a harmless no-op outside Windows. kill_on_parent_death is best-effort by design: guaranteed on Windows (regardless of the knob), direct-child-only on Linux, unavailable on macOS/BSD — the graceful-exit guarantee via Drop holds everywhere either way. Containment is preserved in every combination; the platform fine print (the Linux cgroup × uid interaction, setsid × process-group coordination, the pdeathsig thread caveat) is collected in Platform support.

Scheduling: CPU priority and umask

Two more spawn-time knobs, reusing the same seams as the builders above — Unix pre_exec, Windows creation_flags — for background/batch children that shouldn't starve the foreground, and for controlling the permissions of files a child creates:

use processkit::{Command, Priority};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // Run at a lower CPU-scheduling priority — supported on BOTH platforms.
    Command::new("batch-job")
        .priority(Priority::BelowNormal)
        .run().await?;

    // Unix only: files this child creates get 0644/0755 instead of 0666/0777.
    Command::new("worker").umask(0o022).run().await?;
    Ok(())
}

priority maps onto nice/setpriority on Unix and a priority class on Windows (Idle/BelowNormal/Normal/AboveNormal/High); unlike the privilege builders, every variant is supported on both platforms, so this knob never yields Error::Unsupported. One caveat: lowering nice below its inherited value on Unix — raising priority via Priority::AboveNormal/High, or even requesting Priority::Normal under a positively-niced parent (e.g. a niced CI/batch launcher) — needs CAP_SYS_NICE/root; without it the OS rejects the change and the spawn fails loud (Error::Spawn), never silently downgrading to a lower priority.

umask is Unix-only — like setsid/groups, requesting it on Windows fails with Error::Unsupported rather than being silently ignored.

Interactive auth / TTY. processkit wires pipes, not a pseudo-terminal, so a tool that demands a tty — an ssh/sudo password prompt, some credential helpers — won't get one (PTY support is not implemented; the trade-off is recorded in decisions/permissions-privileges-pty-network.md). Drive such tools non-interactively instead: key-based auth, ssh -o BatchMode=yes, GIT_SSH_COMMAND / GIT_TERMINAL_PROMPT=0, or feed a known answer over interactive stdin. Conversational tools that read stdin without needing a tty already work today via keep_stdin_open + stdout_lines.

Consuming verbs

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // probe(): the exit code as a boolean.
    let clean = Command::new("git").args(["diff", "--quiet"]).probe().await?;

    // first_line(): stop as soon as the interesting line appears.
    let first_match = Command::new("git")
        .args(["log", "--oneline"])
        .first_line(|l| l.contains("fix:"))
        .await?;
    Ok(())
}

first_line returns Ok(None) when stdout closes without a match, and kills the (private-group) child once it has its answer — you never wait out a long log for one line. A cancel_on token that fires while the search is still running surfaces as Error::Cancelled, so a readiness probe with a shutdown token can't misread token-driven teardown as "the line never appeared" — while a run that genuinely ends with no match still reports Ok(None), even if the token happens to fire an instant later.

Results and errors

The capturing verbs hand back a ProcessResult:

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let result = Command::new("git").args(["merge", "feature"]).output_string().await?;

    result.code();         // Option<i32> — None = killed (timeout/signal), no code
    result.signal();       // Option<i32> — the signal number (Unix), else None
    result.is_success();   // code in ok_codes (default {0})
    result.timed_out();    // the run's own deadline expired
    result.outcome();      // the explicit three-way enum behind the accessors above
    result.stdout();       // &str (or &[u8] from output_bytes)
    result.stderr();       // &str
    result.combined();     // stdout + stderr concatenated
    result.diagnostic();   // stderr if non-empty, else stdout — the human-facing line
                           // (git/jj put "CONFLICT …" on stdout!)

    // Opt into erroring whenever you're ready:
    let ok = result.ensure_success()?; // Exit / Timeout / Signalled (signal-kill) as typed errors
    Ok(())
}

When the three-way distinction matters, match on Outcome instead of mentally decoding the code()/timed_out() pair:

use processkit::Outcome;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let result = processkit::Command::new("git").args(["merge", "feature"]).output_string().await?;
    match result.outcome() {
        Outcome::Exited(0) => println!("clean"),
        Outcome::Exited(code) => println!("failed with {code}"),
        Outcome::Signalled(signal) => println!("killed by signal {signal:?}"),
        Outcome::TimedOut => println!("hit its deadline"),
        _ => {} // non_exhaustive: future dispositions
    }
    Ok(())
}

For a single query you usually don't need the match (and its #[non_exhaustive] wildcard): Outcome carries the same code() / signal() / timed_out() accessors as ProcessResult, so a bare Outcome (from RunningProcess::wait or Finished::outcome) answers directly — outcome.code(), outcome.signal(), outcome.timed_out(). There is no Outcome::is_success (success is ok_codes-aware — use ProcessResult::is_success).

The error enum is structured and #[non_exhaustive]:

Error::diagnostic() returns the most useful human-facing line out of a failure that captured output — Exit, and (D12) Timeout / Signalled (the partial streams of a hung-then-killed or crashed tool). Each of those variants' one-line Display also appends a bounded excerpt of that diagnostic (the last non-empty line, capped at 200 bytes), so a bare eprintln!("{e}") reads `git` exited with code 2: fatal: boom — actionable in a log line without dumping multi-KiB streams into it.

Next: Streaming & interactive I/O · Timeouts, retries & cancellation · Process groups

Process groups

‹ docs index

A ProcessGroup ties the lifetime of a whole child-process tree to a Rust value: every process spawned into the group — and everything those processes spawn — is killed when the group is dropped. An exiting, panicking, or ?-returning owner never leaks subprocesses; the kernel object enforcing this (Job Object / cgroup / POSIX process group) catches even grandchildren you never knew about. (Killing grandchildren is the problem duct.py's gotchas list files under "currently unsolved" for pipe-based designs — kernel containment is the solution, and the reason this crate exists.)

• Creating a group
• Putting processes in
• Tearing down: drop, terminate, shutdown
• Signalling the whole tree
• Suspending and resuming
• Listing members
• Resource limits
• Stats and sampling

Creating a group

use processkit::{ProcessGroup, ProcessGroupOptions};
use std::time::Duration;

fn main() -> processkit::Result<()> {
    // Defaults: 2s graceful-shutdown grace, escalate to SIGKILL.
    let group = ProcessGroup::new()?;

    // Tuned:
    let group = ProcessGroup::with_options(
        ProcessGroupOptions::default()
            .shutdown_timeout(Duration::from_secs(10))
            .escalate_to_kill(true),
    )?;

    // Which kernel mechanism is actually containing the tree?
    println!("{:?}", group.mechanism()); // JobObject | CgroupV2 | ProcessGroup
    Ok(())
}

mechanism() reports what you actually got: CgroupV2 quietly falls back to ProcessGroup on Linux hosts without cgroup delegation (see Platform support).

You rarely create a group explicitly for one-shot runs: every Command::run()-style call makes a private group automatically. Reach for an explicit group when several children should share one fate, or when you need the group verbs below.

Putting processes in

Three doors, in order of preference:

use processkit::{Command, ProcessGroup};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let group = ProcessGroup::new()?;

    // 1. start(): the full Command experience (capture, streaming, timeouts) in a
    //    SHARED group. The handle does not own the group — dropping the handle
    //    kills that child, dropping the group kills everyone.
    let server = group.start(&Command::new("dev-server")).await?;

    // 2. spawn(): the raw escape hatch for a tokio::process::Command you already
    //    have. You get the bare Child back; pipes and reaping are your problem.
    //    spawn() takes the command BY VALUE (reuse would stack pre-exec hooks).
    let raw = tokio::process::Command::new("background-helper");
    let child = group.spawn(raw)?;

    // 3. adopt(): contain a child that was spawned OUTSIDE the group.
    let external = tokio::process::Command::new("legacy-launcher").spawn()?;
    group.adopt(&external)?;
    let _ = (server, child);
    Ok(())
}

adopt moves only the named process: descendants it already has keep their old containment (future forks are captured — on Windows/cgroup). A few sharp edges worth knowing:

• A child that already exited but has not been reaped (no wait() yet — a zombie whose pid/handle is still valid) is a successful no-op: there is nothing left to contain, so adopt returns Ok on the containment backends.
• A child that already exited and was reaped (wait()ed) has no pid left — adopt returns an error rather than silently tracking nothing.
• On the POSIX process-group mechanism, a child that has already exec'd can't be re-grouped (POSIX forbids it), so it is tracked individually: the child itself is signalled/killed with the group, but its future forks are not. The caller keeps the Child handle and is responsible for reaping.

Tearing down: drop, terminate, shutdown

use processkit::{Command, ProcessGroup, ProcessGroupOptions};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::with_options(
        ProcessGroupOptions::default()
            .shutdown_timeout(Duration::from_secs(5))
            .escalate_to_kill(true),
    )?;
    let _service = group.start(&Command::new("my-service")).await?;

    // SIGTERM, give it 5s to flush and exit, SIGKILL stragglers:
    group.shutdown().await?;
    Ok(())
}

A child that handles SIGTERM ends the grace early — shutdown returns as soon as the tree is empty, not after the full timeout. One subtlety: the liveness probe sees an exited-but-unreaped child (a zombie) as alive on the process-group backends, so keep wait()ing your handles concurrently if you want the early return. Drop can't await, which is why the graceful tier lives in this async method — dropping without calling it performs only the hard kill.

Signalling the whole tree

signal/suspend/resume/members/adopt — this section and the two below — require the default-on process-control feature. The teardown verbs above are core and always present.

use processkit::{Command, ProcessGroup, Signal};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::new()?;
    let _server = group.start(&Command::new("my-server")).await?;

    group.signal(Signal::Hup)?;        // "reload your configuration"
    group.signal(Signal::Usr1)?;       // whatever the tool defines
    group.signal(Signal::Other(34))?;  // raw signal number escape hatch
    Ok(())
}

Signal::Kill always takes the same atomic whole-tree kill path as kill_all (cgroup.kill / killpg / job terminate), so it cannot miss a process forked mid-broadcast. Other signals are a per-member broadcast — best-effort against a tree that is forking at that exact moment. An empty group accepts any deliverable signal trivially. On the cgroup mechanism a real per-member delivery failure (e.g. EPERM from a member that changed uid, or a seccomp/container restriction) is surfaced as an Err rather than swallowed — an ESRCH race (the member already exited) is still success; the pgroup (macOS/BSD, Linux-without-cgroup) backend remains purely best-effort.

Suspending and resuming

Freeze a tree (to snapshot it, to starve a runaway while you investigate, to pause background work), then thaw it:

use processkit::{Command, ProcessGroup};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::new()?;
    let _cruncher = group.start(&Command::new("cpu-hog")).await?;

    group.suspend()?;   // the whole tree stops consuming CPU
    // … inspect, snapshot, wait for the user …
    group.resume()?;
    Ok(())
}

Per-platform machinery — and its visible differences:

Two caveats that bite in practice:

• Spawning into a suspended group diverges. Under the cgroup mechanism a child spawned or adopted while the group is frozen starts frozen — and start() may never return until resume (the forked child joins the cgroup before exec, so it can freeze before completing the spawn handshake). Windows and the pgroup backends freeze only members present at the call. Rule of thumb: resume before starting new work.
• A suspended tree can still be hard-killed (drop / kill_all / Signal::Kill all act on frozen processes), but a graceful shutdown starts with a SIGTERM the frozen tree can't act on — it would wait out the whole grace. Resume first for a clean shutdown.

Listing members

use processkit::{Command, ProcessGroup};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::new()?;
    let _a = group.start(&Command::new("worker-a")).await?;
    let _b = group.start(&Command::new("worker-b")).await?;

    let pids: Vec<u32> = group.members()?;
    println!("live members: {pids:?}");
    Ok(())
}

What "members" means depends on the mechanism: Windows and Linux-cgroup list the whole tree (every descendant pid); the POSIX process-group backends list the tracked group leaders (one pid per started/adopted child) — their descendants are contained but not enumerated. An exited child still counts until it is reaped. The snapshot is point-in-time: a tree that is forking races it.

To wait on members rather than list them, race the handles with wait_any.

Resource limits

Requires the limits feature. Caps are a property of the group, set once at creation and enforced by the same kernel object that contains the tree:

use processkit::{Command, ProcessGroup, ProcessGroupOptions};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::with_options(
        ProcessGroupOptions::default()
            .max_memory(512 * 1024 * 1024) // bytes, whole tree
            .max_processes(64)             // fork-bomb ceiling
            .cpu_quota(0.5),               // half of one core
    )?;
    let _sandboxed = group.start(&Command::new("untrusted-tool")).await?;
    Ok(())
}

cpu_quota is a fraction of a single core (2.0 = two cores). Limits need a real container; when a requested cap can't be enforced — no Job Object/cgroup, or a Linux cgroup whose controllers can't be enabled — with_options returns Error::ResourceLimit { kind, reason, detail } instead of handing back a silently-unbounded group: kind names the limit (max_memory/max_processes/cpu_quota), reason says whether the value was simply invalid, the platform has no whole-tree mechanism at all (Unsupported), or a mechanism exists but rejected this request (Unenforceable) — branch on these instead of parsing detail. On Linux this needs the process to run at the real cgroup-v2 root: the crate enables the controllers in this process's own cgroup, which cgroup v2's "no internal processes" rule allows only for the real hierarchy root — not a cgroup-namespace root (so an ordinary container fails too), not under systemd — and the crate doesn't migrate your process. See the limits prerequisites in Platform support. The uid()-drop interaction lives under its Caveats.

Stats and sampling

Requires the opt-in stats feature (features = ["stats"], or limits).

use processkit::prelude::StreamExt;
use processkit::{Command, ProcessGroup};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::new()?;
    let _worker = group.start(&Command::new("worker")).await?;

    // Point-in-time:
    let snap = group.stats()?;
    println!(
        "procs={} cpu={:?} peak_rss={:?}",
        snap.active_process_count, snap.total_cpu_time, snap.peak_memory_bytes,
    );

    // …or a series: first sample immediate, then every 250ms; missed ticks are
    // skipped; the stream ends when the group can no longer report.
    let mut samples = group.sample_stats(Duration::from_millis(250));
    while let Some(s) = samples.next().await {
        println!("rss now: {:?}", s.peak_memory_bytes);
    }
    Ok(())
}

CPU time and peak memory are available where the kernel accounts for the whole tree (Windows, Linux cgroup); the process-group backends report the member count only — the Option fields stay None. The sampler borrows the group, so it can neither outlive it nor keep it (and the kill-on-drop guarantee) alive. For a single run's end-to-end summary, see profile.

Next: Streaming & interactive I/O · Platform support · Supervision

Streaming & interactive I/O

‹ docs index

The one-shot verbs in Running commands buffer the whole output. For long-running or conversational children, Command::start() returns a live RunningProcess you drive yourself: stream stdout as it arrives, write stdin incrementally, probe for readiness, race several children, or profile a run.

• Lifecycle
• Streaming stdout
• Interactive stdin
• Readiness probes
• Racing children with wait_any
• Per-run telemetry

Lifecycle

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let mut run = Command::new("dev-server").start().await?;

    run.pid();        // Option<u32> — None once the child is reaped
    run.elapsed();    // time since spawn

    // Consume the handle exactly one way:
    //   output_string() / output_bytes()  → capture everything (same as the one-shot verbs)
    //   wait()                            → just the Outcome; output is discarded
    //   finish()                 → after streaming stdout (below)
    //   profile(every)                    → resource samples; output discarded, like wait() (stats feature)
    let outcome = run.wait().await?;   // Outcome: Exited(code) / Signalled(sig) / TimedOut
    Ok(())
}

start() puts the child in a private group the handle owns: dropping the RunningProcess kills the whole tree, exactly like dropping a one-shot run's future. The shared-group variant — group.start(&cmd) — gives the same handle but the group controls the tree's fate (see Process groups).

There is also an explicit run.start_kill() for "stop it now, I'll wait() for the code myself".

Streaming stdout

stdout_lines() yields decoded lines as the child produces them — no waiting for exit, no full-output buffering. StreamExt (in processkit::prelude, re-exported from tokio-stream) provides .next():

use processkit::prelude::StreamExt;
use processkit::{Command, Finished, Outcome};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let mut run = Command::new("cargo")
        .args(["build", "--release"])
        .start()
        .await?;

    let mut lines = run.stdout_lines()?;
    while let Some(line) = lines.next().await {
        println!("build: {line}");
    }

    // The stream ended (stdout closed). Collect the outcome and stderr —
    // stderr was drained in the background the whole time, so a noisy child
    // could never block on a full pipe.
    let Finished { outcome, stderr, .. } = run.finish().await?;
    if outcome != Outcome::Exited(0) {
        eprintln!("build failed ({outcome:?}):\n{stderr}");
    }
    Ok(())
}

Things to know:

• Call stdout_lines() once. It is fallible: a second stdout_lines / output_events call (stdout is consumed once), or a non-piped stdout (StdioMode::Inherit/Null), returns Err rather than a silently-empty stream.
• The command's timeout bounds the stream: at the deadline the tree (own-group handle) or the direct child (shared-group handle) is killed, the pipes close, and the stream ends — a streamed run can't hang past its deadline. A cancel_on token ends it the same way; the following finish then reports Error::Cancelled. Details in Timeouts & cancellation.
• Line counters tick live: run.stdout_line_count() / stderr_line_count() are cheap progress gauges even while you stream.
• The buffer policy and line handlers apply to streamed runs too — a handler sees each line on the pump, in addition to your loop.
• The whole streaming surface is hermetically testable: a ScriptedRunner's start() returns a handle whose canned lines flow through the same pump machinery — stdout_lines, the readiness probes, and finish behave identically with no subprocess. See Testing → scripted streaming.

Carriage-return progress output

Tools like curl, pip, and apt redraw a progress bar in place with a carriage return (\rProgress: 50%\rProgress: 100%) and emit no \n until the very end. By default the pump splits on \n only, so that whole sequence is a single, ever-growing line: nothing streams live, and under a byte cap the one over-cap line is dropped whole.

Set line_terminator(LineTerminator::CarriageReturn) to treat a bare \r as a line terminator too. Each carriage-return frame then arrives as its own line — live, one at a time:

use processkit::prelude::StreamExt;
use processkit::{Command, LineTerminator};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let mut run = Command::new("pip")
        .args(["install", "big-package"])
        // Frame the stream you actually consume: `stdout_lines()` surfaces
        // stdout only (stderr is drained in the background and discarded), so
        // put the CR framing on stdout. If a tool draws its bar on stderr,
        // set `line_terminator(..)` for both streams and read `output_events()`.
        .stdout_line_terminator(LineTerminator::CarriageReturn)
        .start()
        .await?;

    let mut lines = run.stdout_lines()?;
    while let Some(frame) = lines.next().await {
        // Overwrite your own display with the latest frame.
        print!("\r{frame}");
    }
    run.finish().await?;
    Ok(())
}

The chosen framing is one shared definition of a line for every sink: the streaming verbs, the on_stdout_line/on_stderr_line handlers, a stdout_tee/stderr_tee (which writes each frame followed by \n), and output_string all see the same per-frame lines. A \r\n pair stays a single terminator (no empty line between them), so ordinary CRLF text reads identically to the default; only a \r not followed by a \n splits a frame. The OutputBufferPolicy byte cap now bounds an individual runaway frame — a frame whose content exceeds the cap is skipped as it streams (never assembled whole) — rather than dropping the whole progress stream. Use stdout_line_terminator / stderr_line_terminator for one stream, or line_terminator for both.

Interactive stdin

Conversational tools — write a request, read the response, repeat. Keep stdin open with keep_stdin_open(), take the writer with take_stdin():

use processkit::prelude::StreamExt;
use processkit::{Command, Finished, Outcome};

// `ProcessStdin`'s writer methods return `std::io::Result`; `Box<dyn Error>`
// mixes them with the crate's `Result` (or `.map_err(processkit::Error::Io)?`).
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // `bc` evaluates each stdin line and prints the result.
    let mut run = Command::new("bc").keep_stdin_open().start().await?;
    let mut stdin = run.take_stdin().expect("stdin was kept open");
    let mut answers = run.stdout_lines()?;

    stdin.write_line("2 + 2").await?;             // writes "2 + 2\n", flushed
    println!("= {}", answers.next().await.unwrap());

    stdin.write_line("6 * 7").await?;
    println!("= {}", answers.next().await.unwrap());

    stdin.finish().await?;                        // send EOF — bc exits
    let Finished { outcome, .. } = run.finish().await?;
    assert_eq!(outcome, Outcome::Exited(0));
    Ok(())
}

ProcessStdin offers write(&[u8]), write_line(&str) (newline + flush), flush(), and finish() (EOF). Dropping the writer — or the whole RunningProcess — closes stdin too; finish() just makes the EOF explicit and awaitable.

Avoid the full-duplex deadlock. A child's stdout pipe has a finite OS buffer; once it fills, the child blocks writing stdout until something reads it. If you push a large interactive stdin while nothing drains the child's stdout, the child stops reading stdin (blocked on stdout), your write parks waiting for stdin buffer space, and neither side progresses. The bc example above is safe because it interleaves one write with one read; when you both feed a sizable stdin and the child produces output, drain stdout_lines from one task while writing stdin from another. (The non-interactive Stdin::from_* sources are safe — the crate writes them on a background task that runs concurrently with the output pumps.)

For one-directional streamed input (a channel, a file tail) you don't need interactivity — give the command Stdin::from_lines(stream) / Stdin::from_reader(reader) and let the background writer feed it; see the stdin source table.

Readiness probes

"Start a server, then use it" needs ready, not merely started. Three probes replace the arbitrary sleep, each bounded by its own deadline:

async fn health_check() -> bool { true }
use processkit::Command;
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let mut run = Command::new("my-server").start().await?;

    // 1. A line on stdout (returns the matching line):
    let banner = run
        .wait_for_line(|l| l.contains("listening on"), Duration::from_secs(10))
        .await?;

    // 2. A TCP port accepting connections:
    run.wait_for_port("127.0.0.1:8080".parse().unwrap(), Duration::from_secs(10))
        .await?;

    // 3. Any async predicate (an HTTP /health endpoint, a file appearing, …):
    run.wait_for(|| async { health_check().await }, Duration::from_secs(10))
        .await?;

    // ready — use the server…
    let _ = banner;
    Ok(())
}

Probe semantics, deliberately uniform:

• A probe that can't pass within its deadline fails with Error::NotReady — distinct from Error::Timeout, which is the run's own deadline.
• A probe also fails fast once readiness can no longer happen: the child exits, or (for wait_for_line) its stdout closes — no waiting out a 30s deadline on a dead server.
• A failed probe never kills the child. You decide: retry, log and continue, or tear down.
• All three probes background-drain stdout/stderr while they poll, so a child with a large startup burst can't stall in write() on a full OS pipe buffer. wait_for_line consumes stdout up to (and including) the match — continue with finish. wait_for_port / wait_for drain the same way but never hand any of it back mid-probe; wait / output_string afterward still see the full captured output, but output_bytes or a fresh stdout_lines / output_events call do not compose with any of the three probes (same as calling wait_for_line first).

Racing children with wait_any

The free function wait_any races several running processes and reports whichever exits first — the natural primitive for "restart whatever died" or "first answer wins":

use processkit::{Command, ProcessGroup, wait_any};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::new()?;
    let mut a = group.start(&Command::new("replica-a")).await?;
    let mut b = group.start(&Command::new("replica-b")).await?;

    let (index, outcome) = wait_any(&mut [&mut a, &mut b]).await?;
    println!("contender #{index} exited first with {outcome:?}");

    // Only borrows: the loser is still usable.
    let survivor = if index == 0 { &mut b } else { &mut a };
    Ok(())
}

wait_any takes &mut borrows, applies no timeout of its own (wrap it in tokio::time::timeout to bound the race), and does no output pumping — drain chatty children first or give them bounded buffer policies.

Per-run telemetry

With the opt-in stats feature, a running child reports its own resource usage, and profile() turns a whole run into a summary:

use processkit::Command;
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let run = Command::new("crunch").start().await?;
    run.cpu_time();          // Option<Duration> — user+kernel so far
    run.peak_memory_bytes(); // Option<u64>

    // …or capture + sample on an interval until exit:
    let profile = Command::new("crunch")
        .start().await?
        .profile(Duration::from_millis(100))
        .await?;

    println!(
        "outcome={:?} wall={:?} cpu={:?} peak_rss={:?} avg_cpu_cores={:?} ({} samples)",
        profile.outcome,            // Exited(code) / Signalled(sig) / TimedOut
        profile.duration,
        profile.cpu_time,
        profile.peak_memory_bytes,
        profile.avg_cpu_cores(),    // cpu / wall — e.g. Some(1.7) ≈ 1.7 cores busy
        profile.samples,
    );
    Ok(())
}

These read the child process itself (not a whole tree — that's ProcessGroup::stats), and availability follows the platform: full CPU/memory on Windows and Linux, None where the kernel doesn't account per-process cheaply — see Platform support.

Next: Pipelines · Timeouts, retries & cancellation · Supervision

Pipelines

‹ docs index

a | b | c without a shell. Each stage's stdout feeds the next stage's stdin through an in-process relay (a tokio::io::copy task per boundary) — there is no shell string anywhere, so no quoting rules, no word splitting, no injection surface. Each stage spawns into its own kill-on-drop process group sub-group, so a per-stage Command::timeout tears down that stage's whole subtree (grandchildren of a forking sh -c … included); a chain-wide teardown fans the kill across every stage's sub-group, so the chain still lives and dies as a unit. (The relay is an implementation detail, not a kernel splice: a producer whose consumer exits early stops on a broken pipe when the relay's next write fails, rather than instantly via SIGPIPE.)

• Building and running
• Semantics: pipefail and the ends
• Unchecked stages
• Timeouts
• Re-running a pipeline

Building and running

Command::pipe(next) starts a Pipeline; chain more stages with Pipeline::pipe; drive it with output_string() or run():

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // git log --format=%an | sort | uniq -c
    let authors = Command::new("git").args(["log", "--format=%an"])
        .pipe(Command::new("sort"))
        .pipe(Command::new("uniq").arg("-c"))
        .run()                         // require every stage to succeed
        .await?;
    println!("{authors}");
    Ok(())
}

The verbs mirror Command's, each operating on the pipefail outcome:

Err from output_string itself means a stage couldn't be started or driven at all (spawn failure, broken plumbing) — never a mere non-zero exit.

The streaming first_line probe is deliberately not a pipeline verb: a chain consumes its last stage in full to fold the pipefail outcome. To capture the first matching line of a finished chain, add a | head -n1 (Unix) / grep -m1 / findstr stage and capture. This does not cover a streaming readiness probe over a chain that must keep running (e.g. wait for a banner line, then leave the chain alive) — | head would tear it down; use a single Command with first_line for that.

The | operator is sugar for the same thing — a | b | c ≡ a.pipe(b).pipe(c). Parenthesize the chain before a terminal verb, since method calls bind tighter than |:

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let authors = (Command::new("git").args(["log", "--format=%an"])
        | Command::new("sort")
        | Command::new("uniq").arg("-c"))
        .run()
        .await?;
    Ok(())
}

Semantics: pipefail and the ends

The outcome is pipefail, like set -o pipefail in a shell:

• stdout is always the last stage's output — that's what the chain produced.
• code, stderr, and the reported program come from the culprit stage: the leftmost stage that didn't exit cleanly (non-zero, signal-killed, or timed out), but preferring a real failure over a downstream SIGPIPE victim — a stage killed only because a later stage closed the pipe early. If every failure is such a broken-pipe victim, the leftmost one wins; when every stage succeeded, the last stage speaks.

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let result = Command::new("cat").arg("data.txt")
        .pipe(Command::new("grep").arg("ERROR"))      // suppose grep exits 2 (bad pattern)
        .pipe(Command::new("wc").arg("-l"))
        .output_string()
        .await?;

    // Diagnostics point at grep — the first unclean stage — while stdout is
    // whatever wc managed to print:
    assert_eq!(result.code(), Some(2));
    println!("blamed: {}", result.ensure_success().unwrap_err()); // names `grep`
    Ok(())
}

Failure tears the chain down proactively. The moment a stage ends with a checked failure (a non-zero exit outside its ok_codes, a signal kill, or its own per-stage timeout), every stage's sub-group is torn down at once — the failure does not wait to trickle out through closing pipes. This matters for a quiet sibling that would otherwise hang: an upstream producer that never writes never dies of a broken pipe, so under a purely passive teardown a downstream failure could be held open indefinitely by that silent producer. Now the failure surfaces immediately, and the killed siblings are treated as victims (like a downstream SIGPIPE death) — the stage that actually failed keeps the blame. The one death that does not trigger this is an unchecked_in_pipe() stage's: its unclean exit is forgiven, so it leaves the rest of the chain running. (A stuck stage that never fails — a healthy producer that simply never finishes — is still bounded only by Pipeline::timeout or cancellation.)

The ends of the chain behave like a single Command:

• The first stage's configured stdin source is honored — feed the whole pipeline from a string, file, or stream.
• Inner stages read from the pipe, full stop: any stdin source or keep_stdin_open configured on them is overridden.
• Inner stages' stderr is captured per-stage for pipefail diagnostics; only the last stage's stdout reaches you.

use processkit::{Command, Stdin};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let unique_count = Command::new("sort")
        .stdin(Stdin::from_iter_lines(["b", "a", "b", "c"]))
        .pipe(Command::new("uniq"))
        .pipe(Command::new("wc").arg("-l"))
        .run()
        .await?;
    assert_eq!(unique_count.trim(), "3");
    Ok(())
}

Unchecked stages

Strict pipefail has one classic false positive: a consumer that legitimately stops reading early. In producer | head -1 the consumer exits 0 after one line and closes the pipe; the producer then stops on a broken pipe — its next write fails once the relay's downstream is gone (a broken-pipe write error, or SIGPIPE where the OS delivers it) — a perfectly normal death that strict pipefail would blame the chain for. Mark that stage unchecked_in_pipe():

use processkit::Command;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // seq 1 1000000 | head -1 — the producer's broken-pipe death is expected.
    let first = (Command::new("seq").args(["1", "1000000"]).unchecked_in_pipe()
        | Command::new("head").args(["-n", "1"]))
        .run()
        .await?;
    assert_eq!(first.trim(), "1");
    Ok(())
}

The rules (a design borrowed from duct's unchecked() — the idea, not the code):

• An unchecked stage's unclean exit — a non-zero code, a broken-pipe write failure (or SIGPIPE where the OS delivers it) from a consumer that closed early, or its own per-stage timeout kill — is skipped when the chain decides what to report.
• A checked failure always trumps an unchecked one, regardless of position: unchecked never shields another stage's real failure.
• A chain whose only failures are unchecked reports success with the last stage's stdout and its real exit code preserved (not a fabricated 0 — the accepted-code set is widened to include it). The carve-out is for an exit status only: a last stage killed by a signal or its own timeout is not a status to forgive and still surfaces as the failure.
• unchecked forgives exit status only — never a whole-chain Pipeline::timeout, and it has no effect on a Command run outside a pipeline (a single run's status is already plain data in its ProcessResult).

Timeouts

Two scopes, deliberately distinct:

use processkit::Command;
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let out = Command::new("producer")
        .timeout(Duration::from_secs(10))      // per-STAGE: kills just `producer`
        .pipe(Command::new("consumer"))
        .timeout(Duration::from_secs(30))      // whole-CHAIN: Pipeline::timeout
        .output_string()
        .await?;
    Ok(())
}

• Pipeline::timeout bounds the whole chain: at the deadline every stage's sub-group is torn down and the result reports timed_out (no partial stdout — unlike a single command's captured timeout).
• A per-stage Command::timeout kills that stage's whole subtree — its own sub-group, grandchildren of a forking sh -c … included, not just its direct child. Every stage is evaluated by the same pipefail rule (D14): a stage that hit its own deadline — inner or last — surfaces on run() as that stage's Error::Timeout, reporting that stage's own deadline (not the chain's, and never 0ns).

Cancellation has two forms. Pipeline::cancel_on(token) is the chain-level control: the token gap-fills into every stage that doesn't already carry its own Command::cancel_on (an explicit per-stage token is left intact), so firing it tears the whole chain down and the run resolves to Error::Cancelled. (A cancel_on token on an individual stage Command also cancels that stage and errors the pipeline, but the pipeline-level builder is the clearer authority.) See Timeouts & cancellation.

Re-running a pipeline

A Pipeline is Clone and re-runnable — stages are re-cloned per run. The one caveat is inherited from Command: a one-shot stdin source on the first stage (Stdin::from_reader / from_lines) is consumed by the first run; re-running then fails loud (an Error::Io at launch, D10) rather than silently feeding empty stdin. Use the reusable sources (from_string / from_bytes / from_iter_lines / from_file) when a chain runs more than once.

Next: Timeouts, retries & cancellation · Running commands · Process groups

Timeouts, retries & cancellation

‹ docs index

Three ways a run ends early, with three different philosophies:

• a timeout is data — the deadline was part of the run's contract, so its expiry is captured in the result (and only the success-checking verbs turn it into an error);

• a retry is a policy — the success-checking verbs replay the run while your classifier says the failure is transient;

• a cancellation is an abandonment — the caller changed its mind, so every path reports an error; there is no result worth inspecting.

• Timeouts

• Retries

• Cancellation

• Precedence and interactions

Timeouts

Command::timeout(d) kills the whole process tree at the deadline — not just the direct child, so a wrapper script's grandchildren die too.

use processkit::Command;
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // Captured: inspect the flag yourself.
    let result = Command::new("slow-tool")
        .timeout(Duration::from_secs(5))
        .output_string()
        .await?;
    if result.timed_out() {
        println!("partial output before the kill: {}", result.stdout());
    }

    // Raised: the checking verbs convert the flag into a typed error.
    let err = Command::new("slow-tool")
        .timeout(Duration::from_secs(5))
        .run()
        .await
        .unwrap_err();
    assert!(matches!(err, processkit::Error::Timeout { .. }));
    Ok(())
}

Where each verb lands:

Two distinct deadline families to keep apart:

• Command::timeout — the run's own contract, this section.
• The readiness probes' within parameter — gives Error::NotReady and never kills the child.

Graceful timeout

By default the deadline hard-kills at once. Add timeout_grace(d) to give the tree a chance to clean up: at the deadline it is sent SIGTERM (or the signal chosen with timeout_signal, which needs the process-control feature), allowed up to the grace window to exit, then SIGKILLed — the same SIGTERM → wait → SIGKILL tier as ProcessGroup::shutdown. A signal-handling child that exits ends the grace early.

use processkit::Command;
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let result = Command::new("slow-tool")
        .timeout(Duration::from_secs(30))
        .timeout_grace(Duration::from_secs(5)) // SIGTERM, wait up to 5s, then SIGKILL
        .output_string()
        .await?;
    Ok(())
}

timed_out() is true regardless of whether the child exited on the signal or was SIGKILLed after the grace — the deadline is what fired. Windows has no signal tier: timeout_grace is accepted but the deadline kills the job atomically.

The explicit RunningProcess::shutdown(grace) verb (stop a started handle on demand) composes with a Command::timeout: its own SIGTERM → grace → SIGKILL is the single teardown (it does not also fire the run's timeout teardown), and if the deadline has already elapsed when you call shutdown, the outcome is reported as Outcome::TimedOut — the grace you pass governs the teardown timing.

Retries

retry(max_attempts, backoff, classifier) replays a failed run — up to max_attempts total attempts, sleeping backoff between tries, retrying only while the classifier accepts the error:

use processkit::{Command, Error};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let out = Command::new("curl")
        .args(["-fsS", "https://example.com/api"])
        .timeout(Duration::from_secs(10))
        .retry(3, Duration::from_millis(250), |e| {
            // transient: network timeouts and curl's "couldn't connect" (7)
            matches!(e, Error::Timeout { .. })
                || matches!(e, Error::Exit { code: 7, .. })
        })
        .run()
        .await?;
    Ok(())
}

Ground rules:

• Retries apply to the success-checking paths only (run, exit_code, probe, ProcessRunnerExt::checked — and everything built on them, e.g. CliClient). The non-erroring output_string capture never retries: it didn't fail.
• The classifier sees the typed error — match on variants, codes, even the captured stderr.
• Each attempt re-runs the same Command — so a command whose stdin is a one-shot source (table), consumed by the first run, is not retried at all: the first attempt's error is returned as-is, since a second attempt could only replay empty stdin. Use a reusable stdin source if a stdin-bearing command must retry. (A one-shot source re-run outside the retry loop — a Supervisor incarnation, a pipeline re-run — instead fails loud with an Error::Io (InvalidInput) at launch.)
• A Cancelled error is never retried, classifier or not — the token stays cancelled forever, so another attempt could only fail the same way.

For "keep it alive" (restart a service whenever it exits) rather than "replay this one operation", use a Supervisor — same backoff shape, different loop condition.

Cancellation

Hand any command a CancellationToken (re-exported at the crate root); cancelling the token kills the run's tree and makes every consuming path report Error::Cancelled:

use processkit::{CancellationToken, Command};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let shutdown = CancellationToken::new();

    // Wire the same parent token into many jobs via child tokens:
    let job = tokio::spawn({
        let token = shutdown.child_token();
        async move {
            Command::new("long-export").cancel_on(token).run().await
        }
    });

    // Ctrl-C handler, sibling failure, UI button, …
    shutdown.cancel();

    assert!(matches!(
        job.await.unwrap(),
        Err(processkit::Error::Cancelled { .. })
    ));
    Ok(())
}

The contract, path by path:

Client-level default

A typed wrapper built on CliClient usually constructs and consumes its Commands internally — there is no place to chain a per-call cancel_on. Set the token once on the client; every command it builds carries it:

#![allow(unused)]
fn main() {
use processkit::{CancellationToken, CliClient};

let token = CancellationToken::new();
let gh = CliClient::new("gh").default_cancel_on(token.child_token());
// ... controller cancels `token` → every in-flight command of THIS client
// dies (whole tree), surfacing Error::Cancelled to the awaiting call.
}

Clients are cheap — scope cancellation by building one client per cancellable scope with its own (child) token, instead of threading tokens through call signatures. cli_client!-generated wrappers re-emit the builder, so Git::new().default_cancel_on(t) works for downstream crates too.

Precedence: a per-command cancel_on chained on a built command replaces the client default (explicit beats default, like a per-command timeout after default_timeout). To honor both sources, wire it explicitly — CancellationToken has no built-in merge: derive a child of the default (let c = default.child_token()), hand the command cancel_on(c.clone()), and have the second source call c.cancel(). Or simpler: build a dedicated client per scope.

Precedence and interactions

Timeout vs. cancellation. A timeout is captured; a cancellation is always an error. When both land on the same run, cancellation wins — you asked the run to stop mattering, so no result is synthesized:

use processkit::{CancellationToken, Command};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let token = CancellationToken::new();
    token.cancel();

    let err = Command::new("tool")
        .timeout(Duration::from_millis(1))   // would have been a Timeout…
        .cancel_on(token)                    // …but cancellation takes priority
        .run()
        .await
        .unwrap_err();
    assert!(matches!(err, processkit::Error::Cancelled { .. }));
    Ok(())
}

Which knob for which job:

Next: Supervision · Streaming & interactive I/O · Running commands

Errors

‹ docs index

One Error enum covers every failure mode — spawn, exit, timeout, cancellation, IO — so a caller pattern-matches on a typed variant instead of parsing strings. Several variants look alike at a glance but carry different contracts; this guide is the one page that lays them all out side by side: which variant fires from where, how to classify it, and what to do about it.

• Variant reference
• Variants that look alike but aren't
• Classifiers
• Matching under #[non_exhaustive]
• Errors and retries
• Errors and supervision

Variant reference

Variants that look alike but aren't

• Timeout is captured, Cancelled is always an error. output_string/ output_bytes return Ok with timed_out() == true on a deadline — the caller decides whether that counts as failure. A cancellation, by contrast, reports Err(Error::Cancelled) on every consuming path, streaming included, because it's a deliberate caller action, not run data. When a run both hits its deadline and gets cancelled, cancellation wins (checked first). See Precedence and interactions.
• NotReady is not Timeout. Command::timeout is the run's own contract and kills the tree; a readiness probe's within deadline is a separate clock layered on top of an already-running child, and giving up on it never kills anything. is_timeout() is false for NotReady.
• NotFound vs. Spawn. NotFound is the single representation of "program not found" — bare name or path, any platform, one variant, is_not_found() == true. Spawn is every other OS-level launch failure once the program was located (permissions, a bad cwd, a .cmd/.bat needing cmd.exe) — is_not_found() is false there, so a "not installed?" hint never fires on the wrong condition (e.g. a bad working directory).
• CassetteMiss is not is_not_found. A stale/incomplete cassette and a genuinely missing program are different failures a wrapper needs to tell apart — treating a cassette miss as "optional tool not installed" would silently hide a test-fixture bug.
• Exit vs. Signalled. Both carry captured streams and a Display diagnostic tail, but Exit has a code() and may or may not be a failure (is_success()/ok_codes decide); Signalled has no code at all — code() is None — and is always terminal.

Classifiers

Matching under #[non_exhaustive]

Error — and several of its struct-like variants — are #[non_exhaustive]: a future release can add a variant or field without that being a breaking change, but it also means every downstream match must carry a catch-all arm.

use processkit::{Command, Error};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let err = Command::new("maybe-missing-tool").run().await.unwrap_err();

    match err {
        Error::NotFound { .. } => eprintln!("is it installed?"),
        Error::Timeout { .. } => eprintln!("hit its deadline"),
        Error::Cancelled { .. } => { /* caller-initiated, nothing to log */ }
        Error::Exit { code, .. } => eprintln!("exited with {code}"),
        // #[non_exhaustive]: a future variant (or a today's variant behind a
        // feature this build doesn't enable, e.g. ResourceLimit) falls here.
        other => eprintln!("run failed: {other}"),
    }
    Ok(())
}

Prefer the classifiers above (is_not_found(), is_timeout(), …) over destructuring when all you need is a yes/no answer — they read the variant without you having to keep the catch-all arm in sync as fields are added.

Errors and retries

Command::retry(max_attempts, backoff, classifier) replays a failed run while your classifier accepts the typed error — the classifier is exactly this guide's variant table, read by the caller:

use processkit::{Command, Error};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let out = Command::new("curl")
        .args(["-fsS", "https://example.com/api"])
        .timeout(Duration::from_secs(10))
        .retry(3, Duration::from_millis(250), |e: &Error| {
            // Transient: our own deadline, or a spawn/IO condition a bare retry clears.
            e.is_timeout() || e.is_transient()
        })
        .run()
        .await?;
    Ok(())
}

Error::Cancelled is never retried, whatever the classifier says — the token stays cancelled forever, so another attempt could only fail the same way. See Retries for the full ground rules (stdin re-use, which verbs retry at all).

Errors and supervision

A Supervisor restarts a crashing service rather than replaying one operation, but it faces the same "is this permanent?" question — its give_up_when(classifier) gate is the same classifiers applied to a GiveUpAttempt:

use processkit::{Command, GiveUpAttempt, Supervisor};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let outcome = Supervisor::new(Command::new("maybe-typo'd-binary"))
        .give_up_when(|attempt| match attempt {
            // The child never started at all — e.g. is_not_found() for ENOENT.
            GiveUpAttempt::Failed(err) => err.is_not_found(),
            // A completed run your own domain knows is a permanent crash.
            GiveUpAttempt::Crashed(res) => res.code() == Some(78),
            _ => false,
        })
        .run()
        .await?;
    println!("stopped: {:?}", outcome.stopped);
    Ok(())
}

GiveUpAttempt::Failed(&Error) is the spawn/IO path (no ProcessResult was ever produced); GiveUpAttempt::Crashed(&ProcessResult<String>) is a completed-but-failing run. A recognized-permanent failure reports StopReason::GaveUp instead of restarting forever. Error::Cancelled is terminal here too — supervision returns Err(Cancelled) instead of restarting into a still-cancelled token, give_up_when or not.

Next: Running commands · Timeouts, retries & cancellation · Supervision

Supervision

‹ docs index

Where retry answers "run this once, replaying on failure", a Supervisor answers the different question "keep this alive": restart a child per policy whenever it exits, with bounded restarts, exponential backoff, and jitter — a minimal runit/systemd-style keeper, platform-agnostic because it sits entirely on the ProcessRunner seam.

• The shape
• Policies: what counts as a crash
• Backoff and jitter
• Failure storms
• Stopping
• Giving up on permanent failures
• Outcomes
• Supervising inside a shared group
• Errors and cancellation

The shape

use processkit::{Command, RestartPolicy, Supervisor};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let outcome = Supervisor::new(Command::new("my-server").args(["--port", "8080"]))
        .restart(RestartPolicy::OnCrash)           // default
        .max_restarts(5)                           // default: unlimited
        .backoff(Duration::from_millis(200), 2.0)  // default: 200ms × 2.0
        .max_backoff(Duration::from_secs(30))      // default: 30s cap
        .jitter(true)                              // default: on
        .stop_when(|res| res.code() == Some(0))    // optional exit condition
        .run()
        .await?;

    println!(
        "ended after {} restarts, reason: {:?}, last exit: {:?}",
        outcome.restarts, outcome.stopped, outcome.final_result.code(),
    );
    Ok(())
}

Each incarnation is one full captured run of the command (so the command's own timeout, stdin, env, … all apply per run — with the usual one-shot-stdin caveat for the second run onward).

Policies: what counts as a crash

A crash is any run that is not a success (ProcessResult::is_success, which honors the command's ok_codes): an exit code outside the accepted set (default {0}), a timeout, a signal-kill, or a spawn failure. A command with ok_codes([0, 2]) that exits 2 is a success, so OnCrash treats it as clean, not a crash.

Backoff and jitter

The n-th restart (0-based) sleeps

delay(n) = min(base × factor^n, max_backoff) × jitter

with jitter drawn uniformly from [0.5, 1.5) per restart. Jitter is on by default so a fleet of supervised workers restarted by the same incident doesn't stampede back in lockstep; jitter(false) gives deterministic delays (useful in tests with a paused tokio clock). A non-finite or < 1.0 factor is treated as 1.0 — constant delay, never a shrinking one.

base=200ms, factor=2.0, cap=30s:
restart #0 → ~200ms   #1 → ~400ms   #2 → ~800ms … #7 → ~25.6s   #8+ → 30s (cap)

The exponent n is not the lifetime restart count — it resets whenever a run stays up at least as long as max_backoff. So a long-lived service that crashes now and then restarts near base each time (it demonstrated health between crashes), while only a tight loop — each incarnation shorter than the cap — climbs to the ceiling. The floor is on uptime, not exit kind: under Always a worker that exits (cleanly or not) in under max_backoff is treated as flapping and escalates, which is what stops an exit 0 spin loop from hammering at the base delay. A single jittered delay can reach up to 1.5 × max_backoff.

Failure storms

Backoff spaces individual restarts; max_restarts is a lifetime cap. Neither distinguishes a service that fails once a day from one that is suddenly crash-looping. The opt-in storm guard does (a design borrowed from Go's suture supervisor — the idea, not the code):

use processkit::{Command, Supervisor};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let outcome = Supervisor::new(Command::new("worker"))
        .storm_pause(Duration::from_secs(15))     // master switch — off by default
        .failure_decay(Duration::from_secs(30))   // score half-life (default 30s)
        .failure_threshold(5.0)                   // trip point (default 5.0)
        .run()
        .await?;

    println!("storm pauses taken: {}", outcome.storm_pauses);
    Ok(())
}

Each failed run adds 1 to a score that halves every failure_decay:

score = score × 0.5^(Δt / failure_decay) + 1

• Fails rarely: the score decays back toward 1 between failures and never reaches the threshold — the guard stays out of the way.
• Failure storm: failures arrive faster than the half-life drains them, the score climbs past failure_threshold, and the supervisor takes one collective pause of storm_pause (jittered into [0.5, 1.5) like the backoff), resets the score, and resumes.

Only failures feed the score — crashes and spawn errors, not clean exits restarted under RestartPolicy::Always. The pause stacks with (runs before) the per-restart backoff, and the max_restarts budget is checked first, so a storm pause never extends an exhausted budget. Pauses taken are reported in SupervisionOutcome::storm_pauses.

Stopping

Four gates, checked in this order after every completed run:

1. stop_when(predicate) — sees the run's ProcessResult; returning true ends supervision regardless of policy (→ StopReason::Predicate). "Exit 0 is done, anything else is a crash" is the classic: stop_when(|res| res.code() == Some(0)) under RestartPolicy::Always.
2. The policy — OnCrash stops on a clean exit (→ PolicySatisfied).
3. give_up_when(classifier) — only consulted for a crash the policy would otherwise restart; recognizing it as permanent ends supervision (→ StopReason::GaveUp). See Giving up on permanent failures.
4. max_restarts(n) — at most n restarts = n + 1 total runs; an exhausted budget reports the last result (→ RestartsExhausted). max_restarts(0) means exactly one run.

give_up_when is checked before max_restarts, so a recognized-permanent crash reports the more specific GaveUp even when the budget hasn't run out yet — and before the failure-storm guard, so giving up never pays for a storm pause it was going to end anyway.

Giving up on permanent failures

Without more, the supervisor cannot tell a transient crash from a permanent one — a command that can never succeed (a missing binary, a config error that crashes on startup, a permanently-taken port) restarts forever under the default unlimited OnCrash, throttled only by backoff. give_up_when lets you recognize the unrecoverable case and stop instead:

use processkit::{Command, GiveUpAttempt, Supervisor};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let outcome = Supervisor::new(Command::new("maybe-typo'd-binary"))
        .give_up_when(|attempt| match attempt {
            // The child never even started — e.g. ENOENT for a mistyped name.
            GiveUpAttempt::Failed(err) => err.is_not_found(),
            // A completed crash your own domain knows is permanent, e.g. a
            // documented "config invalid, do not restart" exit code.
            GiveUpAttempt::Crashed(res) => res.code() == Some(78),
            _ => false,
        })
        .run()
        .await?;
    println!("stopped: {:?}", outcome.stopped);
    Ok(())
}

GiveUpAttempt distinguishes the two shapes a permanent failure can take:

• Crashed(&ProcessResult<String>) — a completed run that counts as a crash. A match reports StopReason::GaveUp in the SupervisionOutcome, same as any other stop reason.
• Failed(&Error) — the child never started at all (spawn/IO failure, e.g. ENOENT). There is no ProcessResult to report here, so a match surfaces the classified error directly as run()'s Err — the same contract an exhausted budget already has on this path (see Errors and cancellation).

Unset by default: a permanent failure restarts forever exactly as before, bounded only by max_restarts / the storm guard — adding give_up_when never changes existing behavior until you set it.

Outcomes

run() resolves to a SupervisionOutcome:

use processkit::{Command, Supervisor};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let outcome = Supervisor::new(Command::new("job")).run().await?;

    outcome.final_result; // ProcessResult<String> of the LAST run
    outcome.restarts;     // how many restarts happened (not counting run #1)
    outcome.stopped;      // StopReason::{Predicate, PolicySatisfied, GaveUp, RestartsExhausted}
    outcome.storm_pauses; // failure-storm pauses taken (0 unless storm_pause is set)
    Ok(())
}

Note run() returning Ok does not mean the child succeeded — it means supervision concluded. Inspect final_result (or ensure_success() it) for the child's own verdict.

Supervising inside a shared group

The supervisor runs through any ProcessRunner. The headline production variant injects a ProcessGroup so every incarnation — and everything it spawns — lives in one kill-on-drop container:

use processkit::{Command, ProcessGroup, RestartPolicy, Supervisor};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::new()?;

    let outcome = Supervisor::new(Command::new("worker"))
        .with_runner(&group)                 // &group is itself a ProcessRunner
        .restart(RestartPolicy::OnCrash)
        .max_restarts(10)
        .run()
        .await?;

    // The group outlives supervision: drop it (or shutdown) to reap any strays.
    Ok(())
}

Mind one interaction: don't supervise into a group you've suspended — under the cgroup mechanism the restarted child would start frozen (and the spawn itself can block). Resume first.

The same injection point makes supervision logic hermetically testable — script a sequence of fake results and assert the restart/stop behavior with no real process; see Testing your code.

Errors and cancellation

A run that produces no result at all (spawn/IO failure) can't be judged by stop_when (it needs a ProcessResult) — but it is visible to give_up_when as GiveUpAttempt::Failed. The policy treats it as a crash and restarts (with backoff) unless the policy is Never, give_up_when classifies it as permanent, or the budget is exhausted — any of those surfaces the error itself as run()'s Err.

A cancelled incarnation is terminal: run() returns Err(Error::Cancelled) immediately. The token never un-cancels, so a restart could only produce another instantly-cancelled run — the supervisor refuses the futile loop.

Next: Testing your code · Timeouts, retries & cancellation · Process groups

Testing your code

‹ docs index

Code that shells out is miserable to test — unless the subprocess is behind a seam. In processkit that seam is one small trait. Only output_string is required; output_bytes (raw-byte stdout) and start (a live handle for streaming/probes) are defaulted, so a minimal double implements just output_string:

#[async_trait]
pub trait ProcessRunner: Send + Sync {
    async fn output_string(&self, command: &Command) -> Result<ProcessResult<String>>;
    // Defaulted (route through `start`); override for byte/streaming support:
    async fn output_bytes(&self, command: &Command) -> Result<ProcessResult<Vec<u8>>>;
    async fn start(&self, command: &Command) -> Result<RunningProcess>;
}

Production code takes a runner (generically or as &dyn ProcessRunner); tests hand it a double. Five doubles ship with the crate, plus a macro that makes whole CLI wrappers testable for free.

• The ProcessRunner seam
• Scripting replies: ScriptedRunner
• Asserting invocations: RecordingRunner
• Echoing without spawning: DryRunRunner
• Expectation-style: MockRunner
• Record/replay cassettes: RecordReplayRunner
• Wrapping a CLI tool: CliClient

The ProcessRunner seam

JobRunner is the real implementation (each run in a fresh private group); a ProcessGroup is also a runner (runs land in that shared group); and impl ProcessRunner for &R means a borrowed runner works wherever an owned one does — inject &group or &recording without giving ownership away.

Every runner — real or double — gets the convenience helpers of ProcessRunnerExt for free: run (trimmed stdout, success required), run_unit, exit_code, probe (exit code as a boolean), checked (success-checked full result), and parse/try_parse (feed stdout to a closure). These are all callable on a &dyn ProcessRunner; being generic over the closure, parse/try_parse/first_line simply can't be dispatched through a dyn ProcessRunnerExt object (the ext trait isn't object-safe). Retry policies work through the seam too, so a double exercises your retry handling hermetically.

The seam covers streaming as well as bulk runs: ProcessRunner::start returns a live RunningProcess, and a ScriptedRunner's start hands back a scripted handle whose canned lines flow through the same pump machinery a real child uses — stdout_lines, wait_for_line, and finish behave identically, with no subprocess (see Scripted streaming below). An output_string-only custom runner keeps compiling: start is defaulted to Error::Unsupported.

use processkit::{Command, ProcessRunner, ProcessRunnerExt, Result};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // Production code: generic over the runner.
    async fn current_branch(runner: &impl ProcessRunner) -> Result<String> {
        runner
            .run(&Command::new("git").args(["branch", "--show-current"]))
            .await
    }
    Ok(())
}

Scripting replies

ScriptedRunner returns canned Replys for matched commands — the work-horse double:

use processkit::{Command, ProcessRunnerExt};
use processkit::testing::{Reply, ScriptedRunner};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    #[tokio::test]
    async fn detects_the_branch() {
        let runner = ScriptedRunner::new()
            // Match by program + argument PREFIX (element-wise; first element is
            // the program name, in registration order):
            .on(["git", "branch", "--show-current"], Reply::ok("main\n"))
            // …or by any predicate over the full Command:
            .when(
                |cmd| cmd.working_dir().is_some(),
                Reply::fail(128, "fatal: not a git repository"),
            )
            // …with an optional catch-all:
            .fallback(Reply::ok(""));

        assert_eq!(current_branch(&runner).await.unwrap(), "main");
    }
    Ok(())
}

The pieces:

• Reply::ok(stdout) — exit 0. Reply::fail(code, stderr) — non-zero with stderr. Reply::lines(["a", "b"]) — exit 0 with the lines joined (and streamed one by one on a scripted start). Reply::timeout() — a timed-out run (the checking helpers raise Error::Timeout from it, carrying the command's own configured deadline). On a scripted start it resolves immediately as timed-out; to exercise a real deadline race, use Reply::pending() + a Command::timeout. .with_stdout(text) — attach stdout to any of them (e.g. the CONFLICT … text git prints on a failing merge). .with_line_delay(d) — pace a scripted stream's lines.
• Reply::pending() — parks the call until the command's cancellation token (per-command cancel_on or the client-level default_cancel_on) fires, resolving with Error::Cancelled — so a test can prove an orchestration actually cancels a blocked call, not just that it formats a canned error — or until the command's timeout deadline elapses, resolving timed-out (Outcome::TimedOut) on the bulk verbs and start alike, like a child killed for overrunning its deadline. Whichever fires first wins. With neither a token nor a timeout it parks forever, like a hung child.
• Rules are tried in registration order; first match wins. Prefix matching is element-wise over the program name then the arguments (the first element is the program) — on(["git", "foo"]) matches git foo bar but not git foobar (and not rm foo). Use on_sequence to serve an ordered sequence of replies (each once, then the last repeats) for a fail-then-succeed scenario.
• No match and no fallback is a loud error (Error::Spawn, not-found) — an unexpected invocation can't slip through a test silently.
• Bulk runs also replay the canned lines through the command's on_stdout_line/on_stderr_line handlers, so a wrapper's progress-reporting path is exercised without a subprocess.

Scripted streaming

ScriptedRunner::start returns a live RunningProcess backed by the canned reply instead of an OS child. The canned stdout/stderr feed the same pump machinery a real child uses, so the whole streaming surface works hermetically — stdout_lines yields the lines, wait_for_line probes them, finish reports the canned outcome and stderr:

use processkit::prelude::StreamExt;
use processkit::testing::{Reply, ScriptedRunner};
use processkit::{Command, Finished, Outcome, ProcessRunner};
use std::time::Duration;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    #[tokio::test]
    async fn server_becomes_ready() {
        let runner = ScriptedRunner::new()
            .on(["server", "serve"], Reply::lines(["booting", "listening on 8080"]));

        let mut run = runner.start(&Command::new("server").arg("serve")).await.unwrap();
        run.wait_for_line(|l| l.contains("listening"), Duration::from_secs(5))
            .await
            .unwrap(); // satisfied by the canned banner — no subprocess

        let Finished { outcome, .. } = run.finish().await.unwrap();
        assert_eq!(outcome, Outcome::Exited(0));
    }
    Ok(())
}

Reply::lines([...]) scripts the stdout lines; .with_line_delay(d) paces them (deterministic under #[tokio::test(start_paused = true)]), and the scripted run "exits" after the last line. The honest boundaries: a scripted handle has no OS identity (pid() is None, profile reports empty samples), does not compose into a real Pipeline, and does not model interactive stdin. Reply::pending() scripts a run that never exits on its own — cancel or time it out through the command's own knobs. A command timeout does bound a scripted stream (it ends at the deadline and reports Outcome::TimedOut, like a real child), but a scripted handle has no signal tier, so — like on Windows — it ignores timeout_grace and ends at once.

Asserting invocations

RecordingRunner wraps another runner and records every Invocation — what was asked — so a test asserts inputs, not just outputs:

use processkit::{Command, ProcessRunnerExt};
use processkit::testing::{RecordingRunner, Reply, ScriptedRunner};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    #[tokio::test]
    async fn passes_the_right_flags() {
        let runner = RecordingRunner::new(
            ScriptedRunner::new().fallback(Reply::ok("done")),
        );

        runner
            .run(&Command::new("gh").args(["pr", "create", "--draft"]).current_dir("/repo"))
            .await
            .unwrap();

        let call = runner.only_call(); // panics unless exactly one call
        assert_eq!(call.args_str(), ["pr", "create", "--draft"]);
        assert!(call.has_flag("--draft"));
        assert_eq!(call.cwd.as_deref().map(|c| c.to_str().unwrap()), Some("/repo"));
        assert!(!call.has_stdin);
    }
    Ok(())
}

An Invocation captures the routing knobs — program, args, cwd, envs (explicit overrides, None = removal), has_stdin — not the I/O-shaping ones (timeout, encodings, buffer policy); assert those through a when predicate over the Command itself. calls() returns the full list when more than one run is expected.

Echoing without spawning: DryRunRunner

DryRunRunner never spawns a process at all: it renders each command through Command::command_line — the crate's own display quoting, not a hand-rolled shell-escaper — and hands back a synthetic successful result. It's the seam behind a tool's own --dry-run/--echo mode: wire your production code to it (instead of JobRunner) and it shows what would run instead of running it.

use processkit::{Command, ProcessRunner};
use processkit::testing::DryRunRunner;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    #[tokio::test]
    async fn dry_run_shows_the_command_without_running_it() {
        let runner = DryRunRunner::new();
        let out = runner
            .output_string(&Command::new("rm").args(["-rf", "build"]))
            .await
            .unwrap();
        assert!(out.is_success()); // synthetic — no process ever ran
        assert_eq!(runner.only_command(), "rm -rf build");
    }
    Ok(())
}

Unlike ScriptedRunner, there is nothing to script — a dry run has no real output to fake, only a command line to show — so every call unconditionally succeeds on both output_string and start: empty stdout/stderr, and an exit code drawn from the command's own ok_codes (0 by default) so is_success() and the ergonomic run/run_unit/checked/parse verbs agree it succeeded even for a command whose ok_codes excludes 0. The rendered lines are available two ways, usable together or alone:

• a collected snapshot, in the style of RecordingRunner::calls — commands() (all of them, in order) / only_command() (panics unless exactly one call was made);
• a live on_invocation(|line| …) callback, invoked with the rendered line as each call happens — e.g. printing it to the terminal immediately, in addition to (not instead of) the collected snapshot:

#![allow(unused)]
fn main() {
use processkit::testing::DryRunRunner;

let runner = DryRunRunner::new().on_invocation(|line| println!("+ {line}"));
}

Expectation-style: MockRunner

With the mock feature, mockall generates a MockRunner for expectation-style tests (call counts, argument matchers, ordered expectations) — the right tool when the interaction is the contract.

Note: MockRunner's expect_* surface is generated by mockall and is exempt from this crate's semver guarantees — it tracks the mockall dependency, not a frozen API. For a stable double, prefer ScriptedRunner (canned replies) or RecordingRunner (input assertions) above.

#![allow(unused)]
fn main() {
use processkit::testing::MockRunner;

let mut mock = MockRunner::new();
mock.expect_output_string()
    .times(1)
    .returning(|_cmd| todo!("build a Result<ProcessResult<String>>"));
}

MockRunner does not inherit the defaults. Unlike a hand-written runner (where output_bytes/start are defaulted), mockall::automock replaces every method with an expectation — so a verb that routes through start or output_bytes needs its own expect_start() / expect_output_bytes(), or the unset call panics ("no expectation"). ScriptedRunner provides the defaults and the streaming seam out of the box.

For most tests ScriptedRunner/RecordingRunner read better; reach for the mock when you need mockall's matching machinery.

Record/replay cassettes

With the record feature, RecordReplayRunner closes the loop: record real runs to a JSON cassette once, then replay them deterministically — fast, hermetic, byte-stable, no subprocess in CI:

use processkit::{Command, JobRunner, ProcessRunnerExt};
use processkit::testing::RecordReplayRunner;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // Record once against the real tool (an opt-in `--record` test run, say):
    let runner = RecordReplayRunner::record("fixtures/git.json", JobRunner::new());
    let version = runner.run(&Command::new("git").arg("--version")).await?;
    runner.save()?;                                  // the error-surfacing flush
                                                     // (best-effort on drop too)

    // Replay everywhere else:
    let runner = RecordReplayRunner::replay("fixtures/git.json")?;
    assert_eq!(runner.run(&Command::new("git").arg("--version")).await?, version);
    Ok(())
}

Semantics worth knowing before you commit a cassette:

Only env values are redacted. program, args, cwd, stdout, and stderr are stored verbatim and can carry secrets (a --password=… flag, a token echoed to output), so review a fixture before committing it. On Unix the file is written 0600 and the write refuses to follow a symlink at the cassette path (O_NOFOLLOW, so a planted link can't redirect the secret-bearing write — it fails loud instead). On Windows the file inherits the containing directory's ACL, so restrict that directory (or use a per-user temp dir, not a world-writable shared one) for secret-bearing fixtures.

A neat trick: in tests, record against a ScriptedRunner instead of JobRunner — the whole record→save→replay round trip is then itself hermetic.

Opt-in stricter matching (cwd / selected env values)

The portable default keys on program + args + stdin only. It deliberately leaves cwd and the environment out of the key: a cassette recorded in one absolute working directory (a dev box, a tempdir) then replays cleanly from a different one (a CI workspace), and an env variable that differs between the record and replay machines but doesn't change the tool's output can't cause a spurious miss. That portability is the right default for most tools.

But some tools' output genuinely depends on where they run or on a specific environment variable — and with those out of the key, two such invocations collide on one entry: the first recording silently answers for both on replay. When that matters, opt in to a stricter key:

use processkit::{Command, JobRunner, ProcessRunnerExt};
use processkit::testing::RecordReplayRunner;

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // Record: also key on the working directory and on LC_ALL's value.
    let runner = RecordReplayRunner::record("fixtures/tool.json", JobRunner::new())
        .match_on_cwd()
        .match_on_env(["LC_ALL"]);
    let cmd = Command::new("tool").current_dir("/repo").env("LC_ALL", "C");
    let _ = runner.run(&cmd).await?;
    runner.save()?;

    // Replay: set the *same* policy — it must match on both sides.
    let runner = RecordReplayRunner::replay("fixtures/tool.json")?
        .match_on_cwd()
        .match_on_env(["LC_ALL"]);
    // A differing cwd or LC_ALL value now MISSES (a loud `CassetteMiss`) instead
    // of replaying the wrong entry; the same cwd + LC_ALL hits.
    let cmd = Command::new("tool").current_dir("/repo").env("LC_ALL", "C");
    let _ = runner.run(&cmd).await?;
    Ok(())
}

Notes:

• Env values still never reach the file. match_on_env keys on an FNV digest of the selected (name, value) pairs, not the raw values — the cassette continues to store variable names only, so no env secret is written to disk even under the stricter policy. (cwd is keyed the same way; it is also already stored verbatim on the entry, as before.)
• Symmetric by contract. Set the same policy on the record and replay runner, exactly as you target the same tool. A mismatched policy simply misses (it never serves a wrong entry) — including replaying a policy-keyed cassette with no policy, or vice versa.
• Only the named variables participate. Variables you don't name stay out of the key, preserving portability for env differences that don't matter. A named variable that is set, removed (env_remove), or untouched are three distinct keys.
• On-disk format. A policy-keyed entry carries an extra opaque match_digest number; a cassette recorded without a policy omits it and is byte-identical to before but for the bumped version (now 4). Older cassettes (no match_digest) load and replay unchanged under a no-policy replayer.

Wrapping a CLI tool

CliClient is the foundation for typed wrappers around external tools (git, jj, gh, kubectl, …): it owns the program name, per-client defaults, and the runner; your wrapper contributes only commands and parsers. The cli_client! macro generates the boilerplate:

use processkit::{cli_client, Error, ProcessRunner, Result};
use std::path::Path;
use std::time::Duration;

cli_client!(
    /// A typed `git` client.
    pub struct Git => "git"
);

impl<R: ProcessRunner> Git<R> {
    /// HEAD's commit id.
    pub async fn head(&self, repo: &Path) -> Result<String> {
        self.core.run(self.core.command_in(repo, ["rev-parse", "HEAD"])).await
    }

    /// Is the work tree clean? (exit code IS the answer)
    pub async fn is_clean(&self, repo: &Path) -> Result<bool> {
        self.core.probe(self.core.command_in(repo, ["diff", "--quiet"])).await
    }

    /// Branch list, parsed — the parser is fallible and returns the crate's
    /// `Result`, typically an `Error::Parse` naming the program.
    pub async fn branches(&self, repo: &Path) -> Result<Vec<String>> {
        self.core
            .try_parse(
                self.core.command_in(repo, ["branch", "--format=%(refname:short)"]),
                |out| {
                    let list: Vec<String> = out.lines().map(str::to_owned).collect();
                    if list.is_empty() {
                        Err(Error::parse("git", "no branches"))
                    } else {
                        Ok(list)
                    }
                },
            )
            .await
    }
}

#[tokio::main]
async fn main() -> processkit::Result<()> {
    // Production: the real runner, with per-client defaults.
    let git = Git::new().default_timeout(Duration::from_secs(30));
    let head = git.head(Path::new(".")).await?;
    Ok(())
}

The generated type is Git<R: ProcessRunner = JobRunner> with Git::new(), Git::with_runner(runner), default_timeout / default_env / default_env_remove builders, and a module-private core: CliClient<R> (reach it as self.core from the wrapper's own methods) whose helpers speak the crate-wide verb vocabulary: run (trimmed stdout), output_string (full result), run_unit (success only), exit_code, probe, plus parse (infallible) and try_parse (fallible → Error::Parse).

And the payoff — the wrapper tests hermetically with any double:

#![allow(unused)]
fn main() {
#[tokio::test]
async fn head_is_trimmed() {
    let git = Git::with_runner(
        ScriptedRunner::new().on(["git", "rev-parse", "HEAD"], Reply::ok("abc123\n")),
    );
    assert_eq!(git.head(Path::new("/repo")).await.unwrap(), "abc123");
}
}

…or with a cassette recorded against the real tool once.

Next: Platform support · Supervision · Running commands

Platform support

‹ docs index

processkit supports Unix and Windows only — it requires tokio::process and OS job / process-group primitives that have no equivalent on bare targets like wasm. Building for such a target fails at compile time (a compile_error! guard, or earlier in tokio's own dependencies). Within the supported set, it treats platform support as first-class: every capability is either fully implemented, honestly partial (documented and typed), or refused with Error::Unsupported — never silently skipped. This page collects all the matrices and fine print in one place.

• CI coverage
• Containment mechanisms
• Capability matrices
• Caveats

CI coverage

.github/workflows/ci.yml's test job runs the full real-subprocess suite (--include-ignored, so kill-on-drop is actually exercised) on glibc x86_64 (ubuntu-latest), glibc aarch64 (ubuntu-24.04-arm), Windows and macOS. The aarch64 leg is the only place the native Linux syscall/layout code (sys/{linux,pgroup,unix,pid_gate}.rs) actually runs on Linux/glibc-aarch64 — elsewhere aarch64 is only cargo check-compiled against aarch64-apple-darwin (Darwin, in the msrv job), never executed. A separate test-musl job runs the full suite a further time inside a real rust:alpine container — musl libc and a busybox userland, not merely a cross-compiled x86_64-unknown-linux-musl binary executed by glibc userland tools — because musl/Alpine is the de-facto standard for the container images this crate actually runs in, and its libc, signal, and userland-utility details genuinely differ from glibc's. Alpine's busybox already covers every external utility the suite spawns (sh, cat, sleep, yes, head, grep, sort, id, printf, env, seq) except one: its ps applet has no -p PID filter, so the job installs procps (procps-ng) to get one that does. gcc/musl-dev (needed to link) already ship in the base image.

Two container-runtime quirks — unrelated to musl/Alpine itself, but specific to running any test suite inside a plain container — need working around, via the job's (and the just test-musl recipe's) --init and --cap-add=SYS_NICE options:

• No subreaper. A plain container's PID 1 is just the job's own entry process, not a real init — so a killed process's orphaned grandchildren become zombies that are never reaped and still probe alive via kill(pid, 0). That silently breaks every test that asserts a forked grandchild is actually gone after teardown. --init runs tini as PID 1, a real subreaper, which is what a properly configured container host provides. Any container's own PID 1 has the identical gap for its own orphaned grandchildren — see Running in containers → PID 1.
• CAP_SYS_NICE dropped by default. Docker excludes it from the default capability set even for a root container user, so raising scheduling priority fails EPERM regardless of uid (lowering it never needs the capability). One test exercises Priority::High (a negative nice) together with a privilege drop; --cap-add=SYS_NICE restores just that one narrow, low-blast-radius capability rather than skipping the test.

Run the same job locally with just test-musl (requires Docker); see the recipe in justfile for details.

Containment mechanisms

ProcessGroup::mechanism() reports which one you actually got:

On Linux the cgroup backend requires controller delegation, and resource limits specifically need this process to run at the real cgroup-v2 root. The crate creates the limit cgroup under this process's own cgroup and enables the controllers in that cgroup's subtree_control, which cgroup v2's "no internal processes" rule allows only for the real hierarchy root (the one exempt cgroup). A cgroup namespace root does not qualify — it only virtualizes the view — so an ordinary (private-cgroupns) container fails EBUSY just like a systemd session/scope/service. The crate does not migrate your process into a sub-cgroup to work around it, so in practice limits apply only at a minimal non-systemd init sitting at the real root. Without a usable cgroup it quietly falls back to ProcessGroup — unless you requested resource limits, which fail fast instead (Error::ResourceLimit), because an unapplied cap is no protection. The error's reason distinguishes the two ways this happens: LimitReason::Unsupported when no cgroup v2 is mounted at all (or on macOS/BSD, which has no whole-tree container of any kind), LimitReason::Unenforceable when cgroup v2 exists but this process isn't at the real hierarchy root (the delegation case above) or the OS otherwise rejected the request.

Capability matrices

Teardown & containment

Windows has no signal tier, so a graceful shutdown collapses to the atomic Job kill — but it still honors escalate_to_kill: false spares the survivors (closes the Job handle without KILL_ON_JOB_CLOSE) rather than killing them, so the Windows column is "atomic kill when it kills", not an unconditional kill.

Signals & freezing

On the cgroup mechanism, a non-Kill signal (and the SIGSTOP/SIGCONT fallback used for suspend/resume on pre-5.2 kernels without cgroup.freeze) surfaces a real per-member delivery failure (e.g. EPERM) as an Err rather than swallowing it — consistent with the "never silently skipped" philosophy; an ESRCH race (the member already exited) is still success.

Inspection & accounting

members() is gated on the process-control feature; the CPU / memory / profile rows are gated on the stats feature.

Resource limits (limits feature)

Spawn-time controls

Everything not listed — capture, streaming, interactive stdin, encodings, buffer policies, timeouts, retry, pipelines, supervision, readiness probes, the test doubles, cassettes, cancellation — is platform-agnostic and behaves identically everywhere.

Caveats

The honest fine print, mostly consequences of OS semantics:

Windows: termination is an exit code, never Signalled (D18). Windows has no signal abstraction, so a killed process reports Outcome::Exited, not Outcome::Signalled. TerminateProcess / TerminateJobObject(_, 1) is Exited(1) — indistinguishable from a voluntary exit(1) — and Ctrl-C surfaces as Exited(-1073741510) (STATUS_CONTROL_C_EXIT as a signed i32). The crate reports the platform truth rather than fabricating a Signalled from an NTSTATUS code (that mapping would be a lossy guess). When you need to know the run was killed, use a ProcessGroup deadline or a cancellation token (which surface as TimedOut / Error::Cancelled on every platform). Outcome::Signalled is therefore Unix-only.

Linux cgroup delegation. Creating the per-group cgroup needs write access to the cgroup v2 hierarchy. Dev boxes typically lack it → the pgroup fallback. CI inside containers usually has it. Check mechanism() when behavior must not silently degrade. For the container-specific version of this — what a plain docker run actually gets, and why resource limits stay unenforceable even under --privileged — see Running in containers → mechanism and Running in containers → resource limits.

uid()/gid() × the cgroup mechanism. The OS applies the uid drop before pre_exec hooks, and the cgroup join runs in pre_exec — as the already-dropped user, who can't write the root-owned cgroup.procs. The spawn fails with a permission error (never an uncontained child). Privilege drop composes cleanly with the process-group mechanism.

setsid() × process groups. A new session implies a new process group; the crate coordinates the two (the containment tracking follows the new session's group), so setsid keeps the kill-on-drop guarantee instead of breaking out of it.

kill_on_parent_death() is thread-scoped on Linux. PR_SET_PDEATHSIG fires when the spawning thread dies, not only the process. On a multi-threaded tokio runtime a retired worker thread could kill the child early; spawn from a current-thread runtime for the strongest guarantee. It covers the direct child only — with the parent SIGKILLed, nothing tears the cgroup/pgroup down, so grandchildren survive. The parent-died-before-arming race is closed by re-checking getppid() in the child against the spawner's pid captured before the fork — which stays correct when the spawner itself is PID 1 (a container entrypoint).

Windows: the suspended-spawn handshake. Children are created CREATE_SUSPENDED, assigned to the job, then resumed — closing the classic race where a fast child forks before it's in the job. A consequence: on the raw ProcessGroup::spawn escape hatch, any creation flags the caller set are overwritten — the child is forced to CREATE_SUSPENDED alone, because Win32 exposes no way to read the flags back and OR the suspend bit in. The Command-driven paths don't have this limitation: their extras (incl. create_no_window) travel alongside the OS command and are OR'd in.

Windows: nested suspends. SuspendThread keeps per-thread counts — two suspend() calls need two resume()s. The POSIX backends are level-triggered (idempotent). Suspension is also best-effort against a tree that is spawning threads mid-walk.

Spawning into a suspended cgroup group. The freeze is group state: a child spawned or adopted while suspended joins frozen — the forked child joins the cgroup before exec, so it can freeze before completing the spawn handshake and start() may never return until resume. Resume before starting new work; details in Process groups.

Frozen trees and graceful shutdown. Hard kills penetrate a frozen tree (SIGKILL / cgroup.kill / job terminate), but a graceful shutdown leads with a SIGTERM the frozen processes can't handle — it waits out the full grace. Resume first. For the orchestrator's own SIGTERM to your container's PID 1 (a related but distinct signal from the one shutdown sends to the tree it manages), see Running in containers → graceful shutdown.

pgroup backends: leaders, zombies, pid reuse. members() lists tracked group leaders only; an exited-but-unreaped child (zombie) still probes as alive (keep wait()ing handles if you need prompt liveness, e.g. for shutdown's early return); and pid-based signalling is inherently best-effort against pid reuse — the crate prunes dead entries on every probe to keep the window minimal.

Next: Process groups · Running in containers · docs index

Running in containers

‹ docs index

Containers are where most processkit-using services actually run, and it's the environment with the most sharp edges: which containment mechanism you get depends on privileges the orchestrator may or may not grant, your process is usually PID 1 (with everything that implies for signal delivery and reaping), and the image itself is often minimal (musl/Alpine, sometimes no shell at all). None of this is new machinery — every fact below is already documented in Platform support — this page is the container-shaped tour through it, with the Dockerfile fragments and gotchas that only show up once you actually run the crate inside docker run / Kubernetes.

• Which containment mechanism you get
• PID 1: signals, zombies, and what's contained
• Graceful shutdown on the orchestrator's SIGTERM
• Minimal images: musl/Alpine, no shell, no setpriv
• Container resource limits vs the crate's limits

Which containment mechanism you get

On Linux, ProcessGroup's cgroup backend needs write access to the cgroup v2 hierarchy at the real hierarchy root (see Platform support → containment mechanisms for exactly why). A plain, unprivileged docker run container gets neither: /sys/fs/cgroup is typically mounted read-only, so the crate quietly falls back to the ProcessGroup (POSIX process-group) mechanism — kill-on-drop still works, but the whole-tree accounting and limits capabilities of the cgroup backend do not (confirmed by running the crate inside docker run with no extra flags: mechanism() reports ProcessGroup, and /sys/fs/cgroup refuses even a touch). --privileged (or an equivalent cgroup-namespace delegation) makes the filesystem writable enough for mechanism() to report CgroupV2, but the container's cgroup is still a namespace root, not the real hierarchy root — /proc/self/cgroup reads 0::/ either way — so resource limits stay unenforceable even then.

use processkit::{Mechanism, ProcessGroup};

fn main() -> processkit::Result<()> {
    let group = ProcessGroup::new()?;
    if group.mechanism() == Mechanism::ProcessGroup {
        // The common case inside an unprivileged Linux container: kill-on-drop
        // still holds (see the next section), but `members()`'s CPU/memory
        // totals and `limits` are not available — see the capability
        // matrices in Platform support. `CgroupV2`/`JobObject` get those too.
    }
    Ok(())
}

Never assume a mechanism; check mechanism() if your service's behavior must not silently degrade (e.g. it relies on limits for sandboxing an untrusted child) — see Container resource limits for the fail-fast alternative when a cap truly matters. Whichever mechanism you land on, whole-tree kill-on-drop itself is unconditional — the fallback only narrows accounting and limits, never containment.

PID 1: signals, zombies, and what's contained

Inside a container your process is almost always PID 1 — Docker and Kubernetes don't run a real init unless you ask for one. PID 1 has two kernel duties an ordinary process doesn't: it's the implicit reparent target for every orphaned descendant in the container's PID namespace, and — for signals without an installed handler — some default dispositions are ignored instead of applied (irrelevant to SIGKILL/SIGSTOP, which are never blockable, but relevant to graceful termination — see the next section). Neither duty is something processkit does for you, and neither needs to be:

• Containment (killing the tree) is unaffected. The whole point of the cgroup/pgroup mechanisms is that kill_all/shutdown/Drop reach every descendant, including ones spawned after your direct child — that's true whether or not your process happens to be PID 1.
• Reaping an orphan that isn't yours is not. A grandchild your own child forked and then exited without waiting for still needs someone to call wait() on it once it dies, or it lingers as a zombie (kill(pid, 0) still reports it alive) forever. Once that grandchild reparents to PID 1, "someone" has to be PID 1 itself — and an ordinary process, processkit-managed or not, doesn't indiscriminately reap processes it never spawned.

This is exactly the gap Platform support's CI section documents and works around with --init for the crate's own test suite — and it reproduces identically for any container. Run a process that orphans a short-lived grandchild as the container's PID 1 with a plain docker run (no --init), and the grandchild is left a permanent zombie after it exits; the identical container run with --init (which runs tini as the real PID 1, a subreaper) shows no zombie at all — confirmed by running both side by side. Baking tini into the image itself works identically, and is the only option on Kubernetes, which has no --init-equivalent flag:

FROM alpine:latest
RUN apk add --no-cache tini
COPY my-app /usr/local/bin/my-app
ENTRYPOINT ["/sbin/tini", "--", "/usr/local/bin/my-app"]

(Confirmed against a real pk-container-demo image built from this exact pattern: with tini as ENTRYPOINT, an orphaned grandchild is reaped with no --init flag at all; the same image invoked without tini — the bare binary as PID 1 — leaks the identical zombie that the plain docker run case above does.) sh -c 'sleep 0.3 &'-style orphans are the illustrative worst case; in practice this only bites processes with descendants that outlive their direct parent — most single-service containers with no forking children never hit it. When in doubt, run with a subreaper as PID 1; it's a no-op cost otherwise.

Graceful shutdown on the orchestrator's SIGTERM

Docker (docker stop) and Kubernetes both stop a container by sending SIGTERM to PID 1 — your process — then wait a grace period (Docker's --stop-timeout / docker stop -t, Kubernetes' terminationGracePeriodSeconds, both default to a small handful of seconds) before escalating to SIGKILL. That SIGTERM targets your process, not the tree processkit manages — the two are related but distinct signals: catching the orchestrator's SIGTERM is ordinary application code (e.g. tokio::signal::unix::signal with tokio's own signal feature, or a crate like signal-hook); reacting to it by tearing down the child tree gracefully is ProcessGroup::shutdown (or RunningProcess::shutdown for a single start()ed service): SIGTERM the tree, wait up to shutdown_timeout, then SIGKILL any survivor.

use processkit::{Command, ProcessGroup};

#[tokio::main]
async fn main() -> processkit::Result<()> {
    let group = ProcessGroup::new()?;
    let _server = group.start(&Command::new("app-server")).await?;

    // Left to the application: install a handler for the orchestrator's own
    // SIGTERM (tokio::signal::unix::signal(SignalKind::terminate()) with
    // tokio's `signal` feature, or the `signal-hook`/`ctrlc` crates) and
    // resolve this future when it fires.
    wait_for_orchestrator_sigterm().await;

    // SIGTERM the tree, wait shutdown_timeout, SIGKILL stragglers:
    group.shutdown().await?;
    Ok(())
}

async fn wait_for_orchestrator_sigterm() {
    // …
}

(This exact pattern — tokio::signal::unix::signal(SignalKind::terminate()) followed by group.shutdown() — was built and run for real as a small processkit-consuming binary: docker stop on the resulting container delivered SIGTERM to PID 1, the handler fired, shutdown() tore the tree down, and the process exited well inside Docker's default 10-second grace — no SIGKILL needed.)

Two things worth setting deliberately, both already documented at the ProcessGroup/Command level:

• The orchestrator's grace period must exceed your own. If shutdown_timeout (or a per-Command timeout_grace) is longer than Docker's --stop-timeout / Kubernetes' terminationGracePeriodSeconds, the orchestrator sends the hard SIGKILL to your still-shutting-down process before your own escalation ever gets a chance to run — set the outer grace at least as generous as the inner one.
• A frozen tree can't shut down gracefully. If you've called group.suspend(), the frozen processes can't run their SIGTERM handler — shutdown() waits out the whole grace and then hard-kills. Resume first; see Platform support's frozen-tree caveat.

Minimal images: musl/Alpine, no shell, no setpriv

Two properties of a lean final image turn out to be non-issues for this crate specifically, because of how it does privilege drop and spawning — confirmed by building and running against rust:alpine/alpine:latest:

• No setpriv/su-exec/gosu needed for uid()/gid()/groups(). The privilege drop is setgroups → setgid → setuid called directly as raw syscalls inside the child's pre_exec hook (see Running commands → privileges) — the crate never shells out to an external helper binary, so a final stage that lacks one entirely still drops privileges correctly. Confirmed by running the drop (.uid(...).gid(...).groups(...)) inside a container and getting back the target identity with no such binary invoked.
• No shell needed to spawn. Command/ProcessGroup::spawn always execs the target program directly with an argv array — there is no sh -c step anywhere in the crate's own spawn path (pipelines wire pipes at the OS level, not through a shell either — see Pipelines). A FROM scratch-style final stage with no shell at all works, as long as the programs you spawn are themselves present — the crate not needing a shell doesn't exempt a command you invoke as sh -c "…" yourself.

# Builder: musl/Alpine — the same base the crate's own CI `test-musl` job and
# `just test-musl` build against (see Platform support → CI coverage), so the
# toolchain and libc pairing is already exercised.
FROM rust:alpine AS builder
WORKDIR /src
COPY Cargo.toml Cargo.lock ./
COPY src ./src
RUN cargo build --release

# Runtime: no shell, no setpriv/su-exec/gosu — the crate needs none of them.
FROM alpine:latest
COPY --from=builder /src/target/release/my-app /usr/local/bin/my-app
ENTRYPOINT ["/usr/local/bin/my-app"]

(Built and run for real, end to end: cargo build --release inside rust:alpine, then the resulting binary run from a bare alpine:latest final stage — privilege drop and a no-shell spawn both succeeded exactly as above.) Add tini to the final stage — see PID 1 — if anything you spawn can outlive its own children.

Container resource limits vs the crate's limits

The orchestrator's own cgroup limits (Docker's --memory/--cpus, Kubernetes' resources.limits) apply to the whole container, enforced by the kernel regardless of anything processkit does — a container that hits its memory limit gets OOM-killed independent of any max_memory the crate was asked to set. That outer limit is not the same thing as the crate's own limits feature (max_memory / max_processes / cpu_quota on ProcessGroupOptions), which caps a specific tree within the container and needs this process to sit at the real cgroup v2 hierarchy root — a requirement an ordinary container essentially never meets, privileged or not (see Which containment mechanism you get). Confirmed by actually requesting a limit from inside a container:

plain docker run:       Error::ResourceLimit { kind: Memory, reason: Unenforceable,
                           detail: "…Read-only file system…" }
docker run --privileged: Error::ResourceLimit { kind: Memory, reason: Unenforceable,
                           detail: "…cgroup v2's 'no internal processes' rule…Resource busy…" }

Both fail the same way the crate documents for any non-delegated host — Error::ResourceLimit with reason: LimitReason::Unenforceable — never a silently-unbounded group (see Errors → ResourceLimit and Platform support → containment mechanisms for the delegation requirement in full). In practice: rely on the orchestrator's own memory/CPU limits as the outer boundary for anything running in a container, and reserve the crate's limits feature for hosts where this process genuinely owns the cgroup root — a minimal, non-systemd init on bare metal or a VM, not a container. mechanism()/kill-on-drop containment keep working either way; only the limits cap itself is unavailable.

Next: Platform support · Process groups · docs index

Upgrading processkit

Per-version notes for consumers moving their dependency forward: what breaks, who it affects, and the exact change to make. The CHANGELOG is the full record; this page is the "I depend on it, what do I do" view.

Versioning. From 1.0.0 onward processkit follows Semantic Versioning: the public API is stable, and any breaking change lands only in a new major version, so 2.x upgrades are backward-compatible. The default Cargo requirement processkit = "2" (or "2.1" to also require the latest breaking release) already does the right thing — it allows 2.* but not 3.0. Skim the relevant section here before each major bump. (The mock feature's mockall-generated expect_* surface stays semver-exempt — it tracks the mockall version.)

2.1.0 (from 1.2.x)

2.0.0 and 1.3.0 were withdrawn — upgrade straight from 1.2.x to 2.1.0. 2.0.0 was published in error and yanked; 1.3.0 accidentally shipped this breaking batch under a minor bump and was yanked too. 2.1.0 is the first supported release of the changes below — the crate follows semver, so this break lands in a major as intended. There is nothing extra to do for the skip; the migration from a 1.2.x dependency is exactly the notes here.

Mostly mechanical renames — caught by the compiler — plus two #[non_exhaustive] tightenings on Error (also compiler-caught, once you stop destructuring the affected variants field-exhaustively) and one genuine behavior change on output_bytes that a build alone won't surface.

Renames (mechanical — compiler-caught)

The terminate_all / avg_cpu entries were deprecated forwarding aliases since 1.1.0 (see the 1.1.0 changelog entry); this release removes them outright. RunProfile::exit_code duplicated outcome.code(), which RunProfile::code() already exposed — the field is gone, the method is the one accessor now.

Error's data-carrying variants are now individually #[non_exhaustive]

Exit, Timeout, Signalled, Spawn, NotFound, Parse, OutputTooLarge, Stdin, and — with the limits feature — ResourceLimit can no longer be struct-literal-constructed or field-exhaustively destructured outside the crate.

Before:

match err {
    Error::Exit { program, code, stdout, stderr } => { /* ... */ }
    _ => {}
}

After — add .. to the pattern (or, better, use the existing accessors instead of destructuring at all):

#![allow(unused)]
fn main() {
use processkit::Error;
fn handle(err: Error) {
match &err {
    Error::Exit { program, code, stdout, stderr, .. } => { let _ = (program, code, stdout, stderr); }
    _ => {}
}

// or, accessor-based and immune to the next field addition:
if let Some(code) = err.code() {
    // err.program() / err.stdout() / err.stderr() / err.combined() also work
    let _ = code;
}
}
}

This is prep for future field additions to any of these variants without another breaking change — the Exit/Timeout/Signalled variants already gained one such field this release (next entry).

Error::Exit / Timeout / Signalled gain a stdout_bytes field

A new field, stdout_bytes: Option<Vec<u8>>, carries the exact captured stdout bytes for a checking-verb error built over output_bytes (e.g. output_bytes().await?.ensure_success()?); read it through Error::stdout_bytes() -> Option<&[u8]>, not by destructuring the variant directly (they are #[non_exhaustive] — see above). None on the text path (output_string/run/checked/…), where the decoded stdout string is already the whole story.

Error::ResourceLimit is restructured (limits feature)

Fix a match:

// Before
Error::ResourceLimit { message } => warn!("limit rejected: {message}"),

// After
Error::ResourceLimit { detail, .. } => warn!("limit rejected: {detail}"),

// or, branch on the structured classification instead of parsing text:
if let (Some(kind), Some(reason)) = (err.limit_kind(), err.limit_reason()) {
    match (kind, reason) {
        (LimitKind::Memory, LimitReason::Unsupported) => { /* ... */ }
        _ => {}
    }
}

output_bytes now honors the byte cap on stdout too — a behavior change

Not compiler-caught: if you configured an OutputBufferPolicy byte ceiling (with_max_bytes) and called output_bytes, the cap previously bounded only the line-pumped stderr; raw stdout capture was unbounded regardless of the configured max_bytes. It now applies to both streams:

• OverflowMode::Error past the cap now errors on stdout overflow too, with Error::OutputTooLarge { max_lines: None, .. } (raw bytes have no lines).
• The drop modes (head/tail) now bound retained stdout bytes the same way they already bounded stderr, and set ProcessResult::truncated.

If nothing sets a byte cap, capture stays unbounded exactly as before — nothing to do. If you do set one and rely on output_bytes returning the full stdout regardless, re-check that call site: it now truncates/errors like every other capture path under the same policy.

Cassette replay: cwd no longer part of the match key — no action needed

RecordReplayRunner (record feature) replays a cassette recorded from one absolute working directory against the same invocation run from a different one, instead of CassetteMissing — cwd is still stored on each entry for visibility, it just no longer discriminates two otherwise-identical recorded runs. The on-disk format revision bumped to 3, but this is not a compatibility gate: a cassette written by a 1.x build still loads and replays fine. The one edge case: an existing cassette that had two entries differing only in cwd now collides on replay, and the first-recorded entry answers for both — re-record it if that matters for your fixtures.

Verify the upgrade

cargo update -p processkit
cargo build      # the renames and non_exhaustive tightenings are compiler-caught
cargo test       # catches the output_bytes byte-cap behavior change if you rely on it

1.0.0 (from 0.11.x)

A few breaking changes, all caught by the compiler — if it builds after the bump, you're done.

OutputLine.text is now an accessor

OutputLine (the per-line payload of RunningProcess::output_events) no longer exposes text as a public field — read it via line.text() -> &str (or line.into_text() -> String to take ownership). This frees the line representation to evolve. Fix: line.text → line.text().

Error::ResourceLimit is now a struct variant

Error::ResourceLimit(String) became Error::ResourceLimit { message: String } (parity with the other rich variants, room for structured detail later). Fix a match Error::ResourceLimit(m) → Error::ResourceLimit { message: m }. (Only relevant with the limits feature.)

The text-capture verb is renamed output → output_string

The verb that runs to completion and returns the full ProcessResult<String> is now spelled output_string on every layer, matching output_bytes (and the spelling Command/Pipeline/RunningProcess already used). Two reasons: the same operation no longer has two names depending on the type, and a bare output clashed with std::process::Command::output, which returns bytes — the explicit name removes that footgun.

Affected if you call ProcessRunner::output, CliClient::output, the free fn processkit::output, or implement a custom ProcessRunner / use MockRunner. The symptom is a build error like "no method named output" / "cannot find function output in crate processkit".

Fix — rename the calls (mechanical):

output_bytes is unchanged, and Command/Pipeline/RunningProcess callers need no change (those already used output_string).

0.11.0 (from 0.10.x)

Two breaking changes, both small and caught by the compiler — if it builds after the bump, you're done. Plus one internal fix that needs no action.

1. stats is now opt-in — a Cargo.toml change

The default feature set is now just process-control; stats is no longer on by default. (It gates a specialized metrics surface the core never needs; on Windows it links an OS library — the ProcessStatus FFI used solely for the peak-memory readout — but unlike mock/tracing/record it pulls in no extra crate.)

Affected if you use any metrics API: ProcessGroup::stats / ProcessGroupStats, RunningProcess::cpu_time / peak_memory_bytes, or RunProfile / RunningProcess::profile. The symptom is a build error like "no method named stats / cpu_time / peak_memory_bytes / profile" or "cannot find type ProcessGroupStats / RunProfile".

Fix — add the feature:

[dependencies]
processkit = { version = "0.11", features = ["stats"] }

If you already enable limits, do nothing — limits still implies stats.

If you don't use metrics: nothing to do. Your default build is now slightly leaner (no Windows ProcessStatus dependency).

2. OutputEvent carries OutputLine — a code change

Affects only callers of RunningProcess::output_events (the ordered lifecycle+output event stream). The per-line payload changed from a bare String to a #[non_exhaustive] OutputLine struct with a public text field.

Before:

use processkit::OutputEvent;

while let Some(ev) = events.next().await {
    match ev {
        OutputEvent::Stdout(s) => println!("out: {s}"),
        OutputEvent::Stderr(s) => eprintln!("err: {s}"),
        _ => {}
    }
}

After — read line.text (in 1.0 this becomes line.text(); see the 1.0.0 section above):

match ev {
    OutputEvent::Stdout(line) => println!("out: {}", line.text),
    OutputEvent::Stderr(line) => eprintln!("err: {}", line.text),
    _ => {}
}

Or, when you don't care which stream produced the line, use the new accessor:

#![allow(unused)]
fn main() {
fn handle(ev: processkit::OutputEvent) {
if let Some(text) = ev.text() {
    println!("{text}");
}
}
}

OutputLine is #[non_exhaustive]: you receive it from the crate and read its fields — you don't construct it, and a match on it should use ... The change exists to reserve room for per-line metadata (e.g. a timestamp or a monotonic line index) in a later release without another break.

3. Cancel-precedence fix ("Issue 7") — no action

A run that reaps on its own is no longer at risk of being misreported as Err(Cancelled) by a cancellation token that fires in the narrow window between the reap and the disposition check. This is an internal correctness fix with no public-API change. If you carried a workaround that tolerated a spurious Cancelled on a self-completing run, you can remove it.

Verify the upgrade

cargo update -p processkit
cargo build      # both breaking changes are compiler-caught
cargo test

Upgrading from older than 0.10

The jumps below 0.10 predate this guide. Read the dated sections of the CHANGELOG for each minor you cross — every breaking entry there is marked Breaking and carries its own migration note. Notable recent non-breaking additions you gain along the way: Command::checked / run_unit (0.10.2) and the record-cassette symlink/Display-injection hardening (0.10.2).

What's next

‹ docs index

ProcessKit is a Rust library today, published as processkit on crates.io. The plan is to bring the same approach — kernel-backed whole-tree containment, honest error semantics, and testable seams — to other ecosystems: a Go package, an F# library, a Kotlin library, and a Python wrapper. Each implementation will follow the same philosophy and be documented here as it ships.