# processkit cookbook

[‹ docs index](./)

Task-oriented snippets — *"I want to … → do this."* Every example assumes
`from processkit import …`. Each recipe is a quick hit; for the full treatment of
any area — every knob, the error semantics, the platform fine print — follow the
links into the [guide set](./): [Running commands](commands.md),
[Process groups](process-groups.md),
[Streaming & interactive I/O](streaming.md), [Pipelines](pipelines.md),
[Timeouts & cancellation](timeouts-and-cancellation.md),
[Supervision](supervision.md), [Testing your code](testing.md), and
[Platform support](platforms.md).

The whole library has two parallel surfaces: a **synchronous** one (plain method
names) and an **asyncio** one (the same names with an `a` prefix). Use whichever
fits your code; they share the same types and the same no-orphan guarantee.

`ProcessStdin`'s write methods and the `stdout_lines()` / `output_events()`
iterators are **async-only**. A `RunningProcess`'s *consuming* methods —
`outcome`/`aoutcome`, `finish`/`afinish`, `output`/`aoutput`,
`output_bytes`/`aoutput_bytes`, `profile`/`aprofile`, `shutdown`/`ashutdown` —
each come in a sync/async pair like everywhere else in the library: the plain
name blocks the calling thread, the `a`-prefixed twin is a coroutine (see
[Streaming](streaming.md#lifecycle) for the full table). Its `stdout_lines()` /
`output_events()` / `take_stdin()` / `kill()` are *synchronous* setup calls
(it's the iterator/handle they return that you await). A `RunningProcess` is
still usable as a **sync or async context manager** for deterministic teardown.

---

## Run a command and capture its output

A non-zero exit is *data*, not an exception:

```python
from processkit import Command

result = Command("git", ["rev-parse", "HEAD"]).output()
print(result.stdout.strip())   # the commit hash
print(result.code)             # 0
print(result.is_success)       # True
```

Async:

```python
result = await Command("git", ["rev-parse", "HEAD"]).aoutput()
```

## Require success and just get stdout

`run()` returns trimmed stdout and raises on a non-zero exit, a timeout, or a
signal-kill:

```python
commit = Command("git", ["rev-parse", "HEAD"]).run()      # or: await ....arun()
```

## Check whether a command succeeds

```python
clean = Command("git", ["diff", "--quiet"]).probe()   # True if exit 0, False if 1
code = Command("mytool").exit_code()                   # the raw exit code
```

## Accept non-zero exit codes

Some tools use non-zero as a normal result (`grep` 1 = no match, `diff` 1 =
differs). `success_codes` **replaces** the success set (default `{0}`) — list every
code you accept:

```python
differs = not Command("diff", ["a", "b"]).success_codes([0, 1]).probe()  # 0 same, 1 differs
Command("grep", ["needle", "file"]).success_codes([0, 1]).run()          # 1 (no match) is OK
```

`success_codes` affects `run()` and `result.is_success`; `exit_code()` (raw) and
`probe()` (0/1) are unchanged.

## Set a timeout

```python
result = Command("slow-tool").timeout(5.0).output()    # result.timed_out == True on expiry
Command("slow-tool").timeout(5.0).run()                # raises Timeout on expiry

# Graceful: signal, wait, then hard-kill.
Command("server").timeout(30.0).timeout_signal("term").timeout_grace(5.0).run()
```

## Pass input on stdin

```python
out = Command("tr", ["a-z", "A-Z"]).stdin_text("hello\n").run()   # "HELLO"
Command("sha256sum").stdin_bytes(b"\x00\x01\x02").run()
```

## Feed a large file to stdin without loading it into memory

```python
# Streams straight from disk to the child — no full read into Python bytes,
# so this works just as well for a multi-gigabyte dump/archive/log.
Command("psql", ["mydb"]).stdin_file("dump.sql").run()
Command("tar", ["-xf", "-"]).stdin_file("archive.tar").cwd("/tmp/extract").run()
```

## Let a child read the parent's real stdin

```python
# The child inherits *this* process's stdin — the real terminal, file, or pipe —
# instead of a crate-managed pipe. Its $EDITOR opens on the actual terminal.
Command("git", ["commit"]).inherit_stdin().run()

# Forward a shell pipeline's stdin straight through to the child:
#   cat notes.txt | python -m your_tool
Command("less").inherit_stdin().run()
```

`inherit_stdin()` is mutually exclusive with a mediated stdin source
(`stdin_bytes()` / `stdin_text()` / `stdin_file()`) or `keep_stdin_open()`;
combining them raises `ProcessError` at launch, not when you build the command.

## Set the working directory and environment

```python
Command("ls").cwd("/tmp").output()
Command("printenv", ["TOKEN"]).env("TOKEN", "secret").run()

# Set several at once, or drop an inherited one:
Command("worker").envs({"HOST": "127.0.0.1", "PORT": "8080"}).run()
Command("worker").env_remove("HTTP_PROXY").run()

# Start from an empty environment (reproducible / locked-down child), then add
# back only what you need:
Command("untrusted-tool").env_clear().env("PATH", "/usr/bin").run()
```

## Capture binary (non-UTF-8) output

`output_bytes()` returns a `BytesResult` whose `stdout` is `bytes` (stderr stays
decoded text):

```python
result = Command("convert", ["in.png", "out:-"]).output_bytes()   # or: await ....aoutput_bytes()
png = result.stdout            # bytes
print(result.code, result.is_success)
```

## Cap captured output (untrusted children)

Bound how much output is retained. To bound the parent's **memory**, cap
`max_bytes` — a `max_lines`-only cap doesn't, because one newline-free flood is a
single (unbounded) line:

```python
from processkit import Command, OutputTooLarge

# Keep only the most recent 1 MiB; older output is dropped (the default):
tail = Command("chatty-tool").output_limit(max_bytes=1024 * 1024).output()

# For an untrusted child, treat hitting the byte cap as a failure:
try:
    Command("untrusted-tool").output_limit(max_bytes=8 * 1024 * 1024, on_overflow="error").run()
except OutputTooLarge as e:
    print(e.total_bytes, e.max_bytes)
```

`on_overflow` is `"drop_oldest"` (keep most recent, the default), `"drop_newest"`
(keep earliest), or `"error"` (raise `OutputTooLarge`). A `max_lines` cap bounds
only line-captured output (raw bytes have no line count), but a `max_bytes` cap
also bounds the raw stdout of `output_bytes()` / `aoutput_bytes()` (since processkit
2.1.0): over the byte ceiling it either raises `OutputTooLarge` (`on_overflow="error"`)
or keeps a bounded head/tail with `BytesResult.truncated` set.

Under `on_overflow="error"` the ceiling (and the `total_bytes` an
`OutputTooLarge` reports) counts **raw bytes read from the pipe** — line
terminators and invalid-UTF-8 bytes included — not the bytes of the decoded text;
a drop-mode cap still bounds the retained decoded content. See
[Bounding captured output](commands.md#what-max_bytes-actually-counts).

## Stream output line by line (async)

```python
proc = await Command("my-build", ["--watch"]).astart()
async for line in proc.stdout_lines():
    print(line)
finished = await proc.afinish()   # outcome + captured stderr
```

Interleaved stdout + stderr:

```python
async for event in proc.output_events():
    print(event.stream, event.text)   # "stdout" / "stderr"
```

## Stream a log to a file and still get the captured result

`stdout_tee(path)` / `stderr_tee(path)` write the live stream to a file *and*
leave the full output in the captured result — no manual `stdout_lines()` loop,
and the one-shot verbs (`output()`, `run()`) still work:

```python
from processkit import Command

result = Command("cargo", ["build"]).stdout_tee("build.log").output()
# build.log has the live, line-by-line stream; result.stdout has the whole thing.
print(result.stdout)                 # capture is untouched — the tee is a copy
```

The file is opened **when you call the builder** (a bad path raises `OSError`
there, not at run) and truncated by default — pass `append=True` to grow an
existing log. Separate files for each stream:

```python
Command("noisy-tool").stdout_tee("out.log").stderr_tee("err.log").run()
```

The sink can be a **file path** (as above) or a **Python writer** — any object
with a `write()` method (`io.StringIO`, `sys.stderr`, a text-mode file, a
logger wrapper) — to mirror the child's output straight into your own
console, buffer, or logger while still capturing it. See
[Streaming](streaming.md#tee-output-to-a-file) for backpressure, the no-op
conditions, and write-error isolation.

## Get live progress from a synchronous run

`stdout_lines()` / `output_events()` need an event loop; `on_stdout_line(callback)`
/ `on_stderr_line(callback)` give the plain, **blocking** `.output()` / `.run()`
call the same live view — `callback` fires on every decoded line as it streams
in, not just once the run finishes:

```python
from processkit import Command

result = (
    Command("cargo", ["build", "--release"])
    .on_stdout_line(lambda line: print("build:", line))
    .output()
)
# capture is untouched — result.stdout still has the whole output.
```

Works the same on the async verbs and on a streamed run — one callback, every
path. A raising callback never derails the run (it goes to
`sys.unraisablehook` instead). See
[Streaming](streaming.md#live-per-line-callbacks) for the no-op conditions and
the one-handler-per-stream rule.

## Tear a standalone process down deterministically

A `RunningProcess` is a context manager. Exiting the block kills the process —
for a standalone `astart()` / `start()` handle that means a hard kill of its
whole private tree — even if the block raises, without waiting on Python's GC:

```python
from processkit import Command

async with await Command("flaky-server").astart() as proc:
    async for line in proc.stdout_lines():
        if "ready" in line:
            break
# proc (and its children) are reaped here

# Sync handles work too — start() is the synchronous twin of astart():
with Command("worker").start() as proc:
    ...                              # do other work
# proc torn down here
```

If you consume the handle inside the block (`proc.output()`/`.outcome()`/
`.finish()`/`.shutdown(...)`, or their `a`-prefixed async twins), exit is a
no-op.

## Talk to a process interactively (async)

```python
proc = await Command("python", ["-i"]).keep_stdin_open().astart()
stdin = proc.take_stdin()
await stdin.write_line("print(1 + 1)")
await stdin.close()                  # EOF
async for line in proc.stdout_lines():
    print(line)
await proc.aoutcome()
```

## Contain a process tree (no orphans)

Everything started in the group — and everything those processes spawn — is
reaped when the block exits:

```python
from processkit import Command, ProcessGroup

with ProcessGroup() as group:
    group.start(Command("dev-server"))
    group.start(Command("worker"))
    # ... use them ...
# the whole tree, grandchildren included, is gone here
```

Async:

```python
async with ProcessGroup() as group:
    await group.astart(Command("dev-server"))
```

## Cancel a run and reap its tree (async)

Cancelling the awaiting task — directly, or via `asyncio.wait_for` /
`asyncio.timeout` — tears the whole tree down:

```python
task = asyncio.ensure_future(Command("long-job").aoutput())
task.cancel()                        # the process tree is reaped; CancelledError propagates
```

## Wait for a server to be ready

```python
from processkit import (
    Command,
    ProcessGroup,
    wait_until,
    wait_for_path,
    wait_for_port,
    wait_for_unix_socket,
    wait_for_http,
    wait_for_line,
)

async with ProcessGroup() as group:
    proc = await group.astart(Command("my-server"))
    await wait_for_port("127.0.0.1", 8080, timeout=10)        # poll the port
    # or probe an HTTP health endpoint (ready only on a 2xx, not merely an open
    # port — a warming-up server accepts the port while still replying 503):
    # await wait_for_http("127.0.0.1", 8080, "/health", timeout=10)
    # or wait for a log line (a plain string is a substring-match shorthand):
    # await wait_for_line(proc.stdout_lines(), "listening", timeout=10)
    # or wait for a Unix socket to accept connections (stronger than a path check):
    # await wait_for_unix_socket("/run/my-server.sock", timeout=10)
    # or wait for a pid file to appear:
    # await wait_for_path("/run/my-server.pid", timeout=10)
    # or poll any (sync or async) condition:
    # await wait_until(lambda: health_check_passes(), timeout=10, interval=0.1)
```

## Wait for a unix socket or pid file to appear

Some daemons (Docker, PostgreSQL, many others) announce readiness through a
Unix-domain socket or a pid file rather than a TCP connection or log line. A
socket's filesystem entry can appear before its daemon accepts connections, so
use `wait_for_unix_socket` for the socket case; keep `wait_for_path` for a pid
file or another marker that only needs to exist:

```python
from pathlib import Path
from processkit import Command, ProcessGroup, wait_for_path, wait_for_unix_socket

socket_path = Path("/run/my-daemon.sock")
pid_path = Path("/run/my-daemon.pid")

async with ProcessGroup() as group:
    await group.astart(Command("my-daemon", ["--socket", str(socket_path)]))
    await wait_for_unix_socket(socket_path, timeout=10, interval=0.05)
    # socket_path accepts connections; a pid-file-only daemon uses:
    # await wait_for_path(pid_path, timeout=10, interval=0.05)
```

A `WaitTimeout` (also a `TimeoutError`) is raised if the path never appears
within `timeout` seconds — it carries `.path` for diagnostics.

## Build a shell-free pipeline

```python
top = (Command("ps", ["aux"]) | Command("grep", ["python"])).run()
# or: Command(...).pipe(Command(...)).run() / .arun()

# Binary tail (e.g. `... | gzip`): capture raw bytes.
blob = (Command("cat", ["big.txt"]) | Command("gzip")).output_bytes().stdout
```

A pipeline is run-to-completion (no `astart()` streaming) and has no
`output_limit` cap of its own — bound a flooding pipeline with `timeout()`. Set
per-stage `env`/`cwd` on each `Command` before piping.

## Run many commands at once

`output_all` runs a batch with bounded concurrency (default: the CPU count
available to the process, respecting affinity/cgroup quotas where the platform
reports them; fallback: `4`) and returns each result in input order. A command
that fails to *spawn* (or hits an I/O error) appears as a `ProcessError` in its
slot (a non-zero exit is still data on a `ProcessResult`):

```python
from processkit import Command, ProcessResult, output_all   # or: await aoutput_all(...)

results = output_all([Command("git", ["-C", d, "rev-parse", "HEAD"]) for d in repos],
                     concurrency=8)
heads = [r.stdout.strip() for r in results if isinstance(r, ProcessResult) and r.is_success]
```

`concurrency` bounds how many run *at once*, but every result is retained until the
whole batch returns — peak memory is the sum of all captured outputs, not just
`concurrency` of them. For a large or untrusted batch, cap each command's output
(`.output_limit(max_bytes=…)`).

For raw-bytes output use `output_all_bytes` / `aoutput_all_bytes` — the same
batch, with each slot a `BytesResult` (or a `ProcessError`).

All four accept `runner=` too, driving the whole batch through a double (see
[Test code without spawning processes](#test-code-without-spawning-processes))
instead of the real runner — no real processes spawned in a batch test.

### Stream results as they finish

`output_all` and its twins are *collect-all* — nothing is visible until the whole
batch is done. For a large fan-out where you want progress, or to react to early
finishers instead of blocking on the slowest command, `aoutput_as_completed` is
an async iterator that yields each `(index, result)` pair the moment its command
completes — in completion order, not input order, with the `index` (the command's
position in the input) re-associating a result with the command that produced it:

```python
from processkit import Command, ProcessResult, aoutput_as_completed

commands = [Command("convert", [f"{i}.png", f"{i}.jpg"]) for i in range(200)]
async for index, result in aoutput_as_completed(commands, concurrency=8):
    if isinstance(result, ProcessResult) and result.is_success:
        print(f"page {index} converted")
```

Same hard concurrency cap (never more than `concurrency` children alive at once)
and the same per-slot-error contract as the collect-all verbs — a command that
fails to spawn yields its `ProcessError` in its own pair without aborting the
stream. Cancelling the consuming task, or `break`ing out of the loop early, tears
down every command still in flight, leaving no orphaned children. Use
`aoutput_as_completed_bytes` for undecoded `bytes` output.

## Wrap a CLI tool

`CliClient` binds a program to default timeout/env, so repeated calls pass only
their args:

```python
from processkit import CliClient

git = CliClient("git", default_timeout=30.0)
head = git.run(["rev-parse", "HEAD"])            # or: await git.arun([...])
clean = git.probe(["diff", "--quiet"])
```

Modern tools (`gh`, `kubectl`, `docker`, `az`, `jj`) emit machine-readable JSON;
`run_json` / `arun_json` run like `run` (requiring a zero exit) and hand back the
already-parsed object, so you skip the `run(...)` + `json.loads(...)` +
error-mapping boilerplate:

```python
from processkit import CliClient, InvalidJson

gh = CliClient("gh")
try:
    pr = gh.run_json(["pr", "view", "42", "--json", "title,state"])
except InvalidJson as exc:
    raise SystemExit(f"{exc.program} did not return JSON: {exc.stdout[:80]!r}")
print(pr["title"], pr["state"])                  # or: await gh.arun_json([...])
```

A non-zero exit still raises `NonZeroExit` (exactly as `run` does); only a
zero-exit run whose stdout will not parse raises `InvalidJson` — a `ProcessError`
carrying the `program` and a bounded stdout fragment, never a bare
`json.JSONDecodeError`.

For testable code, pass `runner=` (a `ScriptedRunner` and friends from
`processkit.testing`) to drive every verb through a double instead of the real
runner — see [Testing your code](testing.md#wrapping-a-cli-tool-cliclient).

## Check a tool is installed before running it

Fail early with a friendly message instead of a spawn error deep in a workflow.
`resolve_program()` (or the module-level `which()`) locates the executable a run
*would* start — reusing the same `PATH`/`PATHEXT`/execute-bit lookup — without
starting any process:

```python
from processkit import CliClient, ProcessNotFound

git = CliClient("git")
try:
    git.resolve_program()
except ProcessNotFound as exc:
    raise SystemExit(f"git is required but was not found (searched: {exc.searched})")

head = git.run(["rev-parse", "HEAD"])
print(head)
```

`which("tool")` is the shorthand for a one-off check against the process `PATH`;
`Command(...).resolve_program()` and `CliClient(...).resolve_program()`
additionally honor a `prefer_local` directory and a relocated child `PATH`. All
three are synchronous and side-effect-free (a few `stat`s, no spawn), and raise
`ProcessNotFound` (also a `FileNotFoundError`) with a `searched` diagnostic on a
miss — the exact error a real run would raise.

## Keep a service alive (supervision)

```python
from processkit import Command, Supervisor

outcome = Supervisor(
    Command("flaky-worker"),
    restart="on_crash",        # "always" | "never" | "on_crash"
    max_restarts=10,
    backoff_initial=0.5,
    backoff_factor=2.0,
    max_backoff=30.0,
).run()                        # or: await ....arun()
print(outcome.restarts, outcome.stopped)
```

The `stop_when=` predicate receives each run's `ProcessResult` and returns a
bool; inspect the passed result rather than calling a synchronous run verb inside
it (a nested sync call from within the supervisor's own loop is unsupported). A
predicate that raises is reported via the unraisable hook and treated as "don't
stop".

`Supervisor` also accepts `runner=` — pass a `ScriptedRunner` with
`.on_sequence(...)` (fail a few times, then succeed) to test a restart/backoff
policy hermetically, with no real flaky process behind it.

## Sandbox an untrusted tree with resource limits

Enforced by the Windows Job Object or a Linux **cgroup-v2 root**. Under a
container / systemd session / non-root cgroup the kernel forbids them and
`ResourceLimit` is raised:

```python
from processkit import Command, ProcessGroup

# Lock down the command too: empty env (allowlisting PATH), cap output, and tie
# its lifetime to ours. All cross-platform.
tool = (
    Command("untrusted-tool")
    .env_clear().inherit_env(["PATH"])
    .kill_on_parent_death()          # die with us even without explicit teardown
    .output_limit(max_bytes=8 * 1024 * 1024)
)
with ProcessGroup(max_memory=512 * 1024 * 1024, max_processes=64, cpu_quota=1.0) as group:
    group.start(tool)
    stats = group.stats()
    print(stats.active_process_count, stats.peak_memory_bytes)
```

On POSIX you can also drop privileges to run as an unprivileged user — but set
**all three** of `gid` / `groups` / `uid` (builder order doesn't matter; the
crate applies them in the kernel-correct order, supplementary groups and gid
before uid):

```python
nobody = (
    Command("untrusted-tool")
    .gid(65534).groups([65534]).uid(65534)   # run as nobody:nogroup
)
```

Setting `uid` (and `gid`) **without** `groups([...])` leaves the child holding the
*parent's* supplementary groups — often including privileged ones (`0`/root,
`docker`, `wheel`, `sudo`) when launched from root or in CI — which is a real
sandbox escape. Always clear/replace the supplementary groups with `groups([...])`
(pass the unprivileged group, or `groups([])` to drop them entirely). These
builders make the **run raise `Unsupported` on Windows** (a privilege drop is
never silently skipped), so apply them only when targeting POSIX.

## Watch a group's resource usage live (async)

`sample_stats(group, every)` turns `group.stats()` into a periodic series — no
self-rolled polling loop:

```python
from processkit import Command, ProcessGroup, sample_stats

async with ProcessGroup(max_memory=512 * 1024 * 1024) as group:
    await group.astart(Command("untrusted-tool"))
    async for snap in sample_stats(group, every=1.0):
        print(snap.active_process_count, snap.peak_memory_bytes)
```

The series is **fused**: the first failed sample (e.g. the group has since been
torn down) ends it for good, and that failure's own exception propagates out of
the `async for` rather than the series just quietly stopping — `break` out of
the loop yourself once you have what you need.

## Signal, suspend, or resume a tree

```python
with ProcessGroup() as group:
    group.start(Command("worker"))
    group.suspend()            # pause the whole tree
    group.resume()
    group.signal("term")       # term | kill | int | hup | quit | usr1 | usr2
    group.kill_all()           # immediate hard kill
```

## Handle errors

```python
from processkit import NonZeroExit, Timeout, ProcessNotFound

try:
    Command("git", ["push"]).run()
except NonZeroExit as e:
    print(e.code, e.stderr)    # structured fields, not just a message
except Timeout as e:
    print(e.timeout_seconds)
except ProcessNotFound as e:
    print("missing:", e.program)
```

Every exception derives from `ProcessError`. Three also derive from the builtin
the stdlib raises for the same condition, so familiar `except` clauses work:
`Timeout` is also a `TimeoutError` (as `asyncio.TimeoutError` is),
`ProcessNotFound` is also a `FileNotFoundError` (as `subprocess` raises), and
`PermissionDenied` is also a `PermissionError`. The async readiness helpers
(`wait_for_port` / `wait_for_http` / `wait_for_line` / `wait_for_path` /
`wait_for_unix_socket` / `wait_until`) raise
builtin `TimeoutError`, so `except TimeoutError` catches both run and
readiness timeouts.

## Test code without spawning processes

Write your code against a runner, then inject a `ScriptedRunner` in tests. The
doubles live in the `processkit.testing` submodule; `Runner` is top-level:

```python
from processkit import Command, Runner
from processkit.testing import Reply, ScriptedRunner

def latest_commit(runner):
    return runner.run(Command("git", ["rev-parse", "HEAD"]))

# production
latest_commit(Runner())

# test
scripted = ScriptedRunner()
scripted.on(["git", "rev-parse"], Reply.ok("deadbeef"))
assert latest_commit(scripted) == "deadbeef"
```

`Reply.ok` / `.fail` / `.timeout` / `.signalled` / `.lines` / `.pending` cover
the outcomes; `ScriptedRunner.start()` even returns a streamable scripted
`RunningProcess`. `.on_sequence(prefix, replies)` scripts a *sequence* of
replies for successive matching calls (fail once, then succeed — the shape a
retry/supervision test needs), repeating the last reply once exhausted.

`output_all`/`aoutput_all` (and their `_bytes` twins), `Supervisor`, and
`CliClient` all accept the same doubles via a `runner=` keyword, so batches,
supervised commands, and CLI wrappers are just as testable as raw `Command`
code — see [Testing your code](testing.md) for the full picture.

To capture *real* tool output once and replay it deterministically offline, use
`RecordReplayRunner` — both share the `Runner` verb surface:

```python
from processkit.testing import RecordReplayRunner

rec = RecordReplayRunner.record("cassette.json")   # records via the real runner
recorded = latest_commit(rec)                       # spawns git once, captures it
rec.save()

rep = RecordReplayRunner.replay("cassette.json")   # offline; no process spawned
assert latest_commit(rep) == recorded
```

To assert on *what* your code ran (not just its output), inject a
`RecordingRunner` spy — it replies uniformly and records every call:

```python
from processkit import Command
from processkit.testing import RecordingRunner, Reply

def deploy(runner):
    runner.run(Command("git", ["push", "--tags"]))

spy = RecordingRunner.replying(Reply.ok(""))
deploy(spy)

inv = spy.only_call()                 # the one call (raises unless exactly one)
assert inv.program == "git"
assert inv.args == ["push", "--tags"]
```

For a `--dry-run`/`--echo` mode — assert on (or print) the *rendered command
line*, with no reply to script and no output to replay — inject a
`DryRunRunner`. It never spawns, renders each command to its display-quoted
line, and returns a synthetic success:

```python
from processkit import Command
from processkit.testing import DryRunRunner

def prune(runner):
    runner.run(Command("rm", ["-rf", "build"]))

dry = DryRunRunner()
prune(dry)
assert dry.only_command() == "rm -rf build"   # nothing spawned
# dry.on_invocation(print) would echo each line live instead.
```

## Use the pytest fixtures

Installing processkit registers a pytest plugin (via a `pytest11` entry point) —
nothing to add to `conftest.py`. It hands you the doubles as fixtures, so
injecting one is a single parameter:

```python
from processkit import Command
from processkit.testing import Reply

def latest_commit(runner):
    return runner.run(Command("git", ["rev-parse", "HEAD"]))

def test_latest_commit(scripted_runner):        # fixture: a fresh ScriptedRunner
    scripted_runner.on(["git", "rev-parse"], Reply.ok("deadbeef"))
    assert latest_commit(scripted_runner) == "deadbeef"

def test_deploy_pushes_tags(recording_runner):  # fixture: a RecordingRunner spy
    recording_runner.run(Command("git", ["push", "--tags"]))
    assert recording_runner.only_call().args == ["push", "--tags"]
```

The `record_replay_runner` fixture serves a per-test cassette — replay by
default, record with `pytest --processkit-record` (or the `PROCESSKIT_RECORD`
env var / `processkit_record` ini). Point `processkit_cassette_dir` (ini) at a
committed fixtures directory to keep cassettes. Mark a test
`@pytest.mark.no_real_spawn` to make any real spawn inside it fail loudly. Full
details in [Testing your code](testing.md#the-pytest-plugin-ready-made-fixtures).

## See what processkit runs (logging)

Opt in once with `enable_logging()` and `processkit` forwards its internal run
events to Python's `logging`:

```python
import logging
from processkit import Command, enable_logging

logging.basicConfig(level=logging.DEBUG)
enable_logging()        # idempotent; returns False if another library already
                        # owns the process-global tracing subscriber

Command("git", ["rev-parse", "HEAD"]).run()
# DEBUG:processkit:child spawned program=git pid=Some(12345) mechanism=…
# DEBUG:processkit:process exited program=git outcome=Exited(0) elapsed_ms=7
```

(`mechanism` is the platform's containment — `JobObject` on Windows, a process
group / cgroup on POSIX. Fields are forwarded verbatim, so `pid` shows the core's
`Some(…)` rendering.)

Records land on the `processkit` logger (filter it like any other) — DEBUG for a
normal run, WARNING for an edge case. **argv and env are never logged** (the core
omits them — they routinely carry secrets). It's a deliberate **opt-in**: enabling
it installs a process-global subscriber and adds a little per-run overhead, so it's
a debugging/observability switch, off by default.

---

# Coming from `subprocess`

[‹ docs index](./)

You already know `subprocess` (or `asyncio.subprocess`). This guide maps the
patterns you write today onto their `processkit` equivalents, so porting existing
code is mechanical — and then shows the one thing the stdlib can't do that is the
reason to switch: **containing the whole process tree**.

Every snippet assumes `from processkit import ...`. For the full treatment of any
verb, follow the links into [Running commands](commands.md).

## The mental-model shift

`subprocess` couples *running* a command with *deciding whether it failed*:
`run(...)` gives you a `returncode` to inspect, `run(..., check=True)` raises. In
`processkit` those are two different verbs:

- `Command(...).output()` **captures** the result — a non-zero exit, a timeout, and
  a signal-kill are all **data** on a `ProcessResult`, never an exception.
- `Command(...).run()` **requires success** — it returns trimmed stdout and raises a
  typed exception on a non-zero exit, a timeout, or a signal-kill.

Pick the verb by what you want; you no longer thread a `check=` flag through.
See [Picking a verb](commands.md#picking-a-verb) for the full set.

## Running a command (sync)

| You wrote (`subprocess`) | Now write (`processkit`) |
|---|---|
| `run(cmd, capture_output=True, text=True)` → inspect `.returncode` / `.stdout` | `Command(prog, args).output()` → `ProcessResult` (`.code`, `.stdout`, `.is_success`, `.timed_out`) |
| `run(cmd, capture_output=True, text=True, check=True).stdout` | `Command(prog, args).run()` (returns **trimmed** stdout, raises on failure) |
| `run(cmd).returncode` | `Command(prog, args).exit_code()` (raw code) |
| `run(cmd).returncode == 0` | `Command(prog, args).output().is_success` (total); `.probe()` is a shortcut for `0`/`1`-exit predicate tools |
| `run(cmd, capture_output=True).stdout` (bytes) | `Command(prog, args).output_bytes()` → `BytesResult` (`.stdout` is `bytes`) |

```python
from processkit import Command

# subprocess: subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True)
result = Command("git", ["rev-parse", "HEAD"]).output()
print(result.stdout.strip(), result.code, result.is_success)

# subprocess: subprocess.run([...], check=True, capture_output=True, text=True).stdout
commit = Command("git", ["rev-parse", "HEAD"]).run()   # trimmed stdout, raises on failure
```

Note the two differences from `run()` in `subprocess`: `.output().stdout` is the
**full** captured text (not stripped — strip it yourself), while `.run()` returns
it **trimmed**; and a non-zero exit is only an error for `.run()`, never for
`.output()`.

One more divergence to know: unlike `subprocess`'s numeric `.returncode`, the
*checking* verbs (`exit_code`, `probe`, `run`) **raise** on a timeout or a
signal-kill instead of returning a code (and `probe()` also raises on any exit code
other than `0`/`1`). Reach for `.output()` when you want an abnormal exit as
inspectable data (`.timed_out`, `.signal`) rather than an exception.

## The common flags

| `subprocess` keyword | `processkit` builder |
|---|---|
| `timeout=5` | `.timeout(5.0)` — captured on `.output()` (`result.timed_out`), raised by `.run()` |
| `input="text"` / `input=b"..."` | `.stdin_text("text")` / `.stdin_bytes(b"...")` |
| `cwd="/path"` | `.cwd("/path")` |
| `env={...}` (**replaces** the whole environment) | `.env_clear().envs({...})` |
| add/override one variable on the inherited env | `.env("KEY", "value")` / `.envs({...})` |
| — (no equivalent) | `.success_codes([0, 1])` — **replaces** the success set with the listed codes (`grep`/`diff`) |

```python
# subprocess: subprocess.run(["slow"], timeout=5) -> raises TimeoutExpired
Command("slow").timeout(5.0).run()                       # raises Timeout on expiry
result = Command("slow").timeout(5.0).output()           # result.timed_out is True instead

# subprocess: subprocess.run(["tr","a-z","A-Z"], input="hello\n", text=True)
Command("tr", ["a-z", "A-Z"]).stdin_text("hello\n").run()

# subprocess: subprocess.run(["grep","x","f"], check=True)  # exit 1 = "no match" -> would raise
Command("grep", ["x", "f"]).success_codes([0, 1]).run()  # 1 (no match) is not a failure
```

`env=` in `subprocess` **replaces** the entire environment; the direct equivalent
is `.env_clear().envs({...})`. To *add to* the inherited environment (the more
common intent), use `.env(...)` / `.envs(...)` without `env_clear()`. More in
[Environment and sandboxing](commands.md#environment-and-sandboxing).

## Shell pipelines, without the shell

`subprocess` pipelines usually mean `shell=True` (and a shell-injection footgun) or
hand-wiring two `Popen`s. `processkit` pipes are shell-free:

```python
# subprocess: subprocess.run("ps aux | grep python", shell=True)
from processkit import Command

out = (Command("ps", ["aux"]) | Command("grep", ["python"])).run()
```

See [Pipelines](pipelines.md) for pipefail attribution and binary tails.

## Async

If you reach for `asyncio.subprocess`, every verb has an `a`-prefixed twin that
shares the same types:

```python
# asyncio: proc = await asyncio.create_subprocess_exec("git","status", stdout=PIPE)
#          out, _ = await proc.communicate()
result = await Command("git", ["status", "--short"]).aoutput()

# Streaming stdout line by line (asyncio-native):
proc = await Command("my-build", ["--watch"]).astart()
async for line in proc.stdout_lines():
    print(line)
finished = await proc.afinish()
```

Streaming, interactive stdin, and readiness probes are covered in
[Streaming & interactive I/O](streaming.md).

## Exceptions

The exception hierarchy is independent, but the three that mirror a stdlib builtin
also *subclass* it — so your existing `except` clauses keep working:

| `subprocess` raises | `processkit` raises | Also a subclass of |
|---|---|---|
| `CalledProcessError` (from `check=True`) | `NonZeroExit` (`.code`, `.stderr`) | — |
| `TimeoutExpired` | `Timeout` (`.timeout_seconds`) | `TimeoutError` |
| `FileNotFoundError` (missing program) | `ProcessNotFound` (`.program`) | `FileNotFoundError` |
| `PermissionError` | `PermissionDenied` (`.program`, `str \| None` — `None` for a program-less permission-denied OS error, not just a spawn-time denial) | `PermissionError` |

```python
# This subprocess-style handler keeps working, because ProcessNotFound *is* a
# FileNotFoundError and Timeout *is* a TimeoutError:
from processkit import Command

try:
    Command("mytool").timeout(5.0).run()
except FileNotFoundError:
    print("not installed")
except TimeoutError:
    print("timed out")
```

Every exception derives from `ProcessError`; see [Errors](commands.md#errors).

## What you actually gain: containing the tree

Everything above is convenience — the *reason* to switch is that `subprocess` and
`asyncio.subprocess` reach only the **direct child**. The processes *it* spawns (a
build tool's compilers, the real payload behind a `sh -c` wrapper, a test's helper
servers) survive a timeout, an exception, or a cancelled task and keep running as
orphans. `processkit` spawns every child into the operating system's own
containment primitive, so teardown is one kernel operation over the whole tree:

```python
from processkit import Command, ProcessGroup

with ProcessGroup() as group:
    group.start(Command("dev-server"))
    group.start(Command("worker"))
    # ... use them ...
# leaving the block reaps the whole tree — grandchildren included
```

Even a single one-shot verb gets this for free: `Command(...).output()` runs inside
a private group that dies with the call, and cancelling an awaited `aoutput()`
reaps its tree. On top of the guarantee you also get whole-tree **resource limits**
(memory / process-count / CPU caps) for sandboxing untrusted children — something
`subprocess` cannot express at all. See [Process groups](process-groups.md) and
[Resource limits](process-groups.md#resource-limits-the-sandbox).

## When to stay with `subprocess`

`processkit` earns its place when you run process *trees*, need them reaped
reliably, or want resource-limited sandboxes. If you only ever run leaf commands
that never spawn children of their own, don't need async cancellation to be
leak-safe, and want zero third-party dependencies, the stdlib is a perfectly good
choice — `processkit` is deliberately **not** a general `subprocess`-convenience
replacement. The wedge is the no-orphan guarantee.

---

Next: [Running commands](commands.md) · [Cookbook](cookbook.md)

---

# Running commands

[‹ docs index](./)

`Command` is the entry point of the runner layer: a builder that describes *what*
to run and *how*, plus a family of verbs that decide *what you get back*. Every
one-shot verb spawns the child into a fresh, private, kill-on-exit process tree,
so an early return, an exception, or a cancelled task can never leak a child.

- [The two surfaces: sync and async](#the-two-surfaces-sync-and-async)
- [Picking a verb](#picking-a-verb)
- [Program, arguments, working directory](#program-arguments-working-directory)
- [Local program search](#local-program-search)
- [Environment and sandboxing](#environment-and-sandboxing)
- [Standard input](#standard-input)
- [Redirecting stdout and stderr](#redirecting-stdout-and-stderr)
- [Text decoding](#text-decoding)
- [Bounding captured output](#bounding-captured-output)
- [Timeouts](#timeouts)
- [Privileges and spawn flags](#privileges-and-spawn-flags)
- [Results](#results)
- [Errors](#errors)
- [Pipelines](#pipelines)

## The two surfaces: sync and async

The capture verbs come in two flavors: a **synchronous** one with a plain name,
and an **asyncio** one with the same name under an `a` prefix. They share the same
builder, the same result types, and the same no-orphan guarantee — pick whichever
fits the call site. (`start()` / `astart()` hand back a live `RunningProcess` for
streaming and interactive I/O — see [Streaming & interactive I/O](streaming.md).
That handle's *consuming* verbs (`outcome`/`aoutcome`, `finish`/`afinish`,
`output`/`aoutput`, …) come in sync/async pairs too, like everywhere else in
this library — use whichever matches your code, regardless of whether the
handle came from `start()` or `astart()`.)

```python
from processkit import Command

head = Command("git", ["rev-parse", "HEAD"]).run()          # sync
head = await Command("git", ["rev-parse", "HEAD"]).arun()    # asyncio
```

The rest of this guide shows the sync form and only repeats the async form where
the behavior differs. A blocked synchronous call **on the main thread** is
interruptible by Ctrl+C: it raises `KeyboardInterrupt` and reaps the process tree
on the way out (off the main thread CPython can't deliver the signal — use the
async API or a `timeout()` there). *Deeper:
[Timeouts & cancellation](timeouts-and-cancellation.md).*

## Picking a verb

| Verb | Returns | Non-zero exit | Timeout / signal-kill | Use when |
|---|---|---|---|---|
| `output()` | `ProcessResult` | captured (`.code`) | captured (`.timed_out` / `.signal`) | You want to inspect the outcome yourself |
| `output_bytes()` | `BytesResult` | captured | captured | stdout is binary (images, archives) |
| `run()` | trimmed stdout `str` | raises `NonZeroExit` | raises `Timeout` / `Signalled` | "Give me the answer, or fail" |
| `exit_code()` | `int` (raw) | returns the code | raises (no `-1` sentinel) | The exit code *is* the answer |
| `probe()` | `bool` | `0`→`True`, `1`→`False`, else raises | raises | Predicate tools: `git diff --quiet`, `grep -q` |
| `start()` / `astart()` | `RunningProcess` | — | — | Streaming / interactive I/O — see [Streaming](streaming.md) |

The capturing verbs (`output`, `output_bytes`) treat a non-zero exit, a timeout,
and a signal-kill as **data** — they never raise on the child's outcome. The
checking verbs (`run`, `exit_code`, `probe`) turn those into exceptions. Async
twins: `aoutput`, `aoutput_bytes`, `arun`, `aexit_code`, `aprobe`, `astart`.

```python
result = Command("git", ["merge", "feature"]).output()
print(result.code, result.is_success, result.stdout)   # nothing raised
```

## Program, arguments, working directory

Arguments are a list — there is **no shell** between you and the child, so no
quoting, no word-splitting, and no injection surface. Build them up one at a time
or in bulk; `cwd` sets the working directory. The program, the arguments, and
`cwd` accept a `str` or any `os.PathLike[str]` (e.g. `pathlib.Path`) — so a `Path`
argument needs no `str()`. (`bytes` paths are not accepted.)

```python
from pathlib import Path

out = (
    Command("git")
    .arg("log")                       # one at a time...
    .args(["--oneline", "-n", "10"])  # ...or in bulk
    .cwd(Path("/srv/repo"))           # run there
    .run()
)
```

The program name reaches the OS verbatim: a bare name is resolved on `PATH` by
the OS, and `cwd` does **not** re-anchor a *relative* program path against the new
directory. Pass an absolute program path when you combine a relative tool with a
`cwd`.

Read back what you built with the `program` / `arguments` properties (`arguments`,
not `args` — that name is already the builder method that *appends* args), or
render the whole thing as a single shell-quoted line with `command_line()` — for
display only (logs, error messages, a dry-run echo): it never invokes a shell,
and the escaping targets human legibility, not any shell's actual parsing rules.
Unlike the redacted `repr()`, `command_line()` **does** include argv, so render it
only into a sink you control.

```python
cmd = Command("login", ["--password", "hunter2"])
cmd.program              # "login"
cmd.arguments            # ["--password", "hunter2"]
cmd.command_line()       # "login --password hunter2" — includes the secret!
repr(cmd)                # redacted: shows arg COUNT, never values
```

## Local program search

Use `prefer_local(dir)` when a bare-name program should resolve from a project or
toolchain directory before falling back to the system `PATH`: for example
`node_modules/.bin`, `target/debug`, or a vendored tool directory. The directory
argument accepts `str` and `os.PathLike[str]`, like `cwd`.

```python
out = (
    Command("ruff")
    .prefer_local(Path(".venv/bin"))
    .prefer_local(Path("tools/bin"))
    .arg("--version")
    .run()
)
```

Repeated calls accumulate in priority order, so the first preferred directory is
searched first, then the next, then the normal `PATH`. The search reuses the same
platform behavior as `PATH` resolution, including `PATHEXT` on Windows.

`prefer_local` affects only bare-name programs such as `"ruff"` or `"cargo"`.
Path-form programs such as `"./ruff"`, `"tools/ruff"`, or an absolute path are
used as written. It also does not rewrite the child's own `PATH`; it only changes
how processkit finds the executable to spawn. If the program is not found, the
preferred directories are included in the failure diagnostics along with the
normal search locations.

Here is a self-contained example that creates two local tool directories and
prefers both before the system `PATH`:

```python
import os
import stat
import tempfile
from pathlib import Path

from processkit import Command

name = "demo-tool"


def write_tool(directory: Path, text: str) -> Path:
    directory.mkdir(parents=True)
    if os.name == "nt":
        tool = directory / f"{name}.cmd"
        tool.write_text(f"@echo off\necho {text}\n", encoding="utf-8")
    else:
        tool = directory / name
        tool.write_text(f"#!/bin/sh\necho {text}\n", encoding="utf-8")
        tool.chmod(tool.stat().st_mode | stat.S_IXUSR)
    return tool


with tempfile.TemporaryDirectory() as tmp:
    project = Path(tmp)
    first = write_tool(project / "node_modules" / ".bin", "node tool")
    second = write_tool(project / "target" / "debug", "debug tool")

    # Search order for the bare name "demo-tool":
    #   1. ./node_modules/.bin
    #   2. ./target/debug
    #   3. the parent process PATH
    out = (
        Command(name)
        .cwd(project)
        .prefer_local(first.parent)
        .prefer_local(second.parent)
        .run()
    )
    assert out == "node tool"

    # The child still receives the inherited PATH unless you change it with
    # env(...). prefer_local only affects processkit's spawn-time lookup.
    assert str(first.parent) not in os.environ.get("PATH", "")

    # Path-form programs bypass prefer_local and are used exactly as written.
    assert Command(second).prefer_local(first.parent).run() == "debug tool"
    old_cwd = Path.cwd()
    os.chdir(project)
    try:
        assert (
            Command(f"target{os.sep}debug{os.sep}{second.name}")
            .prefer_local(first.parent)
            .run()
            == "debug tool"
        )
    finally:
        os.chdir(old_cwd)

    print(out)
```

## Preflight: is a program installed?

Sometimes you want to check that a tool is present *before* you run it — a
"doctor" subcommand, or a friendlier error than a spawn failure surfacing deep
in a workflow. `resolve_program()` locates the executable a run *would* spawn,
**without starting any process**:

```python
from processkit import Command, ProcessNotFound

try:
    path = Command("ruff").resolve_program()
    print(f"ruff is installed at {path}")
except ProcessNotFound as exc:
    print(f"ruff is not installed (searched: {exc.searched})")
```

The lookup reuses the **same** resolution the real launch performs — a bare name
against any `prefer_local()` directories first, then `PATH`, honoring `PATHEXT`
on Windows and the execute bit on Unix; a path-form program (`"./tool"`, an
absolute path) probed directly. So a hit is exactly what a spawn of the same
command would run, and a miss is exactly the `ProcessNotFound` it would raise —
`searched` diagnostic included. It also honors a relocated child `PATH`
(`env()` / `env_clear()` / `inherit_env()`), so the preflight never disagrees
with the actual spawn. It is synchronous and cheap (a few `stat`s); there is no
`a`-prefixed async twin, because no runtime is involved.

For a one-off check against the process `PATH`, the module-level `which()` is
shorthand for `Command(program).resolve_program()`:

```python
import processkit

interpreter = processkit.which("python3")   # absolute path, or raises ProcessNotFound
print(interpreter)
```

A `CliClient` offers the same preflight for the tool it wraps, with the client's
defaults (including a `default_env` that relocates `PATH`) applied:

```python
from processkit import CliClient, ProcessNotFound

client = CliClient("git")
try:
    client.resolve_program()   # is git installed, per this client's config?
except ProcessNotFound:
    raise SystemExit("git is required but was not found")
```

## Environment and sandboxing

The environment builders compose, applied in a fixed order at spawn:

```python
# Mutate the inherited environment.
Command("worker").env("RUST_LOG", "debug").env_remove("HTTP_PROXY").run()
Command("worker").envs({"HOST": "127.0.0.1", "PORT": "8080"}).run()

# Allow-list: clear everything, then copy only the named parent variables.
Command("sandboxed-tool").inherit_env(["PATH", "HOME", "LANG"]).env("MODE", "ci").run()

# Scorched earth: the child starts with an empty environment.
Command("hermetic-tool").env_clear().env("PATH", "/usr/bin").run()
```

`inherit_env` is the sandboxing middle ground: it implies `env_clear`, then copies
the listed variables *from the parent at each spawn* (a re-run sees fresh values),
and repeated calls accumulate names. A name the parent doesn't have is skipped,
not set to empty. Explicit `env` / `env_remove` still apply on top.

## Standard input

By default stdin is **closed at spawn** — the child reads EOF immediately and can
never hang waiting for input. Feed a one-shot payload with `stdin_text` (a `str`)
or `stdin_bytes` (raw `bytes`):

```python
loud = Command("tr", ["a-z", "A-Z"]).stdin_text("hello\n").run()   # "HELLO"
Command("sha256sum").stdin_bytes(b"\x00\x01\x02").run()
```

The payload is written on a background task, so a large input can't deadlock
against the child's own output; the pipe is closed afterward to signal EOF.

For a large input already sitting in a file — a database dump piped into `psql`,
an archive fed to `tar`, a multi-gigabyte log run through a filter — use
`stdin_file(path)` instead of reading the file into Python `bytes` yourself. The
file streams straight to the child's stdin in chunks, so it never has to fit in
Python memory:

```python
Command("psql", ["mydb"]).stdin_file("dump.sql").run()
Command("tar", ["-xf", "-"]).stdin_file("archive.tar").cwd("/tmp/extract").run()
```

`stdin_file()` doesn't touch the filesystem when you call it — the path is
opened lazily when the command actually spawns, so a not-yet-existing path is
not an error there. If the file turns out to be missing or unreadable once the
command runs, that surfaces as a generic `ProcessError` from the run/output
verb (not `FileNotFoundError`), since the child process has, by then, already
spawned successfully.

For a conversational, request/response exchange — write a line, read the answer,
repeat — call `keep_stdin_open()` and drive the process through the streaming API
instead. *Deeper: [Streaming & interactive I/O](streaming.md).*

To let the child read the parent's **own** stdin directly — the real terminal, a
file, or a pipe this process was launched with — call `inherit_stdin()`. It is
the stdin counterpart of `stdout("inherit")`: the child *shares* the parent's
stream instead of the crate mediating it. Use it when the child must reach the
real terminal — `git commit` opening `$EDITOR`, a tool prompting for a password
or a yes/no confirmation, or forwarding a shell pipeline's stdin straight
through:

```python
# The editor opens on the real terminal; the crate doesn't touch stdin.
Command("git", ["commit"]).inherit_stdin().run()
```

The crate neither feeds nor captures that input, so there is no writer to
`take_stdin()` — but stdout/stderr are untouched, so `run()` / `output()` still
return the child's captured stdout as usual. `inherit_stdin()` is **mutually
exclusive** with any *mediated* stdin: a `stdin_bytes()` / `stdin_text()` /
`stdin_file()` source, or `keep_stdin_open()`. A child either reads the parent's
stdin or has its stdin driven by the crate, not both. Building the conflicting
combination does not raise; the contradiction is rejected as a `ProcessError`
from the run/output verb when the command actually **launches**, not when you
build the `Command` (the same guard fires on the test doubles too — see
[Interactive stdin](streaming.md#interactive-stdin)).

## Redirecting stdout and stderr

Each stream defaults to `"pipe"` (captured). You can also `"inherit"` the
parent's stream or send it to `"null"`:

```python
Command("long-build").stdout("inherit").stderr("inherit").start()
```

This matters: the one-shot capturing verbs (`output`, `output_bytes`, `run`,
`exit_code`, `probe`) need a piped stdout to do their job. If you set stdout to
`"inherit"` or `"null"`, those verbs **raise** — only `start()` / `astart()` plus
streaming work with a non-piped stdout, because there is nothing to capture. Redirect
streams only when you intend to stream or to discard.

### Redirecting a stream straight to a file

To send a stream *directly* to a file — the child writes to the file's own
descriptor, with no parent-side pump or capture in between — use `stdout_file()`
/ `stderr_file()`. This is the direct-redirect cousin of `stdout_tee()`: the tee
*also* captures and mirrors every decoded line, while these simply hand the child
the file (a `>` / `>>` shell redirect, minus the shell). `append=False` (the
default) creates or **truncates** the file on each spawn; `append=True` creates
or **appends** — the mode for a shared log across `Supervisor` incarnations or
`retry()` attempts, which write to one file with no separator.

```python
from processkit import Command, Supervisor

# Truncate a fresh file on each spawn (the default).
with Command("build", ["--all"]).stdout_file("build.log").start() as proc:
    proc.outcome()

# Append across restarts — one shared log for every Supervisor incarnation.
Supervisor(
    Command("worker").stdout_file("worker.log", append=True).stderr_file("worker.log", append=True)
).run()
```

Unlike `stdout_tee()`, the file is opened **at spawn time**, not when you build
the command — so a not-yet-existing path is not an error here, and each re-run or
retry reopens it. An unopenable path (a missing parent directory, a permission
denial) surfaces from the run verb when the command launches.

Because a file-redirected **stdout** has no pipe for the parent to read, the
verbs that actually read stdout back — `output()`, `run()`, `output_bytes()`
(and their `a`-twins), plus `start()` + `stdout_lines()` / `output_events()` —
**raise** the same "not piped" `ProcessError` as `stdout("null")`. `exit_code()`
and `probe()` are *not* capture verbs: they discard output entirely and never
touch the stdout pipe, so they work fine with a file-redirected stdout — they
(and their async twins `aexit_code()` / `aprobe()`) are the recommended way to
drive such a command to completion, alongside `start()` + `outcome()` /
`aoutcome()` (as the example above does). A file-redirected **stderr** leaves
stdout piped, so `output()` keeps working there; the child's stderr just lands
in the file and `result.stderr` comes back empty. A later `stdout(...)` /
`stderr(...)` call **clears** the redirect and restores the normal stdio mode,
so the builder chain stays composable.

## Text decoding

Output is decoded line by line, UTF-8 by default; invalid bytes become `U+FFFD`
rather than raising. Legacy-encoding tools can override per stream. Labels are
**WHATWG encoding labels** (as the web platform uses) — e.g. `"iso-8859-1"`,
`"windows-1252"`, `"windows-1251"`, `"shift_jis"`. Common **Python codec
aliases** are accepted too (`"latin_1"`, `"utf_8"`, `"euc_jp"`, …), normalized to
the WHATWG form. One caveat to know: WHATWG's `"iso-8859-1"` (and the Python
`"latin_1"` that maps to it) decodes as **windows-1252**, which differs from
strict ISO-8859-1 only in the `0x80`–`0x9F` range. The Windows ANSI code page
(`"mbcs"`/`"ansi"`) has no portable label — pass it explicitly (e.g.
`"windows-1251"`). An unmappable label raises `ValueError` naming the WHATWG form.

```python
out = Command("legacy-tool").encoding("shift_jis").output()        # both streams
out = Command("tool").stdout_encoding("iso-8859-1").output()       # ...or each its own
# .stderr_encoding(...) sets stderr independently
```

When stdout is genuinely binary, skip decoding entirely with `output_bytes()`
(below) instead of guessing an encoding.

## Bounding captured output

Captured lines are held in memory; a multi-gigabyte log would grow the buffer to
match. `output_limit` bounds *retention* — the pipe is always fully drained, so
the child never blocks on a full buffer.

```python
from processkit import Command, OutputTooLarge

# Keep only the most recent 1 MiB; older output is dropped (the default).
tail = Command("chatty-tool").output_limit(max_bytes=1024 * 1024).output()

# For an untrusted child, treat hitting the cap as a failure.
try:
    Command("untrusted-tool").output_limit(
        max_bytes=8 * 1024 * 1024, on_overflow="error"
    ).run()
except OutputTooLarge as e:
    print(e.total_bytes, e.max_bytes)
```

`on_overflow` is `"drop_oldest"` (keep the newest, the default), `"drop_newest"`
(freeze the head), or `"error"` (raise `OutputTooLarge`). To bound the parent's
**memory** against an untrusted child, cap `max_bytes`: a `max_lines`-only cap
does *not*, because one newline-free flood is a single, unbounded line. A
`max_lines` cap applies to line-captured output only — raw bytes have no line
count, so it never bounds the stdout of `output_bytes()`. A `max_bytes` cap
applies to *both* that line-captured output **and** the raw stdout of
`output_bytes()` / `aoutput_bytes()` (since processkit 2.1.0 — earlier the byte
ceiling bounded only the line-pumped stderr and raw stdout was always unbounded).
Over the byte cap an `output_bytes()` run either raises `OutputTooLarge` (with
`max_lines=None`) under `on_overflow="error"`, or keeps a bounded head/tail with
`BytesResult.truncated` set under a drop mode.

### What `max_bytes` actually counts

Which bytes the cap counts depends on `on_overflow` — the asymmetry is deliberate
upstream, so check which half applies to you before sizing a cap:

- **`on_overflow="error"`** — the ceiling counts the **raw bytes read from the
  child's output pipe**, before decoding and *cumulatively* over the whole run
  (a streaming consumer draining lines frees buffer space but does not reset it).
  The line terminator each line arrived with (`\n`, or *both* bytes of a `\r\n`)
  and bytes that are **not valid UTF-8** are charged against it, even though
  neither survives into `ProcessResult.stdout`. `OutputTooLarge.total_bytes`
  reports that same raw count, so it can exceed `len(result.stdout.encode())` for
  the same output. **Changed in processkit 3.0.0**: this ceiling used to count
  decoded line content, so a cap sized against decoded text now trips slightly
  sooner — by one byte per line for ordinary UTF-8 output, more for CRLF or
  binary-ish output.
- **`on_overflow="drop_oldest"` / `"drop_newest"`** — the cap bounds what is
  **retained**, measured in the bytes of the decoded line *content*, terminators
  excluded. Unchanged in 3.0.0: a drop-mode cap keeps exactly the head/tail it
  always did.

Either way `max_bytes` is a real bound on the parent's memory — the reason to
prefer it over `max_lines` for an untrusted child — the two modes just measure at
different points: what was *read* for the fail-loud ceiling, what is *kept* for
the ring buffers.

Raw stdout captured by `output_bytes()` is never decoded, so there is no
distinction to draw there: its byte cap counts the bytes as they were read, in
every mode, exactly as it did before 3.0.0.

## Timeouts

```python
result = Command("slow-tool").timeout(5.0).output()   # result.timed_out is True on expiry
Command("slow-tool").timeout(5.0).run()               # raises Timeout on expiry

# Graceful shutdown: send a signal, wait, then hard-kill.
Command("server").timeout(30.0).timeout_signal("term").timeout_grace(5.0).run()
```

Durations are floats of seconds — never a duration object. `timeout` kills the
whole process tree at the deadline; on the capturing verbs the expiry is captured
(`ProcessResult.timed_out`), on the checking verbs it raises `Timeout`. The
signal name in `timeout_signal` is one of `term | kill | int | hup | quit | usr1
| usr2`, or a raw platform signal number (an `int`, POSIX only — Windows raises
`Unsupported` for anything but a hard kill, same as the named variants).
*Deeper: [Timeouts & cancellation](timeouts-and-cancellation.md).*

`no_timeout()` runs without a deadline, and — unlike simply never calling
`timeout()` — also opts out of a `CliClient`'s `default_timeout` gap-fill
(useful for the one deliberately unbounded call — a `tail -f`, a watch loop —
against a client that otherwise imposes a deadline on every call). Whichever
of `timeout()` / `no_timeout()` you call **last** wins.

### `idle_timeout` — a silence watchdog

`idle_timeout(seconds)` bounds a *silent gap* rather than total runtime: it kills
the child if it emits no stdout/stderr **line** for that long — for a tool that
hangs silently while a healthy long job keeps printing. It **composes** with
`timeout()` (whichever threshold is reached first wins) and validates like it
(finite, `> 0`).

```python
from processkit import Command, IdleTimeout

proc = Command("./flaky-build").timeout(600.0).idle_timeout(30.0).start()
try:
    async with proc:
        async for event in proc.output_events():
            print(event.text)
except IdleTimeout as e:
    print(f"silent for {e.idle_timeout_seconds}s — killed")
```

It fires as a **distinct** `IdleTimeout` (a `ProcessError` sibling of `Timeout`,
carrying `idle_timeout_seconds`), never the wall-clock `timed_out`/`Timeout`, so
the two timeout classes stay tellable apart. **Boundaries:** idle monitoring
rides the per-line output channel, so it is enforced only on the
streaming/interactive surface (`start()`/`astart()` +
`stdout_lines()`/`output_events()`); the one-shot capture verbs, `Pipeline`, and
`Supervisor` do not enforce it (processkit's core has no native idle-timeout —
that awaits upstream support). A redirected stdout (`stdout_file`/`inherit`/
`null`) carries no line events, so the streaming verbs raise the usual "stdout is
not piped" `ProcessError` there — the combination is diagnosed, not silently
un-watched; redirect only stderr if you still want stdout watched.
*Deeper: [Timeouts & cancellation](timeouts-and-cancellation.md#idle-inactivity-timeout).*

## Retrying a run

```python
Command("flaky-fetch").retry(
    "transient_or_timeout",       # or "transient" — see below
    max_retries=3,                # up to 4 total attempts (default)
    initial_backoff=0.1,          # seconds before the first retry (default)
    multiplier=2.0,                # exponential growth per retry (default)
    max_backoff=30.0,             # cap on a single delay (default)
    jitter=True,                  # spread the wait over [0, delay] (default)
).run()
```

Honored only by the success-checking verbs (`run`/`exit_code`/`probe`) — the
non-erroring `output()`/`output_bytes()` never retry, since they never raise
in the first place. `retry_if` is a named preset over the error-classification
accessors, not an arbitrary predicate: `"transient"` covers a bare-retry-clears
spawn/IO condition (interrupted, would-block, a busy resource);
`"transient_or_timeout"` also retries a `.timeout()` expiry. Each attempt
**re-executes the whole command from scratch** — only retry operations safe to
repeat (a `git push` that already reached the server, then dropped the
connection, will be replayed if retried). A one-shot `stdin_bytes()`/
`stdin_text()` source can't survive a retry, so a command built with one is
never retried at all. Ignored by `Supervisor` (its own restart policy governs
keep-alive restarts — a different concern), `output_all`, and `Pipeline`.

`CliClient` has the same knobs, prefixed `default_` (`default_retry_if=`,
`default_max_retries=`, …) — `default_retry_if` is the required opt-in gate;
setting a tuning knob without it raises `ValueError`.

## Privileges and spawn flags

Spawn-time controls for sandboxing and service launch:

```python
# POSIX: drop privileges (groups and gid before uid) and detach.
(
    Command("worker")
    .gid(1000).groups([1000]).uid(1000)   # a correct drop sets all three
    .setsid()                             # new session: survives the controlling terminal
    .run()
)

# Windows: don't flash a console window from a GUI app.
Command("helper").create_no_window().run()

# Windows: give a console child a CTRL_BREAK to shut down cleanly before the hard kill.
Command("service").windows_graceful_ctrl_break().timeout(30.0).timeout_grace(5.0).run()

# Take the direct child down even if THIS process is killed before teardown runs.
Command("worker").kill_on_parent_death().start()

# Ask what scope that hardening actually reaches on THIS platform (build-time
# fixed; no prior kill_on_parent_death() needed).
scope = Command.kill_on_parent_death_scope()  # "whole_tree" | "direct_child_only" | "unsupported"
```

Platform honesty, not silent no-ops:

- `uid` / `gid` / `groups` / `setsid` are **POSIX-only**. On Windows the run
  raises `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.
- `windows_graceful_ctrl_break` is a **Windows-only opt-in**: at a graceful
  timeout (`timeout_grace`) or a group shutdown it sends the direct console
  child a `CTRL_BREAK` before the grace window, so a child that handles it can
  exit cleanly before the hard `TerminateJobObject` fallback (Windows otherwise
  has no soft-signal tier). Console-only — inert under `create_no_window` /
  detached, and it delivers `CTRL_BREAK`, not `CTRL_C`. A harmless no-op outside
  Windows (Unix's graceful tier already sends a real signal), like
  `create_no_window` — not one of the POSIX-only knobs that raise `Unsupported`.
- `kill_on_parent_death` is best-effort by design: kernel-guaranteed on Windows,
  `PR_SET_PDEATHSIG` on the direct child on Linux, a documented no-op on
  macOS/BSD. The graceful `with`-block teardown holds everywhere regardless.
  `Command.kill_on_parent_death_scope()` reports that reach programmatically —
  `"whole_tree"` on Windows, `"direct_child_only"` on Linux, `"unsupported"` on
  macOS/BSD — so you can read the *actual* abrupt-death scope instead of trusting
  the prose caveat. It is a static capability query fixed at build time: it needs
  no prior `kill_on_parent_death()` call (read it off the class or any instance)
  and describes only abrupt owner death — graceful teardown still kills the whole
  tree everywhere.

processkit wires **pipes**, not a pseudo-terminal, so a tool that *demands* a tty
(an `ssh`/`sudo` password prompt) won't get one. Drive such tools
non-interactively — key-based auth, `ssh -o BatchMode=yes`,
`GIT_TERMINAL_PROMPT=0`, or a known answer fed over
[interactive stdin](streaming.md). *Deeper: [Platform support](platforms.md).*

## Results

The capturing verbs hand back a `ProcessResult`:

```python
r = Command("git", ["merge", "feature"]).output()

r.stdout            # str (decoded)
r.stderr            # str
r.code              # int | None — None means killed (timeout / signal), no code
r.signal            # int | None — the signal number on Unix, else None
r.is_success        # code is in success_codes (default {0})
r.timed_out         # the run's own deadline expired
r.program           # the program name, for diagnostics
r.duration_seconds  # wall-clock duration
r.truncated         # an output_limit cap dropped output
r.combined          # stdout + stderr concatenated (property)
```

`output_bytes()` returns a `BytesResult` with the same fields (minus `combined`,
which can't join `bytes` stdout with `str` stderr), except `stdout` is raw `bytes`
(stderr stays decoded `str`). On a `BytesResult`, `truncated` is set when an
`output_limit` cap dropped output — the line-captured stderr under any cap, and
(since processkit 2.1.0) the raw bytes stdout too when a `max_bytes` ceiling bounds
it to a head/tail. A `max_lines` cap never truncates raw stdout (bytes have no line
count); only a `max_bytes` cap does.

```python
png = Command("convert", ["in.png", "png:-"]).output_bytes().stdout   # bytes
```

By default the success set is `{0}`. `success_codes([...])` **replaces** it — list
every code you accept. It affects `run()` and `is_success`, but **not**
`exit_code()` (always the raw int) or `probe()` (always 0/1). An empty sequence
raises `ValueError` (it would accept nothing).

```python
# diff exits 1 when files differ; treat that as success, not a failure.
differs = not Command("diff", ["a.txt", "b.txt"]).success_codes([0, 1]).probe()
Command("grep", ["needle", "log"]).success_codes([0, 1]).run()   # 1 (no match) is OK
```

## Errors

Every exception derives from `ProcessError`. The checking verbs raise these; the
capturing verbs do not (call `ProcessResult.ensure_success()` /
`BytesResult.ensure_success()` on an already-captured result to raise the same
exception after the fact — it returns `self` unchanged on success, so it
composes: `cmd.output().ensure_success().stdout`). Each carries **structured
fields**, not just a message:

| Exception | Raised when | Fields |
|---|---|---|
| `NonZeroExit` | a checking verb saw a non-success exit code | `program`, `code`, `stdout`, `stderr`, `stdout_bytes` (`bytes \| None`), `diagnostic` |
| `Timeout` | the run's deadline killed it | `program`, `timeout_seconds`, `stdout`, `stderr`, `stdout_bytes` (`bytes \| None`), `diagnostic` |
| `Signalled` | the process was killed by a signal | `program`, `signal`, `stdout`, `stderr`, `stdout_bytes` (`bytes \| None`), `diagnostic` |
| `ProcessNotFound` | the program couldn't be located / spawned | `program` |
| `PermissionDenied` | the program couldn't be spawned for lack of permission (e.g. a non-executable file), or a permission-denied OS error surfaced from elsewhere in the run (e.g. a group signal the OS refused) | `program` (`str \| None` — `None` for the broader "refused OS operation" case, where no program is being named) |
| `OutputTooLarge` | an `on_overflow="error"` cap was crossed | `program`, `max_lines`, `max_bytes`, `total_lines`, `total_bytes` |
| `ResourceLimit` | a memory / process / CPU cap was invalid or couldn't be enforced | — (reason is `str(exc)`) |
| `Unsupported` | the platform can't perform the requested operation | `operation` |
| `Cancelled` | a wired `CancellationToken` fired | `program` |

`diagnostic` (on the three stream-bearing exceptions) is the best human-facing
message — captured stderr if it carries text, otherwise captured stdout,
`None` if both streams are blank — so a generic `except ProcessError` handler
can log something useful without knowing which of the three it caught.

```python
from processkit import Command, NonZeroExit, Timeout, ProcessNotFound

try:
    Command("git", ["push"]).run()
except NonZeroExit as e:
    print(e.code, e.stderr)        # structured, not a parsed message
except Timeout as e:
    print(e.timeout_seconds)
except ProcessNotFound as e:
    print("missing:", e.program)
```

Three exceptions also derive from the builtin the stdlib raises for the same
condition, so familiar `except` clauses keep working: `Timeout` is also a
`TimeoutError` (as `asyncio.TimeoutError` is), `ProcessNotFound` is also a
`FileNotFoundError` (what `subprocess` raises), and `PermissionDenied` is also a
`PermissionError`. Cancelling an *awaited* run via asyncio (`task.cancel()`,
`asyncio.wait_for`, `asyncio.timeout`) surfaces as `asyncio.CancelledError`
instead of raising `Cancelled` (that's for an explicit `CancellationToken` wired
with `.cancel_on()`) — either way the tree is reaped.
*Deeper: [Timeouts & cancellation](timeouts-and-cancellation.md).*

### Secrets in diagnostics

`repr(Command(...))` is **redacted**: it shows the program, the argument *count*,
and env variable *names* — never argv values or env values. So a secret passed as
a flag or an `env(...)` value does **not** leak through a REPL echo, an `%r` log,
or a traceback frame.

The remaining channels carry raw values, so handle them with care:

- **Exception `stdout` / `stderr` fields carry the child's raw output verbatim**,
  and the exception *message* appends a **bounded last-line excerpt** of the
  captured output (stderr's last line, or stdout's when stderr is blank) — so if a
  tool echoes a token on failure, it can land in both. Don't forward exception
  text/fields to a low-trust log sink unredacted.
- **argv is visible to the OS** regardless of this library — any local user can
  read it via `ps` / `/proc/<pid>/cmdline` while the child runs. So for real
  secrets, prefer `env(...)` over a command-line flag: the env *value* is kept out
  of the `repr` and out of record/replay cassettes (only the variable name is
  recorded), and isn't exposed in the process listing.

## Pipelines

To connect stages `a | b | c` without a shell, use a `Pipeline` — either the `|`
operator or `.pipe()`. It runs to completion and exposes the same verbs:

```python
top = (Command("ps", ["aux"]) | Command("grep", ["python"])).run()
blob = (Command("cat", ["big.txt"]) | Command("gzip")).output_bytes().stdout
```

*Deeper: [Pipelines](pipelines.md).*

---

Next: [Streaming & interactive I/O](streaming.md) ·
[Process groups](process-groups.md) ·
[Timeouts & cancellation](timeouts-and-cancellation.md) ·
[Supervision](supervision.md) · [Testing your code](testing.md) ·
[Cookbook](cookbook.md) · [Platform support](platforms.md)

---

# Process groups

[‹ docs index](./)

A `ProcessGroup` ties the lifetime of a whole child-process **tree** to a
context manager: every process you start in the group — and everything *those*
processes spawn — is killed when the block exits. A returning, raising, or
cancelled owner never leaks subprocesses, because the kernel object that
contains the tree (a Windows Job Object, a Linux cgroup, or a POSIX process
group) catches grandchildren you never knew about.

You rarely need an explicit group for one-shot runs: a standalone
`Command(...).astart()` / `Runner().start(...)` handle already owns a *private*
tree that its own context manager reaps (see [Running commands](commands.md)).
Reach for `ProcessGroup` when several children should **share one fate**, or
when you want the group verbs below — whole-tree signals, suspend/resume,
member listing, resource limits, and stats.

- [Creating a group and the mechanism](#creating-a-group-and-the-mechanism)
- [Spawning into the group](#spawning-into-the-group)
- [Tearing down](#tearing-down)
- [Signalling the whole tree](#signalling-the-whole-tree)
- [Suspending and resuming](#suspending-and-resuming)
- [Inspecting members](#inspecting-members)
- [Resource limits: the sandbox](#resource-limits-the-sandbox)
- [Stats](#stats)
- [Live monitoring](#live-monitoring)

## Creating a group and the mechanism

The constructor is keyword-only. With no arguments you get a plain container
with the default graceful-shutdown grace (a short window, then escalate to a
hard kill):

```python
from processkit import ProcessGroup

with ProcessGroup() as group:
    print(group.mechanism)   # "job_object" | "cgroup_v2" | "process_group"
```

`mechanism` reports what you actually got at runtime. On a Linux host without
cgroup-v2 delegation it quietly reads `"process_group"` instead of
`"cgroup_v2"` — the same fallback that decides which features below are
available. See [Platform support](platforms.md) for the per-OS matrix; the
short version is *Windows strongest, macOS weakest.*

Tune the teardown timing at construction:

```python
group = ProcessGroup(shutdown_grace=10.0, escalate_to_kill=True)
```

`shutdown_grace` is a float of **seconds**. The resource-limit keywords
(`max_memory`, `max_processes`, `cpu_quota`) are covered under
[Resource limits](#resource-limits-the-sandbox).

## Spawning into the group

`start()` (sync) and `astart()` (async) put a full `Command` — capture,
streaming, timeouts, all of it — into the **shared** group and hand back a
`RunningProcess`:

```python
from processkit import Command, ProcessGroup

with ProcessGroup() as group:
    server = group.start(Command("dev-server"))
    worker = group.start(Command("worker"))
    # ... use them ...
# both, and every grandchild they forked, are gone here
```

```python
async with ProcessGroup() as group:
    server = await group.astart(Command("dev-server"))
```

A child started into a shared group does **not** own a private tree: its
`owns_group` is `False`. That distinction matters for teardown. Exiting *that
child's* own context manager (or dropping it) kills only that one child; it is
the **group's** teardown that reaps the whole tree.

```python
with ProcessGroup() as group:
    proc = group.start(Command("worker"))
    assert proc.owns_group is False
    with proc:                # this block kills only `proc`...
        ...
    # ...but other group members keep running until the group exits
```

The streaming and consuming surface of the returned `RunningProcess`
(`stdout_lines()`, `take_stdin()`, `outcome()`/`aoutcome()`, `finish()`/
`afinish()`, …) is documented in [Streaming & interactive I/O](streaming.md).

Since a `ProcessGroup` is itself a runner, you can also run a one-shot command
as a shared member without ever getting a `RunningProcess` handle back — the
same verb surface `Runner`/`ScriptedRunner`/… expose:

```python
with ProcessGroup() as group:
    result = group.output(Command("check-something"))   # a non-zero exit is data
    version = group.run(Command("tool", ["--version"]))  # requires a zero exit
```

## Tearing down

Prefer the context manager — its exit path is the no-orphan guarantee. For
explicit control you also have three verbs:

| Verb | What it does |
|---|---|
| `with` / `async with` exit | **Graceful** teardown of the whole tree — the same as `shutdown()` (signal → wait up to `shutdown_grace` → hard-kill survivors if `escalate_to_kill`). Always on, even if the block raises. |
| `group.kill_all()` | Immediate hard kill of the whole tree, mid-flight; idempotent. |
| `group.shutdown()` / `await group.ashutdown()` | **Graceful**: signal → wait up to `shutdown_grace` → hard-kill survivors if `escalate_to_kill`. |

```python
group = ProcessGroup(shutdown_grace=5.0, escalate_to_kill=True)
with group:
    group.start(Command("my-service"))
    ...
    group.shutdown()    # SIGTERM, give it 5s to flush, then SIGKILL stragglers
```

```python
async with ProcessGroup(shutdown_grace=5.0) as group:
    await group.astart(Command("my-service"))
    await group.ashutdown()
```

A child that handles `SIGTERM` and exits ends the grace **early** —
`shutdown` / `ashutdown` returns as soon as the tree is empty, not after the
full timeout. Use `kill_all()` when you want the tree gone *now* with no
grace at all.

**The no-orphan guarantee and its platform asymmetry.** The `with` /
`async with` exit path reaps the tree on every platform, and so does cancelling
an awaited run (`task.cancel()`, `asyncio.wait_for`, `asyncio.timeout`).
**Surviving a hard kill of the Python parent itself** — `SIGKILL`,
`os._exit` — is a *Windows-only* property, enforced by the kernel's
`KILL_ON_JOB_CLOSE`; on Linux and macOS teardown runs from the normal exit
path, which a hard kill skips. There is **no** Python destructor guarantee:
`__del__` and `atexit` do not run under `SIGKILL` / `os._exit`, so never lean on
them. Lean on the context manager. Full matrix in
[Platform support](platforms.md).

**The `process_group` backend's `setsid()`/`setpgid()` escape.** On
macOS/BSD, and on Linux whenever the group falls back from `cgroup_v2` to
`process_group` (no cgroup-v2 delegation — see
[the mechanism](#creating-a-group-and-the-mechanism)), every teardown path
above — the graceful `with`-exit *and* `kill_all()` — reaches the tree via
`killpg` against the POSIX process group. A child that calls `setsid()` or
`setpgid()` to leave that group before teardown runs is no longer a member,
so `killpg` does not reach it: it survives even a normal, non-crashing
`with`-exit, not just a hard kill of the parent. This is the standard trick
hostile code uses to outlive a sandbox; an ordinary double-fork that never
calls `setsid()`/`setpgid()` stays in the group and is reaped normally. The
Windows Job Object and the Linux cgroup-v2 backend have no such escape —
membership there is kernel-tracked, not session-based, so a descendant
cannot opt itself out. If a child appears to have escaped, see
[Troubleshooting](troubleshooting.md#a-child-escaped-a-posix-process-group-with-setsid).

*Deeper: keeping a service alive across crashes is [Supervision](supervision.md).*

## Signalling the whole tree

`signal(name)` broadcasts a POSIX signal to every member. Accepted names are
`"term"`, `"kill"`, `"int"`, `"hup"`, `"quit"`, `"usr1"`, `"usr2"`:

```python
with ProcessGroup() as group:
    group.start(Command("my-server"))
    group.signal("hup")     # "reload your configuration"
    group.signal("usr1")    # whatever the tool defines
```

`signal("kill")` and `kill_all()` take the same *atomic* whole-tree kill
path, so they cannot miss a process forked mid-broadcast. Every other signal is
a best-effort per-member broadcast against a tree that may be forking at that
instant.

Signals are POSIX-real on Linux, macOS, and BSD. On **Windows** only `"kill"`
maps onto the Job Object terminate; **every other name, including `"term"`,
raises `Unsupported`.** Catch it if you target multiple platforms:

```python
from processkit import Unsupported

try:
    group.signal("hup")
except Unsupported:
    ...   # no SIGHUP on this platform — reload some other way
```

## Suspending and resuming

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

```python
with ProcessGroup() as group:
    group.start(Command("cpu-hog"))
    group.suspend()     # the whole tree stops consuming CPU
    # ... inspect, snapshot, wait for the user ...
    group.resume()
```

Suspend/resume work on every current backend (anywhere a container exists — all
supported platforms). Two gotchas bite in practice:

- **Resume before starting new work.** Under the cgroup mechanism a child
  spawned into a *frozen* group starts frozen, and `start()` may not return
  until you `resume()`.
- **Resume before a graceful shutdown.** `shutdown` opens with a signal a
  frozen tree can't act on, so it would wait out the whole `shutdown_grace`.
  An immediate hard kill (`kill_all()` or `signal("kill")`) works on a frozen
  tree regardless; the `with`-exit is itself a graceful shutdown, so it carries
  the same caveat — `resume()` first.

## Inspecting members

`members()` returns the live member pids as a point-in-time snapshot:

```python
with ProcessGroup() as group:
    group.start(Command("worker-a"))
    group.start(Command("worker-b"))
    print(group.members())   # e.g. [4123, 4124]
```

What "members" means depends on the mechanism. On Windows and the Linux cgroup
backend it is the **whole tree** — every descendant pid. On the POSIX
process-group backends (macOS/BSD, Linux without cgroup) it is the tracked
group *leaders*, one pid per started child; their descendants are contained but
not enumerated. A tree that is forking races the snapshot.

`members_info()` returns that **same** set of members — the same point-in-time
snapshot, the same mechanism-dependent matrix above — but carries each pid in a
`MemberInfo` alongside best-effort metadata (parent pid, image name, start time):

```python
with ProcessGroup() as group:
    group.start(Command("worker"))
    for member in group.members_info():
        print(member.pid, member.ppid, member.exe_name, member.start_time)
```

Every field beyond `pid` is `None` wherever the platform can't report it —
`ppid`/`exe_name`/`start_time` are populated on Windows, Linux, and macOS, and
are all `None` on the BSDs (no wired-up per-process reader). Values are never
fabricated: a member that exits mid-snapshot is simply omitted rather than
reported with invented fields.

`start_time` is **not** a wall-clock timestamp — it is an *opaque per-process
identity token* whose unit and epoch are platform-specific (a Windows creation
`FILETIME`, Linux clock ticks since boot, macOS microseconds since the Unix
epoch). Do not interpret it or compare it across platforms; its sole use is
pairing with `pid` — two snapshots whose `pid` *and* `start_time` both match name
the same process instance — to tell a recycled pid apart from the original. And,
like the crate's `tracing` output, `MemberInfo` deliberately never carries the
raw command line or environment on any platform: an argv routinely holds
secrets, and redaction is a policy the consumer must own.

## Resource limits: the sandbox

The three limit keywords turn the group into a sandbox. They are a property of
the group, set once at construction and enforced by the same kernel object that
contains the tree:

```python
from processkit import Command, ProcessGroup

with ProcessGroup(
    max_memory=512 * 1024 * 1024,   # bytes, whole tree
    max_processes=64,               # fork-bomb ceiling
    cpu_quota=1.0,                  # one core (0.5 = half, 2.0 = two)
) as group:
    group.start(Command("untrusted-tool"))
```

`cpu_quota` is a fraction of a **single** core. On Windows it is converted
against the host CPU count and is approximate (a CPU-*rate* cap, not a hard
quota); on the Linux cgroup it is exact.

Limits need a **real container** — a Windows Job Object or a Linux **cgroup-v2
root**. If a requested cap can't be enforced, the constructor raises
`ResourceLimit` rather than handing you a silently-unbounded group:

```python
from processkit import ResourceLimit

try:
    group = ProcessGroup(max_memory=256 * 1024 * 1024)
except ResourceLimit:
    ...   # no Job Object / cgroup-v2 root here — limits unavailable
```

On Linux this requires the process to run at the real cgroup-v2 root. The
kernel's "no internal processes" rule forbids it under a container, a systemd
session/scope/service, or any non-root cgroup — so an ordinary container fails
too. macOS/BSD and the Linux process-group fallback have **no** whole-tree
limits at all. The prerequisites live in [Platform support](platforms.md); pair
limits with a locked-down `Command` (`env_clear().inherit_env(["PATH"])`,
`output_limit(...)`) per the [Cookbook](cookbook.md). For a quick diagnosis of a
`ResourceLimit` failure in those environments, see
[Troubleshooting](troubleshooting.md#resourcelimit-under-docker-systemd-or-a-non-root-cgroup).

## Stats

`stats()` returns a point-in-time `ProcessGroupStats` snapshot:

```python
with ProcessGroup() as group:
    group.start(Command("worker"))
    snap = group.stats()
    print(snap.active_process_count)    # int
    print(snap.peak_memory_bytes)       # int | None
    print(snap.total_cpu_time_seconds)  # float | None
```

`active_process_count` is always available. `peak_memory_bytes` and
`total_cpu_time_seconds` are populated only where the kernel accounts for the
whole tree (Windows, Linux cgroup); on the process-group backends they stay
`None` and only the count is reported.

For a single run's end-to-end resource profile, use `RunningProcess.profile()`,
covered in [Streaming & interactive I/O](streaming.md).

## Live monitoring

`stats()` alone is a snapshot you poll yourself. `sample_stats(group, every)`
turns that into a periodic series — a pure-Python async generator (no
`ProcessGroup` verb of its own) built directly on `stats()`, for a dashboard,
adaptive throttling, or an alert as the tree approaches a resource cap:

```python
from processkit import Command, ProcessGroup, sample_stats

async with ProcessGroup(max_memory=512 * 1024 * 1024) as group:
    await group.astart(Command("untrusted-tool"))
    async for snap in sample_stats(group, every=1.0):
        print(snap.active_process_count, snap.peak_memory_bytes)
        if snap.active_process_count == 0:
            break
```

The first snapshot is taken immediately, then one every `every` seconds, for as
long as you keep consuming — there is no overall deadline; `break` out of the
loop (or otherwise stop iterating) when you're done.

**Fused, and louder than the crate's stream.** The crate's `StatsSampler`
swallows the error on the first failed sample and the series just ends
silently. This generator instead lets `stats()`'s own exception (e.g.
`ProcessError` — "ProcessGroup is already closed" — once the group has torn
down) propagate out of the `async for` untouched, so you learn *why* the
series stopped instead of just that it did. That failure still ends the
series for good: the exception is never retried, and — because it is an
ordinary Python async generator — a further iteration attempt afterwards
raises `StopAsyncIteration` rather than calling `stats()` again. If the group
is already closed/invalid before you ever start iterating, that same
exception surfaces on the very first `async for` step, not as a silently
empty series.

*Deeper: testing code that drives a group without spawning is
[Testing your code](testing.md).*

---

Next: [Streaming & interactive I/O](streaming.md) ·
[Supervision](supervision.md) · [Platform support](platforms.md) ·
[Cookbook](cookbook.md)

---

# Sandboxing untrusted tools

[‹ docs index](./)

Agent/LLM frameworks routinely hand a model the ability to run a "tool" it
picked itself, with arguments it generated itself — a shell command, a code
interpreter, a scraper. That tool is, by construction, less trusted than code
you wrote: it should never be able to outlive your process, exhaust the host,
or run forever. This guide is not a new capability — it is a **composition** of
pieces documented individually elsewhere: [Running commands](commands.md)
(environment, output caps), [Process groups](process-groups.md) (whole-tree
resource limits), and [Timeouts & cancellation](timeouts-and-cancellation.md)
(deadlines). It ties them into one recipe, a checklist, and — most
importantly — an honest statement of what this buys you and what it does not.

- [The threat model](#the-threat-model)
- [The recipe](#the-recipe)
  - [1. Locked-down environment](#1-locked-down-environment)
  - [2. Bounded output](#2-bounded-output)
  - [3. Whole-tree resource limits](#3-whole-tree-resource-limits)
  - [4. A timeout](#4-a-timeout)
  - [5. Teardown](#5-teardown)
- [Checklist: run an untrusted tool safely](#checklist-run-an-untrusted-tool-safely)
- [Full example](#full-example)

## The threat model

Be precise about what a `ProcessGroup` sandbox is — and is not — before
leaning on it for anything that matters.

**processkit protects against:**

- **Process-tree leakage — on Windows, and on Linux at a cgroup-v2 root.**
  Every process the tool spawns, and everything *that* spawns, dies when the
  sandbox exits — enforced by the kernel container (Job Object / cgroup v2),
  not a best-effort signal to one pid. The `process_group` backend — macOS/BSD
  always, and Linux whenever it falls back from cgroup v2 without delegation
  (see [the mechanism](process-groups.md#creating-a-group-and-the-mechanism))
  — is **not** in this category: its teardown is `killpg`, which cannot reach
  a child that called `setsid()`/`setpgid()` to leave the group before
  teardown runs — a standard daemonization trick, and exactly how hostile
  code escapes it. (An ordinary double-fork that never calls
  `setsid()`/`setpgid()` stays in the group and is still reaped.) See [the
  no-orphan guarantee and its escape](process-groups.md#tearing-down).
- **Resource exhaustion — only when the kernel container is real.**
  Whole-tree memory, process-count (fork bombs), and CPU caps are enforced by
  the kernel on Windows, and on Linux only when this process runs at a
  **cgroup-v2 root** (see
  [Resource limits](process-groups.md#resource-limits-the-sandbox)). A
  container, a systemd session/scope/service, or any non-root cgroup gets you
  nothing — the kernel's "no internal processes" rule forbids delegation
  there — same as macOS/BSD having no whole-tree limit primitive at all (see
  [Platform support](platforms.md)). The Python API fails closed:
  `ProcessGroup(...)` raises `ResourceLimit` / `Unsupported` rather than
  handing back a silently-uncapped group. `python -m processkit`, however,
  catches that and silently re-spawns the child in an **uncapped**
  `ProcessGroup()`, only warning on stderr (see [Resource limits: hard cap or
  best effort?](cli.md#resource-limits-hard-cap-or-best-effort)) — if the cap
  exists to contain hostile code, treat that stderr warning as a hard
  failure, not something to shrug off and continue. Captured output is
  bounded independently of these caps, so a chatty or malicious child cannot
  grow the parent's memory without limit (see
  [Bounding captured output](commands.md#bounding-captured-output)).
- **Runaway execution time.** A timeout kills the whole tree at a deadline —
  see [Timeouts & cancellation](timeouts-and-cancellation.md).
- **Ambient credential/environment leakage.** `env_clear()` /
  `inherit_env([...])` starts the child from nothing rather than handing it
  the parent's full environment, secrets included — see
  [Environment and sandboxing](commands.md#environment-and-sandboxing). On
  POSIX you can additionally drop privileges — see
  [Privileges and spawn flags](commands.md#privileges-and-spawn-flags).

**processkit does NOT protect against:**

- **Filesystem access.** The tool can read and write anything the OS
  permits its (possibly privilege-dropped) user to touch. processkit does not
  chroot, bind-mount, or otherwise virtualize the filesystem.
- **Network access.** No firewalling or network namespace is applied; a
  sandboxed tool can still make outbound connections unless you restrict that
  another way (a container, a network policy, an egress proxy).
- **Syscall/namespace isolation.** This is not seccomp, and not a
  PID/mount/user-namespace container. A Job Object, cgroup, or process group
  bounds a *tree*'s lifetime and resource consumption — it does not restrict
  *which* syscalls the tree may issue.
- **Vetting the tool's behavior.** processkit does not sanitize, statically
  analyze, or judge what the program does — it bounds the blast radius (time,
  memory, CPU, process count, orphaned children), not the tool's actions
  within those bounds.

**In short:** this is *resource and lifetime* containment, not *security*
isolation. If you need syscall, filesystem, or network isolation, pair
processkit with an actual sandbox — a container, a VM, gVisor, a seccomp
profile, a restricted service account — processkit composes cleanly with any
of those; it just spawns and bounds whatever program you point it at. Do not
let this guide's checklist read as "fully isolated" — it is not.

## The recipe

Compose these five ingredients, in this order, for a locked-down run of an
untrusted tool:

```python
from processkit import Command, ProcessGroup, ResourceLimit, Unsupported

tool = (
    Command("untrusted-tool")
    .env_clear().inherit_env(["PATH"])                              # 1
    .output_limit(max_bytes=8 * 1024 * 1024, on_overflow="error")    # 2
    .timeout(30.0)                                                   # 4
    .kill_on_parent_death()
)

try:
    with ProcessGroup(                                              # 3
        max_memory=512 * 1024 * 1024, max_processes=64, cpu_quota=1.0,
    ) as group:
        group.start(tool)
        ...
    # 5. the `with` block's exit reaps the whole tree here — no orphans, ever.
except (ResourceLimit, Unsupported) as exc:
    ...   # no Job Object / cgroup-v2 root here (container, non-root cgroup, macOS)
```

### 1. Locked-down environment

Start the child from nothing and allow-list only what it needs — never hand
an untrusted tool the parent's full environment (which routinely carries
credentials). Full treatment, including the ordering of `env`/`env_remove`
on top: [Environment and sandboxing](commands.md#environment-and-sandboxing).

### 2. Bounded output

Cap `max_bytes` so a chatty or malicious tool cannot grow the parent's memory
without bound (a `max_lines`-only cap does not — one newline-free flood is a
single, unbounded line). `on_overflow="error"` turns hitting the cap into a
failure rather than a silent drop, which is usually what you want for a tool you
don't trust — and in that mode the ceiling counts the raw bytes read from the
child's pipe, so it holds even for output that is binary or not valid UTF-8. (A
drop mode bounds the *retained* output instead, measured in decoded line content;
see [what `max_bytes` counts](commands.md#what-max_bytes-actually-counts).) Full
treatment: [Bounding captured output](commands.md#bounding-captured-output).

### 3. Whole-tree resource limits

`max_memory` / `max_processes` / `cpu_quota` on the `ProcessGroup` cap the
*whole tree* — not just the direct child — at the kernel level. This needs a
real container (a Windows Job Object or a Linux cgroup-v2 root); where one
isn't available, the constructor raises `ResourceLimit` rather than handing
back a silently-unbounded group. Full treatment, including the platform
matrix: [Resource limits: the sandbox](process-groups.md#resource-limits-the-sandbox).

### 4. A timeout

Untrusted code should never run unbounded. `.timeout(seconds)` kills the
whole process tree at the deadline; pair it with `.timeout_grace(...)` for a
graceful signal-then-kill if the tool might want to clean up first. Full
treatment: [Timeouts & cancellation](timeouts-and-cancellation.md).

### 5. Teardown

Prefer the context manager (`with ProcessGroup() as group: ...` / `with
Command(...).start() as proc: ...`) over any manual verb — its exit path is
the no-orphan guarantee, on every platform, even if the block raises. Never
lean on `__del__` / `atexit`: neither runs if the parent itself is hard-killed.
Full treatment: [Tearing down](process-groups.md#tearing-down).

## Checklist: run an untrusted tool safely

- [ ] Environment locked down: `env_clear()` + `inherit_env([...])` (or an
      explicit allow-list built from `env(...)` calls) — never inherit the
      parent's full environment into an untrusted child.
- [ ] Captured output bounded: `output_limit(max_bytes=...)` — a
      `max_lines`-only cap does not bound memory.
- [ ] Whole-tree resource limits set on a `ProcessGroup`: `max_memory`,
      `max_processes`, `cpu_quota` — with `ResourceLimit` / `Unsupported`
      handled where the kernel container isn't available.
- [ ] A timeout set (`Command.timeout(...)`, and `Pipeline.timeout(...)` for a
      piped chain) — untrusted code should never run unbounded.
- [ ] Teardown via a context manager, never `__del__` / `atexit`.
- [ ] `kill_on_parent_death()` set on the tool, so it dies even if your own
      process crashes before teardown runs — on POSIX this covers only the
      **direct child** (Linux `PR_SET_PDEATHSIG`; not inherited by
      grandchildren, resettable by the child itself via
      `prctl(PR_SET_PDEATHSIG, 0)`, and cleared by credential changes or by
      executing a setuid/setgid/capability-bearing binary; ordinary `execve`
      preserves it), and
      is a documented no-op on macOS/BSD. A tree-wide guarantee against a
      hard kill of your own process is Windows-only (Job Object). See
      [Privileges and spawn flags](commands.md#privileges-and-spawn-flags).
- [ ] (POSIX only, if running as a privileged user) privileges dropped with
      all three of `uid` / `gid` / `groups([...])` set together — `uid` alone
      leaves the child holding the parent's supplementary groups. For this
      incomplete-drop symptom, see
      [Troubleshooting](troubleshooting.md#privilege-drop-sets-uid-but-not-gid-and-groups).
- [ ] Read [the threat model](#the-threat-model) above — this checklist buys
      resource and lifetime containment, not syscall/filesystem/network
      isolation.

## Full example

`examples/04_sandbox_resource_limits.py` runs this recipe end to end for an
agent making a couple of tool calls in one sandboxed session: a locked-down,
output-capped, per-call-timeout command; whole-tree memory/process/CPU limits
on the shared group; and teardown on context-manager exit — degrading
gracefully to "contained, but uncapped" where the kernel container isn't
available (a container, a non-root cgroup, macOS).

```bash
python examples/04_sandbox_resource_limits.py
```

---

Next: [Process groups](process-groups.md) · [Running commands](commands.md) ·
[Timeouts & cancellation](timeouts-and-cancellation.md) ·
[Cookbook](cookbook.md) · [Platform support](platforms.md)

---

# Streaming & interactive I/O

[‹ docs index](./)

The one-shot verbs in [Running commands](commands.md) — `output()`, `run()`,
`output_bytes()` — buffer the *whole* output and hand it back at exit. That is
exactly what you want for a `git rev-parse`. It is exactly what you *don't* want
for a long-running or conversational child: a dev server you watch, a build you
follow, an interpreter you talk to. For those, `await Command(...).astart()`
returns a live `RunningProcess` you drive yourself — stream stdout as it
arrives, write stdin incrementally, probe for readiness, profile a run, and tear
the tree down deterministically.

- [Lifecycle](#lifecycle)
- [Streaming stdout](#streaming-stdout)
- [Tee output to a file](#tee-output-to-a-file)
- [Live per-line callbacks](#live-per-line-callbacks)
- [Interleaved stdout and stderr](#interleaved-stdout-and-stderr)
- [Interactive stdin](#interactive-stdin)
- [Readiness probes](#readiness-probes)
- [Live introspection and per-run telemetry](#live-introspection-and-per-run-telemetry)
- [Deterministic teardown](#deterministic-teardown)

## Lifecycle

```python
from processkit import Command, Runner

# Async setup — the handle owns a private process tree:
proc = await Command("dev-server").astart()

# Sync setup, same live handle (the consuming verbs below have a sync twin too):
proc = Command("dev-server").start()        # or: Runner().start(Command("dev-server"))
# …or hand the tree to a group that owns its fate instead of the handle:
#   proc = group.start(Command("dev-server"))   # see Process groups

proc.pid              # int | None — None once the handle is consumed
proc.elapsed_seconds  # float | None — wall time since spawn
proc.owns_group       # True for a standalone start()/astart() handle; False under a group
```

Whichever way you start it, **consume the handle exactly one way** — each of
these comes in a sync/async pair (like everywhere else in this library) and
*spends* the handle (afterward the getters return `None` and a second
consuming verb raises):

| Verb pair | Returns | Use when |
| --- | --- | --- |
| `proc.outcome()` / `await proc.aoutcome()` | `Outcome` | you only need the exit; output is discarded |
| `proc.finish()` / `await proc.afinish()` | `Finished` | **after streaming stdout** — exit + captured stderr, *without* buffering stdout |
| `proc.output()` / `await proc.aoutput()` | `ProcessResult` | capture everything (same as the one-shot `output()`) |
| `proc.output_bytes()` / `await proc.aoutput_bytes()` | `BytesResult` | capture, stdout as `bytes` |
| `proc.profile(every_seconds)` / `await proc.aprofile(every_seconds)` | `RunProfile` | full outcome + CPU/memory samples; output discarded |
| `proc.shutdown(grace_seconds)` / `await proc.ashutdown(grace_seconds)` | `Outcome` | graceful signal → wait → hard-kill |

(`outcome`/`aoutcome`, not `wait`/`await` — `await` is a reserved word, so it
can't be a method name.) Use whichever half of a pair matches your calling
code — the sync half blocks the calling thread (the same interruptible driver
as `Command.output()`), the async half is a coroutine.

`Outcome` carries `code: int | None`, `signal: int | None`, `timed_out: bool`,
and `exited_zero: bool` (literal "exit code 0" — it has no `success_codes`
context; for the command's own verdict use `ProcessResult.is_success`). There is
also a synchronous `proc.kill()` (like `subprocess.Popen.kill()`) for "stop it
now, I'll read the code myself with `proc.outcome()` / `await proc.aoutcome()`."

`start()`, `astart()`, and `Runner().start()` put the child in a **private group
the handle owns**: tearing the handle down kills the whole tree, and
`shutdown()`/`ashutdown()` work on it — named to match
`ProcessGroup.shutdown()`/`ashutdown()`. The shared-group variant —
`group.start(cmd)` — gives the same handle, but the *group* controls the
tree's fate (`owns_group` is `False`), so `shutdown()`/`ashutdown()` raise
`Unsupported` there; tear such a child down via the group (or `kill()`). See
[Process groups](process-groups.md).

## Streaming stdout

`stdout_lines()` is a synchronous setup call that returns a `StdoutLines` async
iterator of decoded lines, yielded as the child produces them — no waiting for
exit, no full-output buffering:

```python
from processkit import Command

proc = await Command("cargo", ["build", "--release"]).astart()

async for line in proc.stdout_lines():
    print("build:", line)

# The stream ended (stdout closed). finish() collects the outcome and stderr —
# stderr was drained in the background the whole time, so a noisy child could
# never block on a full pipe.
finished = await proc.afinish()
if not finished.exited_zero:
    print(finished.outcome.code, finished.stderr)
```

`Finished` exposes `outcome`, `stderr: str`, `code: int | None`, and
`exited_zero: bool` (same "exit code 0" meaning as `Outcome.exited_zero`). Things
to know:

- **Call `stdout_lines()` once.** stdout is consumed a single time; a second
  `stdout_lines()` / `output_events()` call, or a non-piped stdout, raises
  rather than yielding a silently-empty stream.
- **The command's `.timeout(d)` bounds the stream** on an own-group handle: at
  the deadline the tree is killed, the pipes close, and the iterator ends — a
  streamed run can't hang past its deadline. The following `finish()` reflects
  it (`outcome.timed_out`).
- For an *ad-hoc* bound, wrap the loop in `asyncio.timeout(...)` and let the
  [teardown](#deterministic-teardown) kill the tree (shown below).
- The line counters tick live: `proc.stdout_line_count` /
  `proc.stderr_line_count` are cheap progress gauges while you stream.

*Deeper: output buffering and capture limits apply to streamed runs too —
[Running commands](commands.md).*

## Tee output to a file

Sometimes you want *both*: a live log written somewhere **and** the captured
result in hand — a build whose output tails into `build.log` while you still get
the final `ProcessResult` to inspect. `stdout_tee(sink)` / `stderr_tee(sink)` do
that in one line, with no manual loop over `stdout_lines()`:

```python
from processkit import Command

result = Command("cargo", ["build", "--release"]).stdout_tee("build.log").output()

# The file received the live stream, line by line, as it was produced …
assert open("build.log").read().startswith("   Compiling")
# … and capture is untouched — the tee does not steal output from the result.
print(result.stdout)          # the full captured stdout, same as without the tee
```

Each decoded line is written to the sink as it lands, followed by a `\n` (a CRLF
terminator is normalized to `\n`). The tee runs *independently* of capture, so
`result.stdout` still holds the whole output. It also works with the streaming
verbs — `start()` + `stdout_lines()` / `output_events()` — not just the one-shot
capture verbs; the same lines flow to the iterator and the sink.

The sink can also be a **Python writer** — any object with a `write()` method
(`io.StringIO`, `sys.stderr`, a text-mode file, a logger wrapper) — to mirror
the child's output straight into your own console, buffer, or logger while still
capturing it:

```python
import io
from processkit import Command

buf = io.StringIO()
result = Command("cargo", ["build", "--release"]).stdout_tee(buf).output()

# Each decoded line (plus a "\n") was passed to buf.write() as a str, live …
assert buf.getvalue().startswith("   Compiling")
# … and capture is still whole — the object is only mirrored to, never drained.
print(result.stdout)
```

Things to know:

- **A file path or a Python writer.** The sink is either a filesystem path (`str`
  or `os.PathLike[str]`) or an object with a callable `write()` — the two are
  told apart by whether the argument exposes `write` (neither `str` nor
  `pathlib.Path` does). A writer is a **text** sink: each decoded line is passed
  to `write()` as a `str`, so pass a text-mode object (`io.StringIO`,
  `sys.stderr`, a file opened in text mode, a logger wrapper), not a binary one
  (`io.BytesIO`, a `"wb"` file) whose `write(str)` would raise `TypeError`. The
  writer is **not** owned — it is never closed for you, so you keep using your
  `sys.stderr` / open file after the run. `append` tunes only how a *file path*
  is opened (see below); passing `append=True` with a writer raises `ValueError`
  rather than being silently ignored.
- **A file is opened now, at build time.** `stdout_tee(path)` opens the file
  the moment you call it (the crate takes a concrete sink, not a lazy factory),
  **not** when the command runs. So an unopenable path — a missing parent
  directory, a directory, a permission denial — raises the matching `OSError`
  (`FileNotFoundError`, `IsADirectoryError`, `PermissionError`, …) right at the
  builder call, before any run verb. (A writer object is used as-is, so nothing
  is opened — this timing applies only to the path form.)
- **Truncate by default, or append (file paths).** A file sink is created if
  absent and truncated; pass `append=True` to open it in append mode instead (to
  grow an existing log). Because the open handle is shared across re-runs of the
  *same* built `Command` (retries, a reused command, `Supervisor` incarnations),
  those sequential runs **append** to the one file with no delimiter, and
  concurrent clones (pipeline stages) **interleave**. For per-run separation,
  build a fresh `Command` (a fresh path) per run.
- **A slow sink applies backpressure, it does not block the runtime.** The tee
  write is awaited on the capture pump, so a slow disk slows the pump, fills the
  OS pipe, and makes the child block on its next write — rather than stalling the
  event loop. A Python writer gets the same treatment: each `write()` is
  dispatched to the runtime's blocking pool (re-acquiring the GIL there), so even
  a `write()` that *sleeps* applies backpressure without blocking the async event
  loop or deadlocking the runtime. A sink that blocks *forever* (not merely slow)
  parks the pump until teardown; a plain file or a prompt writer never does this.
- **A tee write error is isolated.** If a write to the sink fails mid-run, the
  tee is disabled for the rest of the run and a warning is emitted (under
  [`enable_logging()`](cookbook.md#see-what-processkit-runs-logging)) — the run
  itself and its captured result are unaffected, never broken by the sink. For a
  Python writer, a `write()` (or `flush()`) exception is additionally reported
  via `sys.unraisablehook`, so it is visible even without `enable_logging()`
  (and catchable in a test via a custom hook).
- **No-op unless the line pump runs.** The tee fires from the line-capture pump,
  so it is inert under `stdout("inherit")` / `stdout("null")` (no pump) and under
  `output_bytes()` (raw capture, no line pump). Reach for it with the line verbs
  — `output()` / `aoutput()`, `run()`, or `start()` + `stdout_lines()` /
  `output_events()`.

## Live per-line callbacks

`stdout_lines()` / `output_events()` are async-only — they hand back an async
iterator, so they need an event loop to drive. `on_stdout_line(callback)` /
`on_stderr_line(callback)` give the **synchronous** surface the same live
observation: `callback` runs on every decoded line *as it is produced*, even
while `.output()` / `.run()` is still blocking:

```python
from processkit import Command

def log_line(line: str) -> None:
    print("build:", line)

result = Command("cargo", ["build", "--release"]).on_stdout_line(log_line).output()
# "build: ..." printed live, one call per line, while output() was still blocking.
print(result.stdout)   # capture is untouched — the callback observes, it doesn't consume.
```

They work identically on the async verbs and on a streamed run (`start()`/
`astart()` + `stdout_lines()` / `output_events()`) — one callback, every path;
adding them does not turn the sync surface async-only, and does not replace the
streaming iterators (which stay the only way to *consume* lines one at a time
from Python — a callback only *observes*).

Things to know:

- **At most one handler per stream.** A repeat call **replaces** the previous
  one (builder semantics, like `timeout()`); compose inside a single Python
  callable to fan out to more than one observer.
- **A raising callback never derails the run.** An exception raised inside
  `callback` is reported via `sys.unraisablehook` (visible on stderr, or
  catchable in a test via a custom `sys.unraisablehook`) instead of
  propagating — the run and its captured result are unaffected either way.
- **No-op unless that stream's line pump runs**, same family as
  `stdout_tee`/`stderr_tee`: `on_stdout_line` is inert under
  `stdout("inherit")` / `stdout("null")` and under `output_bytes()` (stdout is
  captured raw there, bypassing the line pump). `on_stderr_line` is inert under
  `stderr("inherit")` / `stderr("null")` — but **not** under `output_bytes()`:
  that verb only bypasses the *stdout* line pump, stderr keeps decoding through
  it exactly as under `output()`.
- **Runs independently of `stdout_tee`/`stderr_tee`.** Set both and both fire
  per line — a callback and a file tee are not mutually exclusive.

## Interleaved stdout and stderr

When the *interleaving* matters — a `--watch` build that prints progress to
stdout and diagnostics to stderr — `output_events()` returns an `OutputEvents`
async iterator that merges both streams in arrival order:

```python
proc = await Command("vite", ["build", "--watch"]).astart()

async for ev in proc.output_events():
    tag = "ERR" if ev.is_stderr else "out"
    print(f"[{tag}] {ev.text}")     # ev.stream is "stdout" / "stderr"
```

Each `OutputEvent` has `stream: Literal["stdout", "stderr"]`, `is_stderr: bool`,
and `text: str`. Like
`stdout_lines()`, this consumes the pipes once — pick `stdout_lines()` *or*
`output_events()`, not both.

Things to know:

- **Only output lines are yielded.** Underneath, the core stream carries the
  child's whole lifecycle (it reports process start and exit as well as output),
  but those non-line events are filtered out here rather than handed to you as an
  `OutputEvent` with an empty `text` — which would be indistinguishable from a
  real blank line the child printed, and would quietly corrupt anything that
  counts or joins lines. What they carry is already on surfaces you have: the
  start is `proc.pid`, the exit is what the finisher below returns.
- **Iterate fully, then finish.** Draining the iterator also drives the run to
  completion, so the usual order terminates:

  ```python
  async for ev in proc.output_events():
      ...
  finished = await proc.afinish()   # or: await proc.aoutcome()
  ```

  `finish()`/`afinish()` reports the outcome (its `stderr` is empty — you already
  received stderr as events), and `outcome()`/`aoutcome()` reports the exit alone.
- **The capture verbs do not apply to such a run.** `output()` / `output_bytes()`
  / `profile()` (and their `a`-twins) raise a `ProcessError` naming
  `output_events()` once that stream has **taken the run over** — which it does
  as soon as it sees the child exit, and always by the time the iterator ends:
  stdout was consumed by the iterator and stderr was delivered as events, so
  there is nothing left for them to capture, and the run is already complete so
  there is nothing left to sample. Reach for `finish()` / `outcome()` instead.
  (A **breaking** change that came with the processkit 3.0 migration: those verbs
  used to return empty captures alongside the run's real outcome.)
- **Leaving the loop early is fine — with one boundary.** `break` out whenever
  you like: `finish()`/`afinish()` and `outcome()`/`aoutcome()` report the run
  either way, and dropping the handle (or exiting its `with` block) still tears
  the tree down — including after the stream has taken the run over, where the
  teardown claims the run *from* it (see
  [Deterministic teardown](#deterministic-teardown)). The three capture verbs
  above are the exception, and *when* you stopped decides which of two behaviours
  you get:

  | when you stopped iterating | `finish()` / `outcome()` | `output()` / `output_bytes()` / `profile()` |
  |---|---|---|
  | the stream had already seen the child exit — always so once the iterator ended, and possible after a `break` too, out of a command that finished while you were reading it | report the run | raise `ProcessError` |
  | the child was still running | report the run | as before 3.0: wait for exit and return empty captures (`profile()` samples the rest of the run) |

  Which row a given `break` lands in follows the child's timing rather than how
  you wrote the loop, so treat the capture verbs as unavailable once you have
  streamed events and use a finisher.

## Interactive stdin

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

```python
# bc evaluates each stdin line and prints the result.
proc = await Command("bc").keep_stdin_open().astart()
stdin = proc.take_stdin()          # ProcessStdin (raises if stdin wasn't kept open)
answers = proc.stdout_lines()

await stdin.write_line("2 + 2")    # writes "2 + 2\n", flushed
print("=", await anext(answers))   # 4

await stdin.write_line("6 * 7")
print("=", await anext(answers))   # 42

await stdin.close()                # send EOF — bc exits (idempotent)
finished = await proc.afinish()
assert finished.exited_zero
```

Full runnable example: `examples/05_interactive_stdin.py` — a request/response
conversation (multiple exchanges) with a small inline calculator REPL.

`ProcessStdin` is fully awaitable: `await write(bytes)`, `write_line(str)`
(newline + flush), `send_control(str)`, `flush()`, and `close()` (EOF).
`send_control()` accepts exactly one recognized control character and writes
the mapped control byte to the child's stdin pipe: for example,
`await stdin.send_control("c")` writes Ctrl-C (`\x03`) and
`await stdin.send_control("d")` writes Ctrl-D (`\x04`). Invalid input raises
`ValueError`.

This is a byte in a normal pipe, not a terminal signal. It only affects
children that read and interpret that byte from stdin; real terminal semantics
such as SIGINT/SIGTSTP delivery require a pseudoterminal, which `processkit`
does not provide yet.

`take_stdin()` **raises** `ProcessError` if the `Command` didn't
`keep_stdin_open()` or the writer was already taken — so a missing setup fails
right here, not later on a `None`.

**Not the same as `inherit_stdin()`.** `keep_stdin_open()` + `take_stdin()` hands
you a **crate-managed pipe** you write to from Python — the crate mediates every
byte. [`inherit_stdin()`](commands.md#standard-input) is the opposite: it gives
the child the parent's **real** stdin (the actual terminal / file / pipe this
process was launched with), so the crate touches nothing and there is no writer
to take (`take_stdin()` returns nothing there, exactly as for a run that never
kept stdin open). Reach for `inherit_stdin()` when a child must talk to the real
terminal — `git commit` opening `$EDITOR`, a password prompt — and for the
byte-by-byte conversational exchange above, `keep_stdin_open()`. The two are
**mutually exclusive**: setting both is rejected as a `ProcessError` at launch
(not when you build the `Command`).

**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. The `bc` exchange above is safe because it interleaves one small write with
one read. But 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. When you both
feed a sizable stdin **and** the child talks back, drain stdout from one task
while writing stdin from another:

```python
import asyncio

proc = await Command("filter-tool").keep_stdin_open().astart()
stdin = proc.take_stdin()

async def feed():
    for chunk in big_payload:
        await stdin.write(chunk)
    await stdin.close()

async def drain():
    async for line in proc.stdout_lines():
        handle(line)

await asyncio.gather(feed(), drain())
await proc.aoutcome()
```

*Deeper: the non-interactive `stdin_text` / `stdin_bytes` sources never deadlock
— they're pumped on a background task. See [Running commands](commands.md).*

## Readiness probes

"Start a server, then use it" needs *ready*, not merely *started*. Six
free async helpers replace the arbitrary `asyncio.sleep`, each bounded by its
own deadline:

```python
from processkit import (
    Command,
    wait_until,
    wait_for_path,
    wait_for_port,
    wait_for_unix_socket,
    wait_for_http,
    wait_for_line,
)

proc = await Command("my-server").astart()
lines = proc.stdout_lines()        # bind once — you reuse this same iterator

# 1. A line on stdout (returns the matching line) — a plain string is a
#    substring-match shorthand for a str-yielding iterator:
banner = await wait_for_line(lines, "listening on", timeout=10)
# …or a callable predicate, which also works over any async iterator, not
# just str lines (e.g. `proc.output_events()`'s OutputEvent items):
banner = await wait_for_line(lines, lambda l: "listening on" in l, timeout=10)

# 2. A TCP port accepting connections:
await wait_for_port("127.0.0.1", 8080, timeout=10)

# 3. An HTTP endpoint answering with an acceptable status (2xx by default) — a
#    stronger signal than the port alone, which a warming-up server accepts
#    while still replying 503. `expected_status` takes a set/range or a predicate:
await wait_for_http("127.0.0.1", 8080, "/health", timeout=10)

# 4. A Unix-domain socket accepting connections (stronger than a path check):
await wait_for_unix_socket("/run/my-server.sock", timeout=10)

# 5. A path appearing on the filesystem (a pid file or other marker, …):
await wait_for_path("/run/my-server.sock", timeout=10)

# 6. Any predicate — sync bool OR an awaitable (a DB ping, …):
await wait_until(lambda: health_check_passes(), timeout=10, interval=0.1)

# ready — keep consuming from the SAME iterator:
async for line in lines:
    ...
```

(Named `wait_until`, not `wait_for` — the latter would collide with
`asyncio.wait_for`, whose semantics differ: it bounds one *awaitable*, not a
*polled predicate*.)

Semantics, deliberately uniform:

- The six probes are `wait_for_line`, `wait_for_port`, `wait_for_http`,
  `wait_for_unix_socket`, `wait_for_path`, and `wait_until`.
- A probe that can't pass within its deadline raises **`WaitTimeout`**
  (`ProcessError`, `TimeoutError`) — so `except TimeoutError` catches both run
  and readiness timeouts, and `.timeout_seconds` reads the configured deadline
  either way. `wait_for_port` additionally sets `.host`/`.port`, `wait_for_http`
  sets `.host`/`.port`/`.path`, and `wait_for_path` / `wait_for_unix_socket` set
  `.path`. `wait_for_port` / `wait_for_http` / `wait_for_unix_socket` also chain the last failed attempt (a
  connection error, or — for `wait_for_http` — the last unexpected status) as
  `__cause__`.
- `wait_for_line` additionally raises `ProcessError` if the stdout stream ends
  *before* a match — no waiting out a 10s deadline on a dead server. It
  consumes items up to (and including) a match; iteration may continue
  afterward **only when a match was found** — exactly how far it advanced past
  the last inspected item on a timeout is unspecified, so don't rely on the
  iterator's position there. `wait_for_port` / `wait_for_http` / `wait_for_path`
  / `wait_for_unix_socket` / `wait_until` don't touch the pipes at all.
- A failed probe **never kills the child** — you decide: retry, log, or tear
  down.
- `wait_until` / `wait_for_port` / `wait_for_http` / `wait_for_path` /
  `wait_for_unix_socket` poll every `interval` seconds (`ValueError` if
  `interval <= 0`). A sync `wait_until`
  predicate runs on the event loop, so keep it non-blocking; for blocking work,
  pass an awaitable.

*Deeper: bounding the whole run (not just the wait) is
[Timeouts & cancellation](timeouts-and-cancellation.md).*

## Live introspection and per-run telemetry

A running child reports its own resource usage live; the getters are properties
(not calls), and each returns `None` once the handle is consumed:

```python
proc = await Command("crunch").astart()
proc.pid                 # int | None
proc.elapsed_seconds     # float | None — wall time
proc.cpu_time_seconds    # float | None — user + kernel so far
proc.peak_memory_bytes   # int | None
proc.stdout_line_count   # int | None — progress while you stream
```

Or turn a whole run into a summary with `profile()`/`aprofile()`, which
samples the child every `every_seconds` until exit (the run's normal timeout
still applies; like `outcome()`/`aoutcome()`, the output is drained and
discarded, not returned). `RunProfile` is a **superset of `Outcome`**: it
carries the full `outcome` (`code` / `signal` / `timed_out`) *and* the
resource samples:

```python
proc = await Command("crunch").astart()
prof = await proc.aprofile(every_seconds=0.1)

print(
    f"exit={prof.code} signal={prof.signal} timed_out={prof.timed_out} "
    f"wall={prof.duration_seconds:.2f}s cpu={prof.cpu_time_seconds} "
    f"peak_rss={prof.peak_memory_bytes} "
    f"avg_cpu_cores={prof.avg_cpu_cores} ({prof.samples} samples)"
)
# prof.outcome is the same Outcome outcome()/aoutcome() would return.
# avg_cpu_cores = cpu / wall — e.g. 1.7 ≈ 1.7 cores busy
```

These read the *child process itself*, 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](platforms.md).

*Deeper: whole-tree (grandchildren included) resource stats live on
[Process groups](process-groups.md).*

## Deterministic teardown

A `RunningProcess` is a context manager — sync and async. For a standalone
`start()` / `astart()` / `Runner().start()` handle, exiting the block hard-kills
its whole private tree (best-effort; see [Platform support](platforms.md)), even
if the block raises, without waiting on Python's GC:

```python
async with await Command("flaky-server").astart() as proc:
    async for line in proc.stdout_lines():
        if "ready" in line:
            break
# proc and its whole private tree are reaped here
```

This composes with an *ad-hoc* time bound — wrap the loop, let the exit clean up:

```python
import asyncio

async with await Command("tail", ["-f", "app.log"]).astart() as proc:
    try:
        async with asyncio.timeout(5):
            async for line in proc.stdout_lines():
                print(line)
    except TimeoutError:
        pass
# context-manager exit kills the tree on the way out
```

Three rules close the loop:

- **A consumed handle is spent.** If you consume inside the block (`await
  proc.output()` / `.outcome()` / `.finish()` / `.shutdown(...)` — or their
  `a`-prefixed async twins), the exit is a
  no-op — the verb already settled the run. Afterward the getters return `None`
  and a second consuming verb raises.
- **Streaming events does not weaken it.** Once an `output_events()` stream has
  [taken the run over](#interleaved-stdout-and-stderr) the completion of that run
  is being driven for you in the background — and leaving the block still ends
  the tree *there*, by claiming that work back rather than waiting on it. This is
  the case that matters after an early `break`: the child may be gone while a
  grandchild still holds its pipe, and the block's exit is what stops that
  grandchild from outliving it (on Windows, that also means the files and
  directories it holds open are released before the `with` returns, not moments
  later).
- **Prefer `shutdown()`/`ashutdown()` for a graceful stop.** `await proc.ashutdown(grace_seconds=5)`
  signals the tree, waits up to `grace_seconds`, then hard-kills — and returns
  the `Outcome`. Reach for the context manager when you just want the tree
  *gone*; reach for `shutdown()` when the child deserves a chance to flush.
  (After an `output_events()` stream has taken the run over there is nothing left
  to signal — the child has exited — so `shutdown()` reports that run's real
  outcome, waiting for its output to finish draining exactly as `finish()` does,
  rather than escalating against surviving grandchildren. When the *bound*
  matters more than the outcome, leave the block.)

Cancellation is plain asyncio here: `task.cancel()` on the task awaiting a
consuming verb tears the tree down and propagates `CancelledError`. The full
treatment — deadlines, cooperative shutdown — is in
[Timeouts & cancellation](timeouts-and-cancellation.md).

*Deeper: drive this entire surface with no subprocess at all — a
`ScriptedRunner.start()` returns a streamable handle whose canned lines flow
through the same pump. See [Testing your code](testing.md).*

---

Next: [Process groups](process-groups.md) ·
[Timeouts & cancellation](timeouts-and-cancellation.md) ·
[Cookbook](cookbook.md)

---

# Pipelines

[‹ docs index](./)

`a | b | c` **without a shell**. Each stage's stdout feeds the next stage's
stdin through an in-process relay — there is no shell string anywhere, so no
quoting rules, no word splitting, no injection surface. Every stage spawns into
its own kill-on-exit [process-group](process-groups.md) sub-group. A checked
stage failure, chain timeout, or cancellation fans teardown across every
sub-group, so the chain still lives and dies as a unit while a per-stage timeout
can first reap that stage's entire subtree.

```python
from processkit import Command

# git log --format=%an | sort | uniq -c
authors = (
    Command("git", ["log", "--format=%an"])
    | Command("sort")
    | Command("uniq", ["-c"])
).run()
print(authors)
```

## Building a pipeline

`Command.pipe(next)` starts a `Pipeline`; chain more stages with
`Pipeline.pipe`. The `|` operator is sugar for the same thing — `a | b | c` is
exactly `a.pipe(b).pipe(c)`:

```python
authors = (
    Command("git", ["log", "--format=%an"])
    .pipe(Command("sort"))
    .pipe(Command("uniq", ["-c"]))
    .run()
)
```

Python's `|` binds *looser* than a method call, so parenthesize the whole chain
before a terminal verb — `(a | b).run()`, never `a | b.run()` (which would call
`run()` on `b` alone). The `.pipe(...).pipe(...)` form chains cleanly without the
extra parentheses.

The verbs mirror a single `Command`'s, each folding the pipefail outcome
(below). Every verb has an `a`-prefixed asyncio twin:

| Sync | Async | Returns | A failing stage is… |
|---|---|---|---|
| `output()` | `aoutput()` | `ProcessResult` | …reported in the result (code/stderr/`program` of the first unclean stage) |
| `output_bytes()` | `aoutput_bytes()` | `BytesResult` | …same, with the last stage's stdout captured as raw `bytes` |
| `run()` | `arun()` | trimmed final stdout (`str`) | …raised as that stage's exception |
| `exit_code()` | `aexit_code()` | `int` | …its attributed code |
| `probe()` | `aprobe()` | `bool` | `0` → `True`, `1` → `False`, else raises |

`output()`/`output_bytes()` capture a non-zero exit, timeout, or signal as
**data** on the result; `run()`/`exit_code()`/`probe()` raise per the pipefail
attribution. An exception that isn't a clean process outcome — a stage that
couldn't be *spawned*, broken plumbing — surfaces as `ProcessError`, never as a
mere non-zero exit. See [Running commands](commands.md) for the full error model
and the structured exception fields.

## The pipefail outcome

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 **first** stage
  that didn't exit cleanly (non-zero, signal-killed, or timed out) — or from the
  last stage when every stage succeeded.

```python
result = (
    Command("cat", ["data.txt"])
    | Command("grep", ["ERROR"])      # suppose grep exits 2 (bad pattern)
    | Command("wc", ["-l"])
).output()

result.stdout        # whatever wc managed to print (the last stage)
result.code          # 2 — grep, the first unclean stage
result.program       # "grep"
result.is_success    # False
```

`run()` requires **every** stage to succeed and returns the trimmed final
stdout; if any stage exits uncleanly it raises that stage's exception
(`NonZeroExit`, `Timeout`, or `Signalled`) carrying that stage's code, stderr,
and `program`. So the chain above would raise `NonZeroExit(code=2, program="grep")`.

One honest edge: in the `producer | head` shape, a downstream that stops reading
early (`head` exits after one line and closes the pipe) leaves the producer to
die on a **broken pipe** at its next write. Under strict pipefail that counts as
the producer's failure — unless that stage was built with
`.unchecked_in_pipe()`, which exempts it from pipefail attribution (its
unclean exit, including a `SIGPIPE`, is skipped when the chain decides what to
report, and never shields a *checked* stage's own failure):

```python
top = (
    Command("producer").unchecked_in_pipe()   # SIGPIPE from `head` closing early is expected
    | Command("head", ["-1"])
).run()
```

Outside a `Pipeline`, `unchecked_in_pipe()` is a no-op — a single run's status
is already plain data on its own `ProcessResult`, and `ensure_success()` stays
opt-in.

## stdin and stdout at the ends; per-stage env/cwd

The ends of the chain behave like a single `Command`:

- The **first** stage's stdin source is honored — set `stdin_text(...)` /
  `stdin_bytes(...)` on it to feed the whole chain from a string or bytes.
- **Inner** stages read from the pipe, full stop; any stdin set on them is
  ignored. Only the last stage's stdout reaches you; inner stderr is captured
  per-stage for the pipefail diagnostics.

```python
# Feed the chain from a string; inner stages read the pipe.
unique = (
    Command("sort").stdin_text("b\na\nb\nc\n")
    | Command("uniq")
    | Command("wc", ["-l"])
).run()
print(unique)        # "3"
```

Per-stage `env` and `cwd` are plain `Command` builders — set them on each stage
**before** piping:

```python
counts = (
    Command("git", ["log", "--format=%an"]).cwd("/srv/repo")
    | Command("sort")
    | Command("uniq", ["-c"])
).run()
```

## Timeouts bound the chain

`Pipeline.timeout(seconds)` bounds the **whole** chain. At the deadline the
teardown fans across every stage's sub-group; the result reports `timed_out`
(and `run()` raises `Timeout`). Its `program` is the composite pipeline name,
with every stage joined by `" | "`, because no individual stage caused this
chain-level deadline. Durations are floats of seconds:

```python
result = (
    Command("producer")
    | Command("consumer")
).timeout(30.0).output()

result.timed_out     # True if the 30s deadline fired
```

Unlike a single command's captured timeout, a timed-out pipeline yields **no
partial stdout** — the chain is run-to-completion or nothing. A per-stage
`Command.timeout(...)` first reaps that stage's whole subtree; the resulting
checked `Timeout` then tears down the remaining stage sub-groups and is
attributed to the timed-out stage under pipefail. See
[Timeouts & cancellation](timeouts-and-cancellation.md); cancelling an awaited
`arun()`/`aoutput()` reaps the whole chain's tree the same way, and so does
firing a `CancellationToken` wired with `Pipeline.cancel_on(token)` — **gap-fill**
here, not override: a stage with its own explicit `Command.cancel_on(...)` keeps
it, only stages without one pick up the pipeline-level token.

## Binary tails

For a chain that ends in a binary producer (`... | gzip`), capture the last
stage's stdout raw with `output_bytes()` — its `stdout` is `bytes`, while stderr
stays decoded text:

```python
blob = (Command("cat", ["big.txt"]) | Command("gzip")).output_bytes().stdout
# blob is bytes — the gzip stream
```

The pipeline runs to completion and buffers the tail; this is a captured result,
not a streaming splice.

## Limitations

- **Run-to-completion only.** A `Pipeline` has no `astart()` and no
  line-streaming surface — it consumes its last stage in full to fold the
  pipefail outcome. Stream a *single* [Command](commands.md) when you need
  incremental output, or run the pipeline inside a [process group](process-groups.md)
  alongside other handles.
- **No `output_limit` of its own.** A pipeline can't cap retained output the way
  a single `Command` can. Bound a flooding chain with `.timeout(...)`; cap a
  single noisy stage by running it on its own with `output_limit(...)` first.

---

Next: [Running commands](commands.md) ·
[Process groups](process-groups.md) ·
[Timeouts & cancellation](timeouts-and-cancellation.md) ·
[Cookbook](cookbook.md) · [Platform support](platforms.md)

---

# Timeouts & cancellation

[‹ docs index](./)

Two ways a run can end early, with two different philosophies:

- a **timeout** is part of the run's contract, so its expiry is *data* — captured
  on the capture verbs, raised on the success verbs;
- a **cancellation** is an *abandonment* — the caller changed its mind, so the
  run's tree is torn down and there is no result to inspect. Sync →
  `KeyboardInterrupt`; async → `asyncio.CancelledError`.

The one thing to internalize first: the **same deadline** surfaces differently
*by verb* — captured as `timed_out` on the capture verbs, raised as `Timeout` on
the success verbs. Cancellation is never captured: it is always terminal.

- [Setting a timeout](#setting-a-timeout)
- [Idle (inactivity) timeout](#idle-inactivity-timeout)
- [Graceful timeout](#graceful-timeout)
- [Interrupting a blocked sync call (Ctrl+C)](#interrupting-a-blocked-sync-call-ctrlc)
- [Cancelling an awaited async run](#cancelling-an-awaited-async-run)
- [Timeout vs. cancellation](#timeout-vs-cancellation)
- [Readiness-probe timeouts are separate](#readiness-probe-timeouts-are-separate)

## Setting a timeout

`.timeout(seconds)` bounds the whole run and kills the **entire process tree** at
the deadline — a wrapper script's grandchildren die too, not just the direct
child. Durations are plain floats of seconds.

Where the expiry lands depends only on the verb:

```python
from processkit import Command

# Capture verbs: the deadline is DATA. The run does not raise.
result = Command("slow-tool").timeout(5.0).output()
if result.timed_out:
    print("killed at the deadline; partial output:", result.stdout)

# Success verbs: the deadline is an ERROR.
Command("slow-tool").timeout(5.0).run()   # raises Timeout on expiry
```

Async is identical with the `a`-prefixed verbs:

```python
result = await Command("slow-tool").timeout(5.0).aoutput()   # result.timed_out
```

| Verb | Deadline expiry becomes |
|---|---|
| `output()` / `aoutput()`, `output_bytes()` / `aoutput_bytes()` | a result with `result.timed_out == True`, `result.code == None`, partial output kept |
| `run()` / `arun()`, `exit_code()`, `probe()` | raises `Timeout` (partial output attached) |

The `Timeout` exception carries structured fields — `program`,
`timeout_seconds`, `stdout`, `stderr` — so a hung tool's last words survive the
kill:

```python
from processkit import Timeout

try:
    Command("slow-tool").timeout(5.0).run()
except Timeout as e:
    print(e.program, e.timeout_seconds)
    print("last output before the kill:", e.stderr)
```

`Timeout` is **also** a builtin `TimeoutError`, so `except TimeoutError` catches
it too — handy for callers that don't import the processkit hierarchy.

*Deeper: [Running commands](commands.md) for the full verb surface.*

## Idle (inactivity) timeout

`.timeout(...)` bounds *total* runtime; `.idle_timeout(seconds)` bounds a
**silent gap** instead — it tears the child down if it produces no stdout/stderr
**line** for that long. This is the "hung tool" case a wall-clock timeout handles
poorly: a legitimately long job (a build, a test suite, a data export) keeps
printing progress, so you can bound its *silence* tightly without guessing a
generous ceiling for its total runtime. The two **compose** — set both, and
whichever threshold is reached first wins.

Idle-timeout fires as a **distinct** `IdleTimeout` exception — a `ProcessError`
sibling of `Timeout`, deliberately **not** the wall-clock `timed_out`/`Timeout`
signal — so "the child went silent" and "the run took too long overall" stay
tellable apart, and the captured `timed_out` contract is untouched (an
idle-timeout never sets it):

```python
from processkit import Command, IdleTimeout

# A build that keeps printing is fine; one that hangs silently for 30s is killed.
proc = Command("./flaky-build").idle_timeout(30.0).start()
try:
    async with proc:
        async for event in proc.output_events():
            print(event.text)          # live progress, line by line
except IdleTimeout as e:
    print(f"no output for {e.idle_timeout_seconds}s — killed the hung build")
```

Idle monitoring rides the **per-line output channel**, so it is enforced on the
**streaming/interactive surface** — `start()`/`astart()` +
`stdout_lines()`/`output_events()` (piped stdout, the default). It is **not**
enforced by the one-shot capture verbs (`output`/`run`/`exit_code`/`probe` and
their `a`-twins), `Pipeline`, or `Supervisor`: those run entirely inside the
Rust core, which has no native idle-timeout to observe per-line activity mid-run,
so honoring it there awaits upstream support. The setting is carried on the
command regardless, so nothing breaks if you set it and use a one-shot verb — it
simply doesn't fire there.

Because monitoring needs line events, a **redirected stdout** cannot be watched —
and that combination is *diagnosed*, not silently dropped. Under `stdout_file()`
/ `stdout("inherit")` / `stdout("null")` the streaming verbs already raise
`ProcessError` (`"stdout is not piped …"`) at setup, so an `idle_timeout()` on a
redirected stdout surfaces there. `stderr_file()` is the asymmetric case: it
leaves stdout piped, so idle monitoring keeps working on the stdout channel while
stderr goes to the file.

From the CLI, `python -m processkit run --idle-timeout SECONDS` applies the same
mechanism and exits **123** (distinct from `--timeout`'s 124) on a silent child —
see the [CLI reference](cli.md#--idle-timeout-a-silence-watchdog).

*Deeper: [Running commands](commands.md#idle_timeout--a-silence-watchdog) for the
builder's boundaries.*

## Graceful timeout

By default the deadline **hard-kills** at once. `.timeout_grace(g)` instead asks
the tree to clean up first: at the deadline it sends the terminate signal, gives
the tree up to `g` seconds to exit, then hard-kills whatever is still alive.

```python
# At 30s: send SIGTERM, wait up to 5s, then SIGKILL the tree.
Command("server").timeout(30.0).timeout_grace(5.0).run()
```

Choose the first signal with `.timeout_signal(name)` — one of `term` (default),
`kill`, `int`, `hup`, `quit`, `usr1`, `usr2`:

```python
Command("nginx").timeout(30.0).timeout_signal("quit").timeout_grace(5.0).run()
```

A signal-handling child that exits early ends the grace early. `result.timed_out`
is `True` (or `Timeout` is raised) regardless of whether the child obeyed the
signal or was hard-killed after the grace — the **deadline** is what fired, not
the manner of death. This is the same SIGTERM → wait → SIGKILL tier that
[Process groups](process-groups.md) use for graceful shutdown.

Mind the platform asymmetry: Windows has no signal tier, so `timeout_grace` /
`timeout_signal` are accepted but the deadline kills the job atomically. See
[Platform support](platforms.md).

## Interrupting a blocked sync call (Ctrl+C)

A synchronous verb blocked on a child honors **Ctrl+C** (SIGINT). Instead of
hanging until the child decides to exit, it raises `KeyboardInterrupt` *promptly*
and tears down the run's process tree on the way out:

```python
try:
    Command("long-batch-job").run()     # blocks here…
except KeyboardInterrupt:
    # Ctrl+C: the child tree is already reaped; the exception is re-raised at once.
    print("interrupted by the user")
```

This holds for every sync verb (`output()`, `run()`, `exit_code()`, `probe()`,
…) — no orphaned grandchildren are left behind.

> **Main-thread only.** CPython delivers signals to the main thread, so this
> prompt `Ctrl+C` interruption works only when the sync verb runs on the main
> thread. A sync verb called from a `threading.Thread` (more tempting on a
> free-threaded build) blocks until the child exits — it cannot observe the
> signal. Off the main thread, prefer the async API and cancel the task.

The async surface uses task cancellation instead, below.

## Cancelling an awaited async run

Cancelling the task awaiting a run — directly with `task.cancel()`, or via
`asyncio.wait_for(...)` / `asyncio.timeout(...)` — tears down the **whole process
tree** and surfaces as `asyncio.CancelledError`:

```python
import asyncio
from processkit import Command

# Direct cancel: stop a run from elsewhere.
task = asyncio.ensure_future(Command("long-export").aoutput())
# ... later — a shutdown handler, a sibling failure, a UI action ...
task.cancel()                  # the tree is reaped; awaiting `task` raises CancelledError

# Caller-side deadline via asyncio: the run is cancelled, then re-raised to you.
try:
    await asyncio.wait_for(Command("long-export").arun(), timeout=10)
except TimeoutError:           # asyncio re-raises the cancellation as TimeoutError
    ...                        # the run's process tree was already torn down
```

`asyncio.wait_for` (and `asyncio.timeout`, 3.11+) cancel the inner run exactly
like `task.cancel()`, then translate the cancellation into a builtin
`TimeoutError` at the `await` boundary — so *inside*, the run was cancelled, even
though *you* catch `TimeoutError`. Either way the tree is gone.

**Cancellation surfaces as `asyncio.CancelledError`** when you cancel through
asyncio itself, as above (a `BaseException`, deliberately not a
`ProcessError`) — there is no separate processkit exception on this path.

## Cancelling with an explicit `CancellationToken`

For a cancel switch that isn't tied to one asyncio task — shared across
several runs, fired from sync code, or from a different task entirely — wire
a `CancellationToken` instead:

```python
from processkit import Command, Cancelled, CancellationToken

token = CancellationToken()
cmd = Command("long-export").cancel_on(token)

# elsewhere — a signal handler, a UI action, another task:
token.cancel()

try:
    await cmd.arun()   # (or cmd.run() from sync code)
except Cancelled:
    ...  # the whole tree was already torn down
```

Unlike asyncio cancellation, this surfaces as `Cancelled` — a `ProcessError`
subclass carrying `.program`, catchable alongside every other processkit
exception, on *either* the sync or async surface. A cancelled token stays
cancelled forever (never use it to mean "pause" — see
[`ProcessGroup.suspend()`/`resume()`](process-groups.md) for that), and a
cancelled run is never retried (`Command.retry()`) or restarted
(`Supervisor`) — another attempt could only fail the same way.

`Command.cancel_on()` **replaces** any previously set token (last write
wins); the *gap-fill* containers `Pipeline.cancel_on()` and `CliClient`'s
`default_cancel_on=` leave an explicit per-stage/per-command token intact,
only filling in where none was set — the same gap-fill convention
`default_timeout` uses. `token.child_token()` derives a token cancelled
automatically when the parent fires, but cancellable independently — for
scoping a broader shutdown token down to one operation while still reacting
to the parent.

## Timeout vs. cancellation

The two can both stop a run, but they are different kinds of event:

| | Timeout | asyncio cancellation | `CancellationToken` |
|---|---|---|---|
| Meaning | the deadline was part of the contract | the caller abandoned the run | an explicit cancel switch fired |
| Capture verbs (`output*`) | captured as `result.timed_out` | terminal — no result | terminal — no result |
| Success verbs (`run`/`exit_code`/`probe`) | raises `Timeout` | terminal — no result | raises `Cancelled` |
| Sync surface | `Timeout` | `KeyboardInterrupt` | `Cancelled` |
| Async surface | `Timeout` | `asyncio.CancelledError` | `Cancelled` |

A timeout still leaves something to inspect on the capture verbs; a cancellation
never does — the run was abandoned, so there is nothing to report but the
cancellation itself. **When a cancel and a timeout race on the same run,
cancellation wins:** you asked the run to stop mattering, so no `timed_out`
result is synthesized.

On a shared [ProcessGroup](process-groups.md) handle, a timeout or cancellation
that hits one child kills **that child only** — the group's siblings keep
running.

## Readiness-probe timeouts are separate

The `timeout` on the readiness helpers — `wait_until`, `wait_for_port`,
`wait_for_line` — is a **different deadline** from a run timeout. It bounds how
long you wait for a *condition*, and on expiry it raises `WaitTimeout` (also a
builtin `TimeoutError`) **without killing the child** — the process keeps
running; only your wait gave up:

```python
from processkit import wait_for_port

await wait_for_port("127.0.0.1", 8080, timeout=10)   # TimeoutError if not listening in 10s
```

Because `Timeout` is itself a `TimeoutError`, a single `except TimeoutError`
catches both a run timeout and a readiness timeout — but only the run timeout
reaped a tree.

*Deeper: [Streaming & interactive I/O](streaming.md).*

## Bounding pipelines & tuning group shutdown

- A [pipeline](pipelines.md) bounds the **whole chain** with
  `Pipeline.timeout(seconds)`; the same captured-vs-raised rule applies to
  whichever verb you finish it with.
- A [ProcessGroup](process-groups.md)'s graceful teardown timing is set at
  construction with `shutdown_grace=` and `escalate_to_kill=`, independent of
  any per-run timeout. Note: cancelling an in-flight `await group.ashutdown()` (or
  an `async with` exit) falls back to an immediate hard kill — the tree is still
  reaped (no orphan), but the *graceful* signal-then-wait window is skipped.

## Keeping a flaky thing alive

A timeout stops a single run; it does not restart anything by itself. For a
single command replayed on transient failure (including a timeout expiry),
see [`Command.retry(retry_if, ...)`](commands.md#retrying-a-run) (default is
`retry_never()` — no retries unless opted in). For a *service* kept alive
across crashes — a different, non-exclusive concern from per-command retry —
that is [Supervision](supervision.md) — `Supervisor(...)` with a restart
policy and backoff.

---

Next: [Supervision](supervision.md) ·
[Streaming & interactive I/O](streaming.md) ·
[Async runtimes & event loops](event-loops.md) ·
[Process groups](process-groups.md) ·
[Cookbook](cookbook.md)

---

# Supervision

[‹ docs index](./)

A [`timeout`](timeouts-and-cancellation.md) or a cancelled task *bounds one run* — it
caps a single invocation, and then it's over. A `Supervisor` answers the opposite
need: *keep a long-lived child alive*. It runs a [`Command`](commands.md), and
whenever that command exits it restarts it per policy — with a bounded restart count
and exponential, jittered backoff — until a stop condition is met. Think of it as a
pocket `systemd`/`runit`: a keeper loop you can drop into a script. It is
platform-agnostic.

- [A supervised server](#a-supervised-server)
- [Restart policies](#restart-policies)
- [Backoff and jitter](#backoff-and-jitter)
- [Stopping: the predicate](#stopping-the-predicate)
- [Reading the outcome](#reading-the-outcome)
- [Liveness health checks](#liveness-health-checks)
- [Sync vs async](#sync-vs-async)

## A supervised server

The supervisor takes a normal `Command` — build it with all the usual knobs (args,
`env`, `cwd`, `timeout`, …) and they apply to *every* restart:

```python
from processkit import Command, Supervisor

outcome = Supervisor(
    Command("my-server", ["--port", "8080"]).env("LOG", "info"),
    restart="on_crash",        # the default
    max_restarts=5,            # default: unlimited
    backoff_initial=0.2,       # seconds; base delay (default 0.2)
    backoff_factor=2.0,        # multiplier (default 2.0)
    max_backoff=30.0,          # seconds; cap (default 30.0)
).run()                        # or: await ....arun()

print(outcome.restarts, outcome.stopped)
```

Each restart is one full captured run of the command. The one-shot stdin caveat
applies from the second run onward — see [Running commands](commands.md). Leave a
knob unset (`None`) and the crate default shown above is used.

Contrast this with a one-shot run wrapped in a hand-rolled `while True:` loop: you'd
reimplement backoff, jitter, and the stop gates yourself. The supervisor *is* that
loop, written once and correctly.

## Restart policies

`restart=` decides what is worth restarting. A **crash** is any run that is not a
success — an exit code outside the accepted set (default `{0}`, widened by the
command's [`success_codes`](commands.md)), a timeout, or a signal-kill:

| `restart=` | Restarts after… |
|---|---|
| `"on_crash"` *(default)* | crashes only; a clean exit ends supervision (`stopped == "policy_satisfied"`) |
| `"always"` | every completed run, clean or not — pair with `stop_when=`/`max_restarts=` or it loops forever |
| `"never"` | nothing: one run, reported as-is |

Because `success_codes` defines success, a command built with `.success_codes([0, 2])` that
exits `2` is *clean*, so `"on_crash"` treats it as a satisfied policy, not a crash.

## Backoff and jitter

Between restarts the supervisor sleeps. The *n*-th restart (0-based) waits:

```text
delay(n) = min(backoff_initial × backoff_factor**n, max_backoff) × jitter
```

with `jitter` drawn uniformly from `[0.5, 1.5)` per restart. With the defaults
(`0.2`, `2.0`, cap `30.0`):

```text
restart #0 → ~0.2s   #1 → ~0.4s   #2 → ~0.8s … #7 → ~25.6s   #8+ → 30.0s (cap)
```

Jitter is **on by default** so a fleet of supervised workers knocked over by one
incident doesn't stampede back in lockstep. Pass `jitter=False` for deterministic
delays (handy in tests). `backoff_factor` is a finite multiplier `>= 1.0`, and it
rides along with `backoff_initial` — set the base to opt into a custom schedule.

## Stopping: the predicate

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

1. **`stop_when=`** — a callable handed each run's [`ProcessResult`](commands.md);
   returning `True` ends supervision *regardless of policy* (`stopped ==
   "predicate"`). The classic "exit 0 is done" under `restart="always"`:

   ```python
   outcome = Supervisor(
       Command("flaky-worker"),
       restart="always",
       stop_when=lambda r: r.code == 0,   # stop on the first clean exit
   ).run()
   ```

2. **The policy** — `"on_crash"` stops on a clean exit; `"never"` stops after one run.
3. **`give_up_when=`** — a callable consulted only for a crash the policy would
   otherwise restart, ahead of `max_restarts=` and the storm guard. It classifies a
   *permanent* failure so supervision gives up instead of restarting forever. It
   receives one argument mirroring the crate's `GiveUpAttempt` sum type, dispatched
   with `isinstance`: a `ProcessResult` for a crashed run that produced a result
   (classify by e.g. `attempt.code`), or a `ProcessError` subclass for a launch that
   never produced one (classify by e.g. `isinstance(attempt, ProcessNotFound)` for a
   missing binary). Returning `True` for a crash verdict stops with
   `outcome.stopped == "gave_up"`; a launch-failure verdict has no result to report
   and surfaces the classified error directly from `run()`/`arun()`.
4. **`max_restarts=n`** — at most *n* restarts (= *n + 1* total runs); an exhausted
   budget reports the last result (`stopped == "restarts_exhausted"`).
   `max_restarts=0` means exactly one run.

Two honest caveats about `stop_when=`:

- **Inspect the passed result — don't call a synchronous run verb inside it.** Read
  `r.code` / `r.is_success` / `r.stdout` off the argument. The predicate runs *on*
  the runtime, so a nested sync call (`Command(...).run()`/`.probe()`/…) can't drive
  the runtime again — it raises a clear `ProcessError` ("cannot call a synchronous
  processkit verb from inside an async context or a callback"). That error is then
  surfaced through the unraisable hook (next bullet), so the supervisor keeps going
  rather than stopping — i.e. a sync verb in the predicate is a no-op stop gate, not
  a crash. If you must run a check, precompute it before the supervised run, or use
  the result handed to the predicate.
- **A predicate that raises does not stop supervision.** The exception is surfaced
  through Python's [unraisable hook](https://docs.python.org/3/library/sys.html#sys.unraisablehook)
  and treated as "don't stop" — a buggy predicate degrades to *keep going*, it does
  not crash the supervisor.

## Reading the outcome

`run()` (and `arun()`) resolve to a `SupervisionOutcome`:

```python
outcome.final_result   # ProcessResult of the LAST run
outcome.restarts       # restarts performed (run #1 is not a restart)
outcome.stopped        # "policy_satisfied" | "predicate" | "restarts_exhausted"
                        # | "gave_up" | "unhealthy" | "unknown" (forward-compat
                        # fallback, not emitted by the pinned crate version)
outcome.storm_pauses   # how many failure-storm pauses were taken (see below)
outcome.liveness_kills # how many wedged incarnations a health check force-killed
                        # (see "Liveness health checks"; 0 unless one is enabled)
```

A returned outcome means supervision *concluded*, not that the child succeeded —
inspect `final_result` (e.g. `outcome.final_result.is_success`) for the child's own
verdict.

`final_result.stdout` is the **last run's** output, and for a long-lived
supervised process it is kept to a bounded tail (the most recent ~1000 lines)
rather than buffered in full — so `final_result.truncated` may be `True`. Treat it
as a diagnostic tail, not a complete transcript. Widen or re-bound the cap with
`Supervisor`'s own `capture_max_bytes=`/`capture_max_lines=`/`capture_on_overflow=`
constructor kwargs (mirroring `Command.output_limit`'s kwargs — set at least one
of the two cap sizes), or give the base `Command` an explicit
[`output_limit`](commands.md) (respected as-is) before wrapping it in a
`Supervisor`; otherwise stream the process yourself. `capture_max_bytes` uses the
same unit as `output_limit(max_bytes=...)`, which depends on the overflow mode:
`capture_on_overflow` defaults to `"drop_oldest"`, where the cap bounds the
retained decoded line content (unchanged in processkit 3.0.0); pass
`capture_on_overflow="error"` and it becomes a fail-loud ceiling on the raw bytes
read from the pipe instead. See
[what `max_bytes` counts](commands.md#what-max_bytes-actually-counts).

## The failure-storm guard

Backoff slows individual restarts; the **failure-storm guard** distinguishes "fails
once in a blue moon" from "crash-looping" and takes a single collective pause
instead of hammering restarts at backoff speed. It is **off by default** — enable
it by setting `storm_pause`:

```python
outcome = Supervisor(
    Command("flaky-worker"),
    restart="on_crash",
    storm_pause=30.0,          # ENABLES the guard: pause 30s when a storm is detected
    failure_threshold=5.0,     # decaying failure score that trips the pause (optional)
    failure_decay=60.0,        # the score halves every 60s (optional)
).run()

if outcome.storm_pauses:
    log.warning("flaky-worker crash-looped: %d storm pauses", outcome.storm_pauses)
```

Each failure adds to a score that decays every `failure_decay`; once it crosses
`failure_threshold` the supervisor takes one `storm_pause` and increments
`outcome.storm_pauses`. With `storm_pause` unset, the guard is inactive and
`storm_pauses` stays `0` — only the per-restart `backoff` and the lifetime
`max_restarts` cap apply.

A `Supervisor` is single-shot: `run()`/`arun()` consume it, so build a fresh one to
supervise again.

## Liveness health checks

Restart policies react to a process that *exits*. But a long-lived service can wedge
*without* exiting — a deadlocked server, a stuck event loop, a worker that stopped
answering — and an exit-driven policy would happily call that "still running" forever.
A **liveness health check** closes that blind spot: an opt-in probe, re-run on a fixed
cadence, that force-restarts the child when it stops looking healthy. It is the
`Supervisor`'s take on systemd's `WatchdogSec` or a container liveness probe, and it is
**off by default**.

```python
import socket

from processkit import Command, Supervisor


def is_healthy() -> bool:
    # A fast, non-blocking liveness probe: can we still reach the server's port?
    try:
        with socket.create_connection(("127.0.0.1", 8080), timeout=0.5):
            return True
    except OSError:
        return False


outcome = Supervisor(
    Command("my-server", ["--port", "8080"]),
    health_check=is_healthy,       # sync () -> bool; True == healthy
    health_check_interval=5.0,     # seconds between probes — REQUIRED with health_check
    health_check_failures=3,       # consecutive failures before a force-restart (default 3)
    max_restarts=10,
).run()

print(outcome.liveness_kills)      # wedged incarnations that were force-killed
```

`health_check=` is a **synchronous** callable `() -> bool` — *not* a coroutine —
returning `True` for healthy. It is the liveness twin of `stop_when=`/`give_up_when=`
and runs on the supervision runtime, so keep it fast and non-blocking (a quick socket
connect, an HTTP `/healthz` GET, a heartbeat-file check); a slow probe merely stretches
the effective cadence. `health_check_interval=` is its **required** partner — the crate
takes probe and cadence together, so passing either one alone raises `ValueError`. The
first probe fires one interval *after* an incarnation starts (startup grace), then
repeats for that incarnation's life; a healthy child is never disturbed.

A probe that fails `health_check_failures=` checks **in a row** (default `3`; one
healthy probe resets the streak, so a single blip is forgiven) force-restarts the
child. A failed streak is treated **exactly like a crash**: it flows through the restart
policy, `backoff`, the storm guard, and `max_restarts` just as a real crash would — but
it does *not* consult `stop_when=` (there is no cleanly-completed run to judge). So:

- under `restart="never"`, the single force-killed run is the final one, reported as
  `outcome.stopped == "unhealthy"`;
- under a restart-wanting policy (`"on_crash"`/`"always"`), it restarts and surfaces —
  if it ends supervision at all — as the usual `"gave_up"` / `"restarts_exhausted"`.

Either way each force-kill is counted in `outcome.liveness_kills` (and, because it
counts as a crash, is also reflected in `outcome.restarts` when the policy restarted
it), and the final synthetic result is a non-success signal-kill. A probe that *raises*
or returns a non-`bool` cannot answer "healthy": it is treated as unhealthy **and** its
error is surfaced to the caller from `run()`/`arun()` — not swallowed into a spurious
liveness kill — the same fail-loud contract as `stop_when=`/`give_up_when=`.

## Sync vs async

Both verbs return the same `SupervisionOutcome`; pick the one that matches your call
site. Durations are plain floats of seconds throughout.

```python
# Synchronous — blocks the calling thread (Ctrl+C interrupts it):
outcome = Supervisor(Command("my-server"), max_restarts=3).run()

# Asyncio — awaitable, integrates with the event loop:
outcome = await Supervisor(Command("my-server"), max_restarts=3).arun()
```

**`arun()` is lazy — nothing runs until you `await` it.** Like every
`a`-prefixed verb, `arun()` returns an awaitable that starts no supervision
until it is first awaited. So an `arun()` you build but never await — a
dropped awaitable, or `asyncio.ensure_future(sv.arun())` you never follow up
on — starts no restart loop at all; dropping it releases the supervisor and
every `stop_when=`/`give_up_when=` callback it captured, rather than pinning
them (and whatever they close over) for the life of the interpreter. The flip
side is that an unawaited `arun()` never supervises anything, so `await` what
it returns — and, for an unbounded `restart="always"`, give it a
`max_restarts=`/`stop_when=` so supervision also has a defined end:

```python
# Bounded and awaited — runs, then stops after at most 5 restarts:
outcome = await Supervisor(Command("flaky-worker"), restart="always", max_restarts=5).arun()

# Backgrounded — keep the task and await it, so supervision actually runs:
task = asyncio.ensure_future(Supervisor(Command("flaky-worker"), restart="always", max_restarts=5).arun())
outcome = await task
```

A `Supervisor` keeps *one* command alive across restarts; to contain a whole *tree*
of processes under kill-on-exit semantics, reach for a
[process group](process-groups.md) instead. To exercise restart/stop logic without
spawning anything real, see [Testing your code](testing.md), and for the broader
task-oriented recipes, the [Cookbook](cookbook.md).

---

Next: [Timeouts & cancellation](timeouts-and-cancellation.md) ·
[Process groups](process-groups.md) · [Cookbook](cookbook.md)

---

# Testing your code

[‹ docs index](./)

Code that shells out is miserable to test — unless the subprocess sits behind a
seam. In **processkit-py** that seam is a plain object: a *runner*. Write your
code against a `runner` parameter, call its verbs, and never name a concrete
runner inside the logic. In production you pass `Runner()` — the real thing. In
tests you pass a double — a `ScriptedRunner` with canned replies, a replaying
`RecordReplayRunner`, a `RecordingRunner` spy, or a `DryRunRunner` that only
renders each command — and no subprocess is ever spawned. The objects that come
back are genuine `ProcessResult` / `RunningProcess` values, so the code under
test can't tell the difference.

> The doubles — `ScriptedRunner`, `RecordReplayRunner`, `RecordingRunner`,
> `DryRunRunner`, the `Reply` builder, and the `Invocation` record — live in the
> **`processkit.testing`** submodule (mirroring the crate's own
> `processkit::testing` split). `Runner` and the `ProcessRunner` protocol stay
> on the top-level `processkit` — they are production code, not test
> scaffolding.

- [The runner seam](#the-runner-seam)
- [The pytest plugin: ready-made fixtures](#the-pytest-plugin-ready-made-fixtures)
- [Scripting replies: ScriptedRunner](#scripting-replies-scriptedrunner)
- [Scripted streaming: a live handle, no child](#scripted-streaming-a-live-handle-no-child)
- [Record/replay cassettes: RecordReplayRunner](#recordreplay-cassettes-recordreplayrunner)
- [Asserting on calls: RecordingRunner](#asserting-on-calls-recordingrunner)
- [Rendering commands without running: DryRunRunner](#rendering-commands-without-running-dryrunrunner)
- [Wrapping a CLI tool: CliClient](#wrapping-a-cli-tool-cliclient)

## The runner seam

`Runner()` is the real implementation; every double exposes the **same verb
surface**, so swapping one in is the whole technique. Each verb takes a
`Command` and returns the same type the bare `Command` methods do:

| Sync | Async | Returns | Notes |
|---|---|---|---|
| `output(cmd)` | `aoutput(cmd)` | `ProcessResult` | full result; a non-zero exit is *data*, not a raise |
| `output_bytes(cmd)` | `aoutput_bytes(cmd)` | `BytesResult` | raw-bytes stdout |
| `run(cmd)` | `arun(cmd)` | `str` | trimmed stdout; raises on failure |
| `exit_code(cmd)` | `aexit_code(cmd)` | `int` | the raw exit code |
| `probe(cmd)` | `aprobe(cmd)` | `bool` | exit 0 as a boolean |
| `start(cmd)` | `astart(cmd)` | `RunningProcess` | a live handle for streaming / readiness probes |

Write production code against the seam; hand it the real runner there:

```python
from processkit import Command, ProcessRunner, Runner

def current_branch(runner: ProcessRunner) -> str:
    return runner.run(Command("git", ["branch", "--show-current"]))

# Production: the real runner, which actually spawns git.
branch = current_branch(Runner())
```

Annotate the injected runner as **`ProcessRunner`** — a `typing.Protocol` that
describes the verb surface. `Runner`, `ScriptedRunner`, `RecordReplayRunner`,
`RecordingRunner`, and `DryRunRunner` all satisfy it structurally, so the
annotation type-checks (strict `mypy`) against any of them. A custom double can implement the capture/check verbs directly; the
streaming `start`/`astart` verbs must return a `RunningProcess` (no public
constructor), so reach for `ScriptedRunner` when you need a streaming double rather
than building one from scratch. `CliClient` is also a `ProcessRunner`: its sync
and async capture/check verbs accept either per-call `Args` (combined with its
bound program) or a `Command` (whose explicit settings win over client
defaults). It is not a `StreamingRunner`, because it has no `start`/`astart`.

The sync and async surfaces are twins (`run` ↔ `arun`), so async code injects
the very same runner objects and awaits the `a`-prefixed verbs.

> These doubles are the *real* ones — they return genuine `ProcessResult` /
> `RunningProcess` objects, so the code under test behaves identically. (The Rust
> crate also ships a `mock` Cargo feature — a `mockall`-generated mock of its
> runner trait — but that is for *Rust* tests; it has no Python use, so the binding
> does not enable it. You get your doubles here, not from a mocking library.)

*Deeper: the verb vocabulary and what each return type carries — [Running commands](commands.md).*

## The pytest plugin: ready-made fixtures

Installing processkit registers a **pytest plugin** — a `pytest11` entry point,
autoloaded in every pytest session, with nothing to add to your `conftest.py`. It
turns the doubles above into fixtures, so wiring one into a test is a single
parameter rather than a line of construction. Each fixture yields one of the
doubles below, so it satisfies the same `ProcessRunner` seam and spawns no real
process:

| Fixture | Yields | Notes |
|---|---|---|
| `scripted_runner` | a fresh [`ScriptedRunner`](#scripting-replies-scriptedrunner) | teach it replies with `.on()` / `.when()` / `.fallback()` |
| `recording_runner` | a [`RecordingRunner`](#asserting-on-calls-recordingrunner) spy | replies `Reply.ok("")` (a clean exit 0, empty stdout — the neutral default) to every call and records each one |
| `record_replay_runner` | a [`RecordReplayRunner`](#recordreplay-cassettes-recordreplayrunner) cassette | replay by default, record on demand — see below |
| `dry_run_runner` | a fresh [`DryRunRunner`](#rendering-commands-without-running-dryrunrunner) | renders each command to text instead of running it |

```python
from processkit import Command
from processkit.testing import Reply

def latest_commit(runner):
    return runner.run(Command("git", ["rev-parse", "HEAD"]))

def test_latest_commit(scripted_runner):
    scripted_runner.on(["git", "rev-parse"], Reply.ok("deadbeef"))
    assert latest_commit(scripted_runner) == "deadbeef"     # no git spawned
```

### The cassette fixture: record ↔ replay

`record_replay_runner` binds a [cassette](#recordreplay-cassettes-recordreplayrunner)
to the test. Which way it runs is a **switch, off (replay) by default** so CI
never spawns by accident — chosen the way vcr-like tools do it, in precedence
order:

1. `pytest --processkit-record` (CLI flag) forces **record** mode; otherwise
2. the `PROCESSKIT_RECORD` environment variable, when set, decides by its
   truthiness (`1`/`true`/`yes`/`on` → record); otherwise
3. the `processkit_record` ini option (a bool) decides; defaulting to **replay**.

In record mode the cassette is captured against real processes and `save()`d on
teardown; in replay mode it is served offline, never spawning. The file lives
under the test's `tmp_path` by default — set the `processkit_cassette_dir` ini
option (a relative path resolves against the rootdir) to a committed fixtures
directory to **keep** cassettes across runs. Its name is derived deterministically
from the test's node id, so each test gets its own. If replay reports that the
cassette is absent, see
[Troubleshooting](troubleshooting.md#record_replay_runner-cassette-not-found).

The workflow is the usual vcr one — *record once, replay forever*:

```ini
# pytest.ini (or [tool.pytest.ini_options] in pyproject.toml)
[pytest]
processkit_cassette_dir = tests/cassettes
```

```python
import sys
from processkit import Command

def test_offline(record_replay_runner):
    # `pytest --processkit-record` once: spawns for real and writes the cassette.
    # Every run after: served from tests/cassettes/…json, no process spawned.
    out = record_replay_runner.run(Command(sys.executable, ["--version"]))
    assert out.startswith("Python")
```

> Cassettes store `program`/`args`/`cwd`/`stdout`/`stderr` **verbatim** and can
> carry secrets — review one before committing it (see
> [Record/replay cassettes](#recordreplay-cassettes-recordreplayrunner) for the
> full semantics and the redaction boundary).

### The no-real-spawn guard

Mark a test `@pytest.mark.no_real_spawn` and any **real** process spawn through
`Command` / `Pipeline` / `Runner` / `ProcessGroup` inside it fails loudly (via
`pytest.fail`, which no `except` in the code under test can swallow) — so a
forgotten double can't quietly reach the OS:

```python
import pytest
from processkit import Command

@pytest.mark.no_real_spawn
def test_stays_hermetic(scripted_runner):
    scripted_runner.fallback(Reply.ok("ok"))
    assert my_code(scripted_runner) == "ok"   # injected double: fine
    # Command("git", ["status"]).run()        # would fail the test, loudly
```

The marker is registered by the plugin, so it passes `--strict-markers`. Injected
doubles keep working — only the real-spawn primitives are blocked. The interception
replaces those verbs on the compiled classes for the duration of the test (the
reliable seam, since PyO3 forbids subclassing or per-instance patching of them),
which catches a spawn even through a `Command` reference imported before the test
ran. The honest boundary: the injection-point APIs (`CliClient`, `output_all` and
friends, `Supervisor`) reach the OS entirely inside the Rust extension when given
the default real runner, with no Python seam to intercept — so pass them a
test-double `runner=` in a guarded test rather than relying on the guard to catch
their default path.

## Scripting replies: ScriptedRunner

`ScriptedRunner` is the work-horse double: it returns a canned `Reply` for each
command you teach it. Match rules with `.on(prefix, reply)`; add an optional
`.fallback(reply)` for everything else.

```python
from processkit import Command
from processkit.testing import Reply, ScriptedRunner

def current_branch(runner):
    return runner.run(Command("git", ["branch", "--show-current"]))

def test_detects_the_branch():
    runner = ScriptedRunner()
    # Match by program + argument PREFIX (element-wise; the program is the first
    # element). Rules are tried in registration order; first match wins.
    runner.on(["git", "branch", "--show-current"], Reply.ok("main\n"))
    runner.fallback(Reply.ok(""))            # optional catch-all
    assert current_branch(runner) == "main"
```

Build the canned outcomes with the `Reply` factories:

- **`Reply.ok(stdout)`** — exit 0 with this stdout.
- **`Reply.fail(code, stderr)`** — a non-zero exit; `run` / `exit_code` raise `NonZeroExit`, while `output` reports it as data.
- **`Reply.lines([...])`** — exit 0 with the lines joined (and streamed one-by-one on a scripted [`start`](#scripted-streaming-a-live-handle-no-child)).
- **`Reply.timeout()`** — a timed-out run; `run` and the checking verbs raise `Timeout`.
- **`Reply.signalled(signal=None)`** — a signal-killed run; `run` raises `Signalled`.
- **`Reply.pending()`** — parks the call like a hung child; pair it with `asyncio.wait_for` / a `Command.timeout()` to prove your orchestration actually cancels a blocked call.
- **`.with_stdout(text)`** — an instance method that attaches stdout to any reply (e.g. the `CONFLICT …` text git prints on a *failing* merge).
- **`.with_line_delay(seconds)`** — sleep `seconds` before each scripted stdout line on a `start()`/`astart()` run, so a hermetic streaming test can observe genuinely incremental delivery instead of every line arriving at once.

Prefix matching is element-wise over the program name then the arguments, so
`on(["git", "branch"])` matches `git branch --show-current` but not `git
branchx` (and not `hg branch`). An **unmatched command with no fallback raises
a plain `ProcessError`** (not `ProcessNotFound`/`FileNotFoundError` — a miss is
a scripting gap, not a missing *program*) — loud enough that an unexpected
invocation can't slip through a test silently, but distinguishable from a
genuinely missing binary.

Reply each of several successive calls in turn with **`.on_sequence(prefix,
replies)`** — the declarative form for "fail once, then succeed" retry
scenarios: the first matching call gets `replies[0]`, the second `replies[1]`,
and so on; once exhausted, the **last** reply repeats forever.

```python
runner = ScriptedRunner()
runner.on_sequence(["deploy"], [Reply.fail(1, "transient"), Reply.ok("deployed")])
```

For a match that isn't a plain argv prefix, **`.when(predicate, reply)`**
replies with `reply` when `predicate(command)` accepts it — inspecting
`command.cwd`/`command.arguments`/whatever `Command`'s own inspection
accessors expose:

```python
runner = ScriptedRunner()
runner.when(lambda cmd: "--dangerous" in cmd.arguments, Reply.fail(1, "blocked"))
runner.fallback(Reply.ok(""))
```

`predicate` is infallible from the crate's perspective, like
`Supervisor.stop_when`: a raising or non-`bool` predicate is treated as "does
not match" rather than propagating, with the error surfaced via
[`sys.unraisablehook`](https://docs.python.org/3/library/sys.html#sys.unraisablehook)
(visible on stderr) so a buggy predicate is noisy, not silently wrong.

*Deeper: outcome semantics and the exception hierarchy — [Running commands](commands.md).*

## Scripted streaming: a live handle, no child

`ScriptedRunner.start(cmd)` (and `astart`) returns a real `RunningProcess`
backed by the canned reply instead of an OS child. The scripted stdout flows
through the **same line pumps** a real child uses, so `stdout_lines()`,
readiness probing, and `finish()` all behave identically — letting you test a
readiness-gate orchestration hermetically:

```python
import asyncio
from processkit import Command
from processkit.testing import Reply, ScriptedRunner

async def becomes_ready(runner):
    proc = runner.start(Command("server", ["serve"]))
    async for line in proc.stdout_lines():
        if "listening" in line:
            break
    return (await proc.afinish()).exited_zero

def test_server_becomes_ready():
    runner = ScriptedRunner()
    runner.on(["server", "serve"], Reply.lines(["booting", "listening on 8080"]))
    assert asyncio.run(becomes_ready(runner))   # satisfied by the canned banner
```

`Reply.lines([...])` scripts the stdout lines and the scripted run "exits" after
the last one; `Reply.pending()` scripts a run that never ends on its own (bound
it with the command's own `timeout()`). The honest boundary: a scripted handle
has no OS identity — `pid` is `None` and `profile` reports empty samples — so it
tests orchestration logic, not real I/O timing.

*Deeper: the live streaming surface (`stdout_lines`, `output_events`, `take_stdin`) — [Streaming & interactive I/O](streaming.md).*

## Record/replay cassettes: RecordReplayRunner

`RecordReplayRunner` closes the loop: capture real runs to a JSON *cassette*
once, then replay them offline — fast, deterministic, no subprocess in CI. It
shares the `Runner` verb surface, so it drops into the same seam.

```python
from processkit import Command
from processkit.testing import RecordReplayRunner

CMD = Command("python", ["-c", "import random; print(random.random())"])

# Record once against the real tool (an opt-in test run, say):
rec = RecordReplayRunner.record("fixtures/random.json")   # records via the real Runner
recorded = rec.run(CMD)                                    # spawns python once, captures it
rec.save()                                                 # write the cassette to disk

# Replay everywhere else — NEVER spawns:
rep = RecordReplayRunner.replay("fixtures/random.json")
assert rep.run(CMD) == recorded
```

That last assertion is the **no-respawn proof**: the recorded command prints a
fresh random number every real run, so if replay equals the recorded value,
nothing was spawned. (This is exactly how our suite proves it.)

`start()` is covered too: the cassette records a streamed run (capture-whole — the
child runs to completion, then the handle replays its captured lines through a real
`RunningProcess`) and replays it offline, so a readiness-gated `start` flow tests
hermetically. Two limits: an *interactive* run fed stdin mid-stream can't be
cassette-recorded (bound it with `Command.timeout()`, or script it with
`ScriptedRunner`); and **`output_bytes` is not supported through a cassette** — it
stores lossy-UTF-8 *text*, so it can't reproduce exact bytes and raises
`Unsupported` (capture bytes from a real or scripted runner instead).

Semantics worth knowing before you commit a cassette:

| Aspect | Behavior |
|---|---|
| Match key | program + args + cwd + a stdin **source digest** |
| Environment | override **values never reach the file** — only sorted variable names; env is *not* matched, so env differences can't cause spurious misses |
| Duplicates of one key | replayed in capture order, then the **last entry repeats** — a changing sequence (`rev-parse HEAD` before/after a commit) replays faithfully, while a retry/probe loop keeps getting a stable final answer |
| Miss | an invocation **absent from the cassette is a strict error** — replay never spawns a surprise subprocess, so a stale cassette fails loudly |

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**, and
keep secret-bearing cassettes out of shared trees. (`save()` writes the file
owner-only — `0600` on Unix — and refuses to follow a symlink, so a fresh
cassette isn't world-readable; the review is still on you before *committing* it.)

Record from a single thread. The capture buffer is per-runner; recording the same
`RecordReplayRunner` from several threads at once (only possible on a free-threaded
build) can interleave entries non-deterministically. Replay is read-only and has no
such constraint.

*Deeper: how a `ProcessResult` is shaped before it's captured — [the Cookbook](cookbook.md).*

## Asserting on calls: RecordingRunner

`RecordingRunner` is the *spy*: it replies to every command with one canned
`Reply` and records each call, so a test can assert on **what** your code ran —
not just react to a reply. It shares the `Runner` verb surface.

```python
from processkit import Command
from processkit.testing import RecordingRunner, Reply


def deploy(runner) -> None:
    runner.run(Command("git", ["push", "--tags"]))


def test_deploy_pushes_tags() -> None:
    runner = RecordingRunner.replying(Reply.ok(""))
    deploy(runner)

    inv = runner.only_call()            # the one call (raises unless exactly one)
    assert inv.program == "git"
    assert inv.args == ["push", "--tags"]
    assert inv.has_flag("--tags")
```

- **`replying(reply)`** — every command gets `reply`, built with the same `Reply`
  factories as `ScriptedRunner`.
- **`new(inner)`** — wrap `inner` (any of `Runner`, `ScriptedRunner`,
  `RecordReplayRunner`, or another `RecordingRunner`), recording every call
  made through it. The general form behind `replying()`, for combining
  recording with a double you've already built (e.g. a `RecordReplayRunner`
  cassette, or a `ScriptedRunner` with several `.on()` rules already wired
  up) instead of a fresh runner that just replies with one canned `Reply`.
- **`calls()`** — every recorded `Invocation`, in call order.
- **`only_call()`** — the single invocation, or a `ProcessError` if there wasn't
  exactly one.

Each `Invocation` exposes `program`, `args`, `cwd`, `env` (a `dict[str, str |
None]`; a `None` value is an `env_remove`), `has_stdin`, and a `has_flag(flag)`
helper. The values are there for your assertions, but its `repr` is **redacted**
(program, arg count, cwd, env names, has_stdin — never argv or env values), like
`Command`'s — a failing assertion that prints the invocation won't leak a
secret-bearing flag.

Reach for `RecordingRunner` when the *call* is what matters (did my code push the
tags?); for canned per-command replies use
[`ScriptedRunner`](#scripting-replies-scriptedrunner), and to replay real output
offline use [`RecordReplayRunner`](#recordreplay-cassettes-recordreplayrunner).

## Rendering commands without running: DryRunRunner

`DryRunRunner` is the double behind a tool's own `--dry-run`/`--echo` mode: it
never spawns anything, renders each command to its display-quoted line, and
returns a synthetic success. There is nothing to script — a dry run has no real
output to fake, only a command line to show — so every call just succeeds
(empty stdout; an exit code drawn from the command's own `success_codes`, so
the checking verbs stay in agreement even for a command whose accepted set
excludes `0`). It shares the `Runner` verb surface, so it drops into the same
seam.

```python
from processkit import Command
from processkit.testing import DryRunRunner

def prune(runner) -> None:
    runner.run(Command("rm", ["-rf", "build"]))
    runner.run(Command("rm", ["-rf", "dist"]))

def test_prune_targets_the_right_dirs() -> None:
    runner = DryRunRunner()
    prune(runner)                                    # nothing spawned
    assert runner.commands() == ["rm -rf build", "rm -rf dist"]
```

- **`commands()`** — the rendered command line of every call so far, in order,
  each produced by [`Command.command_line()`](commands.md) (the same display
  quoting you'd reach for by hand).
- **`only_command()`** — the single rendered line, or a `ProcessError` if there
  wasn't exactly one call (like `RecordingRunner.only_call()`).
- **`on_invocation(callback)`** — call `callback(line)` with each rendered line
  *as the call happens* — e.g. to print the echo live for a real `--dry-run`
  flag — **in addition to** the collected `commands()` snapshot. The callback is
  a fire-and-forget side effect: a raising one is surfaced via
  [`sys.unraisablehook`](https://docs.python.org/3/library/sys.html#sys.unraisablehook)
  rather than derailing the run it was only observing.

```python
runner = DryRunRunner()
runner.on_invocation(print)          # echo each command as it's "run"
deploy_plan(runner)                  # prints: kubectl apply -f manifest.yaml, …
```

Reach for `DryRunRunner` when the rendered *command line* is what you want to
assert on (or echo), with no reply to script and no output to replay — the
`--dry-run` seam. When a call needs a specific canned outcome, use
[`ScriptedRunner`](#scripting-replies-scriptedrunner); when you also need the
structured call record (cwd/env/stdin), use
[`RecordingRunner`](#asserting-on-calls-recordingrunner).

## Wrapping a CLI tool: CliClient

`CliClient` binds a program to per-call defaults, so repeated calls usually
pass only their `Args`. Every sync and async capture/check verb (`run`,
`output`, `output_bytes`, `exit_code`, `probe`, plus the `a`-prefixed twins)
accepts `Args | Command`: args are combined with the bound program and client
defaults; a `Command` can carry per-call customization, whose explicit settings
win over client defaults. This broader input type makes `CliClient` a valid
`ProcessRunner` implementation. It is not a `StreamingRunner`, because it does
not provide `start`/`astart`:

```python
from processkit import CliClient

git = CliClient("git", default_timeout=30.0)
head = git.run(["rev-parse", "HEAD"])        # or: await git.arun([...])
clean = git.probe(["diff", "--quiet"])
git.run(["fetch", "--quiet"])                # raises on failure; ignore the stdout
```

`CliClient` accepts an optional `runner=` too, driving every verb through the
given runner instead of the real one — a `ScriptedRunner` (or `RecordingRunner`
/ `RecordReplayRunner`) makes a `CliClient`-based wrapper hermetically testable
without restructuring it around a `runner` parameter of its own:

```python
from processkit import CliClient
from processkit.testing import Reply, ScriptedRunner

scripted = ScriptedRunner()
scripted.on(["git", "rev-parse", "HEAD"], Reply.ok("deadbeef\n"))
git = CliClient("git", runner=scripted)
assert git.run(["rev-parse", "HEAD"]) == "deadbeef"   # no real git spawned
```

`run_json` / `arun_json` run like `run` (requiring a zero exit) but parse the
stdout as JSON and return the decoded object — the wrapper for tools that speak
JSON (`gh`, `kubectl`, `docker`, `az`, `jj`). They go through the same `runner=`
seam, so a scripted reply's stdout is parsed exactly as a real tool's would be —
no process, no filesystem:

```python
from processkit import CliClient
from processkit.testing import Reply, ScriptedRunner

scripted = ScriptedRunner()
scripted.on(["gh", "pr", "view", "42", "--json", "state"], Reply.ok('{"state": "OPEN"}'))
gh = CliClient("gh", runner=scripted)
assert gh.run_json(["pr", "view", "42", "--json", "state"]) == {"state": "OPEN"}
```

Stdout that is not valid JSON raises `InvalidJson` (a `ProcessError` carrying the
`program` and a bounded stdout fragment) rather than a bare
`json.JSONDecodeError`, and a scripted reply exercises that path too:

```python
from processkit import CliClient, InvalidJson
from processkit.testing import Reply, ScriptedRunner

scripted = ScriptedRunner()
scripted.on(["gh", "pr", "view"], Reply.ok("not json at all"))
gh = CliClient("gh", runner=scripted)
try:
    gh.run_json(["pr", "view"])
except InvalidJson as exc:
    assert exc.program == "gh"
    assert "not json at all" in exc.stdout
```

`output_all`/`aoutput_all` (and their `_bytes` twins) and `Supervisor` accept
the same `runner=` keyword, for the same reason — a batch or a supervised
command can be driven through a double in a test, with the real `Runner`
the default when `runner=` is omitted.

*Deeper: per-client defaults and the full verb set — [the Cookbook](cookbook.md) → "Wrap a CLI tool".*

---

Next: [Running commands](commands.md) ·
[Streaming & interactive I/O](streaming.md) ·
[Supervision](supervision.md) · [Cookbook](cookbook.md)

---

# Command-line usage

[‹ docs index](./)

Most of this package's value lives behind Python code — but sometimes the
caller is a shell script or a CI step, not a Python program. `python -m
processkit run` is a thin CLI wrapper over `Command` / `ProcessGroup` for
exactly that case: kill-on-exit containment and resource limits for a single
shell command, with no Python to write. `python -m processkit supervise`
exposes restart-based keep-alive supervision (`Supervisor`) the same way, and
`python -m processkit doctor` is `run`'s read-only companion: a preflight
diagnosis of what this environment's kernel actually grants, without running
anything (see [below](#doctor-preflight-diagnose-the-environment)).

- [Basic usage](#basic-usage)
- [Flags](#flags)
- [`--profile`: machine-readable resource usage](#--profile-machine-readable-resource-usage)
- [Exit codes](#exit-codes)
- [supervise](#supervise)
- [Resource limits: hard cap or best effort?](#resource-limits-hard-cap-or-best-effort)
- [`doctor`: preflight-diagnose the environment](#doctor-preflight-diagnose-the-environment)
- [What you don't get here](#what-you-dont-get-here)

## Basic usage

```bash
python -m processkit run -- pytest -x
```

Everything after the **first** `--` is the child's own argv, untouched — a
second `--` in there belongs to the child, not to this wrapper:

```bash
python -m processkit run -- git log -- README.md
#                          ^ separator            ^ the child's own "--"
```

The child runs inside a `ProcessGroup`: even for one command, its whole
process tree — every grandchild it forks — is torn down when this wrapper
exits, and its stdin/stdout/stderr are inherited straight through to your
terminal: the child reads from the same stdin and its output is live, not
buffered up and dumped at the end.

```bash
# Bound the whole run to 30 seconds.
python -m processkit run --timeout 30 -- pytest -x

# Cap memory and process count too (needs a real container — see below).
python -m processkit run --max-memory 536870912 --max-processes 64 -- ./build.sh
```

## Flags

| Flag | Maps to | Notes |
|---|---|---|
| `--timeout SECONDS` | `Command.timeout(seconds)` | Kills the whole tree once the deadline passes. |
| `--timeout-grace SECONDS` | `Command.timeout_grace(seconds)` | Signal first, hard-kill after `SECONDS`. Requires `--timeout`; a usage error otherwise. |
| `--idle-timeout SECONDS` | `Command.idle_timeout(seconds)` | Kill the child if it emits no output line for `SECONDS`. Exit `123` (distinct from `--timeout`'s `124`). Pipes and re-emits stdout/stderr line-by-line; incompatible with `--profile`. See [below](#--idle-timeout-a-silence-watchdog). |
| `--max-memory BYTES` | `ProcessGroup(max_memory=...)` | Whole-tree memory cap. |
| `--max-processes N` | `ProcessGroup(max_processes=...)` | Fork-bomb ceiling for the tree. |
| `--cpu-quota FLOAT` | `ProcessGroup(cpu_quota=...)` | Fraction of a **single** core (`0.5` = half, `2.0` = two cores). |
| `--env-clear` | `Command.env_clear()` | Start the child with an empty environment. |
| `--inherit-env NAME` | `Command.inherit_env([...])` | Allow-list a parent variable through (implies `--env-clear`). Repeatable. |
| `--env KEY=VALUE` | `Command.env(key, value)` | Set/override a child environment variable. Repeatable. A value without `=` is a usage error. |
| `--cwd DIR` | `Command.cwd(dir)` | Run the child with `DIR` as its working directory. |
| `--profile [FILE]` | `RunningProcess.profile(...)` | After the child exits, emit a JSON resource profile — to stderr if `FILE` is omitted, or written to `FILE` otherwise. See [below](#--profile-machine-readable-resource-usage). |
| `--create-no-window` | `Command.create_no_window()` | Do not create a console window for the child. No-op outside Windows (same as the underlying binding method). |

Every numeric flag rejects zero and negative values at the argument-parsing
stage (a usage error, not a traceback). See `docs/process-groups.md` and
`docs/commands.md` for what each underlying builder method does in full —
including how the environment builders (`env_clear` / `inherit_env` / `env`)
compose regardless of call order.

### `--profile`: machine-readable resource usage

Without `--profile`, `run` only ever reports an exit code — the resource side
of the run (wall time, CPU time, peak memory) is invisible from the CLI, even
though the binding already tracks it end-to-end (`RunningProcess.profile()` /
`RunProfile`, see [Streaming & interactive
I/O](streaming.md#live-introspection-and-per-run-telemetry)). `--profile`
exposes exactly that, for a CI step that wants a machine-readable resource
accounting of a containerized run without writing any Python:

```bash
# Print the profile to stderr once the child exits.
python -m processkit run --profile -- pytest -x

# Or write it to a file instead.
python -m processkit run --profile /tmp/run-profile.json -- pytest -x
```

Either way, the child's own stdin/stdout/stderr are still inherited straight
through exactly as without the flag — the profile is only ever emitted
**after** the child has fully exited (the same point `outcome()` itself
returns at), so it never interleaves with the child's own output. It is one
line of JSON with these fields:

| Field | Type | Meaning |
|---|---|---|
| `duration_seconds` | `float` | Wall-clock time the run took. |
| `cpu_time_seconds` | `float \| null` | User + kernel CPU time consumed by the whole run. |
| `peak_memory_bytes` | `int \| null` | Peak memory observed during the run. |
| `avg_cpu_cores` | `float \| null` | `cpu_time_seconds / duration_seconds` — e.g. `1.7` means ~1.7 cores kept busy on average. |
| `samples` | `int` | How many resource samples were taken while the child ran. |
| `code` | `int \| null` | Same meaning as the process's own exit code (`null` if the run ended some other way). |
| `signal` | `int \| null` | Set if the child was killed by a signal (POSIX only). |
| `timed_out` | `bool` | Whether `--timeout` expired. |

The `cpu_time_seconds` / `peak_memory_bytes` / `avg_cpu_cores` fields need the
same kernel-level accounting `ProcessGroup`'s own resource limits do (a
Windows Job Object or a Linux cgroup-v2 root — see [Resource limits: hard cap
or best effort?](#resource-limits-hard-cap-or-best-effort) above); where the
environment doesn't grant that, they serialize as JSON `null` rather than
failing the run — `duration_seconds`/`samples`/`code`/`signal`/`timed_out` are
always available. `--profile`'s own exit-code contract is otherwise unchanged
from the table above — it never introduces a new exit code, and a failure
writing the profile to `FILE` (e.g. an unwritable path) surfaces as the
existing internal-failure code `125`, with a one-line message on stderr.

### `--idle-timeout`: a silence watchdog

`--idle-timeout SECONDS` maps to `Command.idle_timeout(seconds)` — it kills the
child if it produces no output line for that long, for a tool that hangs
silently while a healthy long job keeps printing. It exits **123**, deliberately
distinct from `--timeout`'s `124`, so the two timeout classes stay tellable
apart by exit code.

```bash
# Kill the build if it goes quiet for 30s, even though its total budget is high.
python -m processkit run --timeout 3600 --idle-timeout 30 -- ./flaky-build
```

Idle monitoring needs the per-line output channel, so with this flag `run`
**pipes** the child's stdout/stderr and re-emits each decoded line (one at a
time, with a trailing newline) instead of inheriting the raw streams. That is a
deliberate fidelity trade taken only when the flag is set: output is UTF-8
decoded and line-framed, and the child's streams are not a TTY. For the same
reason `--idle-timeout` is **incompatible with `--profile`** (they need
different consuming operations on the one handle) — combining them is a usage
error.

`--idle-timeout` is **not** available under `supervise`: each supervised
incarnation runs through `Supervisor`'s one-shot verbs, which processkit's core
gives no idle-timeout hook, so passing it there is a usage error until upstream
support lands. Use `run --idle-timeout` for a single command.

## Exit codes

This wrapper's own exit code mirrors the child's — plus a small set of
reserved codes for cases where there is no child exit code to report,
following the same convention GNU coreutils' `timeout` and POSIX shells use:

| Exit code | Meaning |
|---|---|
| *(the child's own code)* | Normal completion — passed through unchanged. |
| `119` | This wrapper could not deliver its own buffered output (see [How the wrapper terminates](#how-the-wrapper-terminates)). Shared by every subcommand. |
| `123` | `--idle-timeout` expired; the child produced no output line in time and was killed. Distinct from `124`. |
| `124` | `--timeout` expired; the tree was killed. |
| `125` | An internal / containment failure (see below). |
| `126` | The program was found but could not be executed. |
| `127` | The program could not be found. |
| `128 + N` | The child was killed by signal `N` (POSIX only). |
| `128 + SIGINT` (`130`) | `python -m processkit` itself was interrupted (Ctrl+C) — anywhere, including during startup, argument parsing, or `doctor`. |

None of these ever surface as a raw Python traceback — every documented
processkit exception (`Timeout`, `Signalled`, `ProcessNotFound`,
`PermissionDenied`, `ResourceLimit`, `Unsupported`) is caught and turned into
one of the codes above, with a one-line message on stderr. Ctrl+C is part of
that promise too: it always ends as `128 + SIGINT` with a single
`processkit: interrupted` line, never a `KeyboardInterrupt` traceback.

### How the wrapper terminates

`python -m processkit` flushes its own stdout/stderr, raises `SystemExit` with
the selected code, and then runs ordinary interpreter finalization (`atexit`
hooks, garbage-collected finalizers, and module teardown included).

`--idle-timeout` is the one path that drives the async surface. Its completion
handoff wakes the event loop through a socket and resolves the Future on the
loop thread, so no detached tokio thread remains inside Python after the await
resumes; normal finalization cannot race the bridge. See
[Async runtimes & event loops](event-loops.md#interpreter-shutdown-and-the-async-bridge).

Two exit duties remain explicit so their outcomes stay part of the CLI's
documented contract rather than depending on CPython's fallback behavior:

- **The final flush of its own stdout/stderr.** Redirected into a pipe,
  stdout is block-buffered, so this is what makes the last lines arrive at
  all. If that flush fails in a way that *loses* output — a full or failing
  disk, a stream closed underneath the process — the wrapper exits **119**
  with one line on stderr instead of the code it was about to report. That is
  deliberate: reporting the child's own code would claim a complete,
  faithfully relayed run. A receiver that simply went away
  (`BrokenPipeError`, e.g. `python -m processkit run ... | head`) is *not*
  that case and stays silent — no exit code can deliver output to a closed
  pipe.
- **Ctrl+C that lands outside `run`/`supervise`'s own guarded blocks** — during
  startup, argument parsing, or `doctor`. It exits `128 + SIGINT` (`130`) with
  the same one-line `processkit: interrupted` message the guarded paths print,
  on every platform. For `doctor` this matters beyond tidiness: `1` is a valid
  `doctor` verdict, so an interrupted probe must never be reported as one.

## supervise

**Basic usage:**

```bash
python -m processkit supervise [OPTIONS] -- PROGRAM [ARG ...]
```

`supervise` keeps a command alive by restarting it according to a selected
policy, with configurable exponential backoff. Its child's stdin is inherited
exactly as with `run` (`Command.inherit_stdin()`). Stdout/stderr are handled
differently than `run`, though: `Supervisor` requires a **piped** stdout to
capture each incarnation's result (to evaluate the restart policy and
populate `SupervisionOutcome.final_result`) — a non-piped stdout errors every
incarnation. To still stream live to this terminal, this wrapper pipes both
streams and tees every decoded line straight through to its own inherited
stdout/stderr (`Command.stdout_tee`/`stderr_tee`); output still appears live,
just line-buffered rather than a byte-for-byte fd passthrough.

| Flag | Description |
|---|---|
| `--restart {always,on_crash,never}` | Restart policy passed to `Supervisor`. |
| `--max-restarts N` | Stop after `N` restarts. `N` must be positive. |
| `--backoff-initial SECONDS` | Initial delay before a restart. Must be positive. |
| `--backoff-factor FLOAT` | Multiplier for successive restart delays. Must be at least `1`. |
| `--max-backoff SECONDS` | Upper bound for restart delay. Must be positive. |
| `--no-jitter` | Disable restart-delay jitter; jitter is enabled by default. |
| `--env-clear` | Start the child with an empty environment. |
| `--inherit-env NAME` | Allow-list a parent variable (implies `--env-clear`). Repeatable. |
| `--env KEY=VALUE` | Set or override a child variable. Repeatable. |
| `--cwd DIR` | Run the child with `DIR` as its working directory. |

```bash
python -m processkit supervise --restart always --max-restarts 5 -- some_command
```

| Exit code | Meaning |
|---|---|
| *(the final child result's code)* | Supervision stopped because the restart policy was satisfied. |
| `119` | This wrapper could not deliver its own buffered output — the entry-point-wide code from [How the wrapper terminates](#how-the-wrapper-terminates), not a `supervise` one. |
| `120` | An internal command/supervisor failure, including a missing or unexecutable program. |
| `121` | The restart policy required another attempt, but `--max-restarts` was exhausted. |
| `122` | Supervision gave up due to a `give_up_when` condition (reserved for API-driven outcomes). |
| `128 + N` | The final incarnation was killed by signal `N` (POSIX only) — mirrors `run`'s own convention. |
| `128 + SIGINT` (`130`) | `python -m processkit` itself was interrupted with Ctrl+C. |

## Resource limits: hard cap or best effort?

`--max-memory` / `--max-processes` / `--cpu-quota` need a real container — a
Windows Job Object or a Linux **cgroup-v2 root** (see
[Process groups](process-groups.md#resource-limits-the-sandbox) and
[Platform support](platforms.md)). Inside an ordinary container, a systemd
user session, or on macOS, the kernel refuses these caps outright.

Rather than fail the whole run over a cap the environment can't grant, this
CLI **degrades**: it prints a warning to stderr and re-runs the child in a
plain, uncapped `ProcessGroup` — "contained, but uncapped" — the same
fallback `examples/04_sandbox_resource_limits.py` uses. The no-orphan
containment guarantee still applies either way; only the specific numeric
caps are dropped. If your script depends on the cap actually being enforced,
check stderr for that warning rather than assuming it always held.

## `doctor`: preflight-diagnose the environment

`--max-memory`/`--max-processes`/`--cpu-quota` depend on kernel primitives
that are not guaranteed to be there (see above) — until now, the only way to
find out was to run `run` for real and read a warning on stderr, or catch
`ResourceLimit`/`Unsupported` from the Python API. `python -m processkit
doctor` answers the same question up front, without running anything:

```bash
python -m processkit doctor
```

```text
processkit doctor
  containment mechanism : cgroup_v2
  resource limits        : available
  verdict: OK - containment and resource limits are both available (exit 0)
```

Degraded (containment holds, but the kernel refuses at least one resource
limit — the typical container / systemd user session / non-root cgroup /
macOS case; `--max-memory`, `--max-processes`, and `--cpu-quota` are probed
**independently**, since on Linux cgroup-v2 they are separate controllers
that can be unavailable one without the others):

```text
processkit doctor
  containment mechanism : process_group
  resource limits        : unavailable --max-memory (ResourceLimit: cgroup v2 root required)
  note: --max-memory/--max-processes/--cpu-quota need a Windows Job Object or
  a Linux cgroup-v2 root; the kernel typically refuses them inside
  containers, systemd user sessions, and non-root cgroups, and always on
  macOS (docs/cli.md#resource-limits-hard-cap-or-best-effort).
  verdict: DEGRADED - containment is enforced, but resource limits are not (exit 1)
```

It never spawns a child process — only constructs (and immediately drops) a
few throwaway `ProcessGroup` instances to see what the kernel actually
grants (one for the containment mechanism, one per resource-limit
controller). `doctor` has its own exit-code namespace, deliberately disjoint
from `run`'s codes above (`124`/`125`/`126`/`127`/`128 + signal`) *and* from
argparse's own usage-error code `2` (the same code `run` itself uses for a
bad invocation) — `doctor` never returns `2` as a diagnostic verdict, so a
CI gate can always read `2` as "you called this wrong", unambiguous from any
of the codes below:

| Exit code | Meaning |
|---|---|
| `0` | Resource limits are available (containment *and* all three caps hold). |
| `1` | Containment is enforced, but at least one resource limit is not — the same "contained, but uncapped" gap `run` degrades around. |
| `2` | *(not returned by `doctor` itself)* — a usage error, e.g. an unknown flag or `doctor`'s disallowed trailing command; reserved to keep it unambiguous from a real diagnostic result. |
| `3` | Containment itself is unavailable (should not happen on any supported platform). |
| `4` | A probe raised an unexpected operational error (`OSError`/`PermissionError`, e.g. failing to read cgroup state) rather than a definitive result — the environment's actual availability could not be determined. |

The two entry-point-wide codes from
[How the wrapper terminates](#how-the-wrapper-terminates) — `119` (output the
wrapper could not deliver) and `128 + SIGINT` (`130`, interrupted) — are
disjoint from those verdicts by construction, so a CI gate reading `doctor`'s
code never has to disambiguate them from a diagnosis. In particular an
interrupted `doctor` reports `130`, never `1`.

For CI, `doctor --json` replaces that text report with one JSON object on
stdout while preserving the same exit code. Its stable base schema is:

```json
{
  "mechanism": "cgroup_v2",
  "verdict": "OK",
  "exit_code": 0,
  "resource_limits": {
    "max_memory": true,
    "max_processes": true,
    "cpu_quota": true
  },
  "caveat": "--max-memory/--max-processes/--cpu-quota need a Windows Job Object or a Linux cgroup-v2 root; the kernel typically refuses them inside containers, systemd user sessions, and non-root cgroups, and always on macOS (docs/cli.md#resource-limits-hard-cap-or-best-effort)."
}
```

`mechanism`, `verdict`, and `caveat` are strings; `exit_code` is an integer;
and all `resource_limits` fields are booleans. `verdict` is one of `OK`,
`DEGRADED`, `UNAVAILABLE`, or `ERROR`, matching exit codes `0`, `1`, `3`, and
`4`. When an `OSError` prevents a definitive probe result, the payload also
contains `error_probe_failures`, a list of error strings.

`doctor` takes only `-h`/`--help` and `--json` — in particular, no trailing
`-- PROGRAM ...` (it is diagnostic-only and never runs a command).

## What you don't get here

This is a v1, deliberately minimal wrapper — reach for the Python API
directly for anything beyond it: piping several commands together
([Pipelines](pipelines.md)), advanced supervision callbacks such as `stop_when` and `give_up_when`
([Supervision](supervision.md)), line-by-line streaming ([Streaming &
interactive I/O](streaming.md)), or running a batch of commands concurrently
(`output_all` / `aoutput_all`). There is also no `--dry-run` mode yet — a
plausible follow-up, not implemented today. There is also no `--output-limit`
flag: stdio here is always inherited straight through to your terminal, so
there is no captured-output buffer for
`Command.output_limit(...)` to bound in the first place — that method only
matters when a caller captures output via the Python API instead.

---

Next: [Process groups](process-groups.md) ·
[Timeouts & cancellation](timeouts-and-cancellation.md) ·
[Cookbook](cookbook.md)

---

# Performance & overhead

**Short answer: the bridge adds no silly overhead.** Spawning a child process
is fundamentally *syscall-bound* — `fork`/`exec`/`posix_spawn` on POSIX,
`CreateProcess` plus Job Object setup on Windows — and that OS-side cost
dominates the total wall-clock time of a run by a wide margin. The PyO3 glue
between Python and the `processkit` Rust crate (argument marshaling, the
async-runtime hop for the asyncio surface, error mapping) adds a small,
constant-time cost on top of a syscall path that already dominates the total.
This page explains what "no silly overhead" means concretely, points at the
benchmark suite that backs the claim with a real number instead of a loose
pass/fail bound, and shows how to reproduce it yourself.

## Why the workload is syscall-bound

Every one-shot verb (`output()`, `run()`, …) and every `ProcessGroup.start()`
does the same thing under the hood: ask the kernel to create a process (and,
for a group, first create and enter its containment mechanism — a Windows Job
Object, a Linux cgroup v2, or a POSIX process group), wait for it to produce
output and/or exit, then tear the containment down. None of that work can be
made faster by changing what happens *above* the crate boundary — the crate
already does the minimum number of syscalls the OS requires, with no
busy-polling. See [Architecture](internals.md#two-layers-one-boundary) for
where the binding crate's thin glue ends and the `processkit` crate's platform
logic begins; the binding layer never reimplements any OS mechanism, so it
never becomes a bottleneck.

Because process creation is what dominates, the per-call overhead in the
Python↔Rust boundary is lost in the noise next to a kernel-side operation
costing orders of magnitude more — see [What each benchmark
measures](#what-each-benchmark-measures) below for the harness that turns
this into a reproducible number instead of a fixed figure here. That is the
whole argument behind "no silly overhead": not that the bridge is free, but
that its cost is negligible relative to the workload it wraps.

## What each benchmark measures

The [`benchmarks/`](https://github.com/ZelAnton/processkit-py/tree/main/benchmarks)
suite (`pytest-benchmark` based) turns that qualitative argument into
reproducible numbers, answering the question
[ROADMAP.md](https://github.com/ZelAnton/processkit-py/blob/main/ROADMAP.md)'s
Phase 5 asks with a real measurement instead of only the loose sanity bound in
`tests/test_hardening.py::test_no_silly_per_call_overhead`:

- **`test_spawn_capture.py`** — spawn + capture a single short-lived command:
  `processkit`'s `Command(...).output()` against the two stdlib "naive"
  equivalents, `subprocess.run(..., capture_output=True)` and
  `asyncio.create_subprocess_exec(...)` + `communicate()`. Same payload on all
  three, so the comparison isolates per-call overhead rather than a differing
  workload.
- **`test_process_group.py`** — `ProcessGroup` start/exit: creating the
  group's kernel container, entering it, starting one short-lived child, and
  tearing the whole tree down. This is the cost of containment itself, on top
  of a bare spawn.
- **`test_streaming_throughput.py`** — `RunningProcess.stdout_lines()` (see
  [Streaming & interactive I/O](streaming.md)) draining a known number of
  lines end to end, i.e. sustained line-streaming throughput rather than a
  single spawn/exit round trip.
- **`test_output_all.py`** — `output_all()` / `aoutput_all()` (see
  [Cookbook](cookbook.md)) at 1/10/50-way concurrency, i.e. how batched
  fan-out scales as concurrency grows.

## Reproducing locally

The suite is **not** part of the PR gate — it lives in its own `bench`
dependency-group and is excluded from `testpaths`, so an ordinary
`pytest`/`uv run pytest` never collects it. Install the group and run it
explicitly:

```console
uv sync --group bench
uv run pytest benchmarks/ --benchmark-only -p no:xdist -o addopts=""
```

`-p no:xdist -o addopts=""` disables `-n auto` (the repo's default
`addopts`) — `pytest-benchmark` needs to run in the main process, in a single
worker, to produce meaningful timings; under `pytest-xdist` it silently skips
measuring instead. See
[`benchmarks/README.md`](https://github.com/ZelAnton/processkit-py/blob/main/benchmarks/README.md)
for the full set of useful flags (`--benchmark-compare`,
`--benchmark-autosave`, `--benchmark-json`, …).

## Qualitative expectations

Rather than pin numbers here — which drift with hardware, OS, and Python
version, and would go stale the moment they were written down — this section
sets expectations you can sanity-check against your own run of the harness
above:

- **Single-call overhead is small relative to spawn cost.** The gap between
  `processkit`'s `output()` and the stdlib equivalents in
  `test_spawn_capture.py` should be a small fraction of the total per-call
  time, not a multiple of it — the bulk of the time in every one of the three
  compared approaches is the OS spawning and reaping the child.
- **Containment adds a bounded, one-time setup/teardown cost per group**, not
  a per-member one — creating and entering a Job Object / cgroup / process
  group happens once in `ProcessGroup.start()`/`__aenter__`, so starting many
  members into an already-open group is cheap relative to opening the group
  itself.
- **Line-streaming throughput scales with the amount of output**, not with a
  fixed per-line Python↔Rust round trip — `stdout_lines()` batches reads on
  the Rust side, so throughput should stay close to linear as line count
  grows.
- **`output_all()`/`aoutput_all()` scale sub-linearly with concurrency** up to
  the point where the workload becomes bound by the number of OS threads/CPUs
  available to run children concurrently, not by anything in the binding
  layer.

## Batch default concurrency

Leaving `concurrency` unset uses the CPU count available to the *process*, not
merely the host-wide hardware count. The Rust batch verbs use
`std::thread::available_parallelism()`; the streaming Python helpers use
`os.process_cpu_count()` on Python 3.13+ and fall back to `os.cpu_count()` on
Python 3.10-3.12. These sources account for CPU affinity and cgroup quotas where
the platform exposes them. If no count is available, every batch entry point
uses `4`. Pass an explicit positive `concurrency` to choose a different cap;
zero and negative values raise `ValueError`.

## Continuous tracking

The `bench` job in
[`nightly-hardening.yml`](https://github.com/ZelAnton/processkit-py/blob/main/.github/workflows/nightly-hardening.yml)
runs this suite on the same `schedule`/`workflow_dispatch` triggers as the
`stress` job — never on `push`/`pull_request` — and publishes the results as a
table in the job summary, so a regression shows up as a trend across nights
rather than only when someone happens to run the harness locally.

---

# Async runtimes & event loops

[‹ docs index](./)

processkit's async surface is **asyncio-native**. Every `a`-prefixed verb
(`aoutput`, `arun`, `astart`, …) and every streaming handle (`stdout_lines()`,
`output_events()`, interactive stdin) is bridged onto the running asyncio event
loop by [`pyo3-async-runtimes`], so it needs a real asyncio loop underneath.
This page says exactly which runtimes provide one — and which don't.

- [Support at a glance](#support-at-a-glance)
- [asyncio & uvloop](#asyncio--uvloop)
- [anyio](#anyio)
- [trio](#trio)
- [Why asyncio-native](#why-asyncio-native)
- [The readiness helpers](#the-readiness-helpers)

## Support at a glance

| Runtime | Supported | Why |
|---|---|---|
| **asyncio** (stdlib) | Yes — native | The bridge targets it directly |
| **uvloop** | Yes | A drop-in asyncio loop policy — the bridge sees an ordinary running asyncio loop |
| **anyio** on the **asyncio** backend | Yes | anyio's asyncio backend runs a real asyncio loop; the bridged awaitables await normally |
| **anyio** on the **trio** backend | No | No asyncio loop is present |
| **trio** (native) | No | No asyncio loop, and the bridge has no trio backend |
| **curio** | No | Same reason as trio |

The dividing line is simple: **is a real asyncio event loop running?** If yes
(plain asyncio, uvloop, or anyio-on-asyncio), the whole async surface works
unchanged. If no (trio, anyio-on-trio, curio), the `a`-prefixed verbs can't be
awaited — the sync surface (`output()`, `run()`, `ProcessGroup`, …) still works
from any thread, since it doesn't touch an event loop at all.

## asyncio & uvloop

The default. Nothing to configure:

```python
import asyncio
from processkit import Command

async def main():
    result = await Command("git", ["rev-parse", "HEAD"]).aoutput()
    print(result.stdout.strip())

asyncio.run(main())
```

[uvloop] is a faster asyncio loop implementation, installed as the loop policy.
Because it *is* an asyncio loop, processkit needs no special handling — install
the policy and every verb behaves identically (only with faster I/O
scheduling):

```python
import asyncio
import uvloop
from processkit import Command

async def main():
    await Command("./build.sh").arun()

uvloop.install()      # or asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
asyncio.run(main())   # 3.12+: asyncio.run(main(), loop_factory=uvloop.new_event_loop)
```

## anyio

[anyio] runs on one of two backends. On its **default asyncio backend**,
processkit works today with no changes — anyio does not hide the underlying
asyncio loop, so the bridged awaitables await normally, and asyncio
cancellation (which anyio maps onto its own cancel scopes) still tears the tree
down:

```python
import anyio
from processkit import Command

async def main():
    result = await Command("git", ["status", "--short"]).aoutput()
    print(result.stdout)

anyio.run(main)   # default backend="asyncio" — supported
```

On the **trio backend** (`anyio.run(main, backend="trio")`) there is no asyncio
loop, so the `a`-prefixed verbs cannot be awaited — see below.

## trio

Native [trio] (and anyio's trio backend, and curio) are **not supported**. A
trio program runs trio's own scheduler, not an asyncio loop, so the awaitables
processkit hands back — `asyncio.Future`s produced by the asyncio-wired bridge —
aren't trio-awaitable, and the binding refuses with a clear "no running asyncio
event loop" error anyway. For a quick symptom-to-solution map, see
[Troubleshooting](troubleshooting.md#a-prefixed-verbs-report-no-running-asyncio-event-loop).

If you're on trio and need processkit, the pragmatic bridge is
[`trio-asyncio`], which runs an asyncio loop inside a trio program; processkit's
verbs then execute in that asyncio context. That is a user-side integration
this package does not ship or test — treat it as unsupported-but-possible, not
a guarantee. The reliable alternative is the **synchronous** surface
(`output()`, `run()`, `ProcessGroup`, …), which needs no event loop and is
usable from a trio worker thread.

## Why asyncio-native

This is a deliberate, standing decision (project ROADMAP, Open decision #2),
not an oversight or a v1-only stopgap:

- The async surface is bridged tokio ↔ asyncio by [`pyo3-async-runtimes`],
  which targets asyncio and ships **no trio backend**. Native trio would mean
  writing a loop-agnostic bridge from scratch.
- That bridge is the single highest-risk part of the binding. Re-implementing
  it against trio's cancellation model — level-triggered cancel scopes and
  checkpoints, versus asyncio's edge-triggered `CancelledError` — while
  preserving the [kill-on-cancel no-orphan guarantee](timeouts-and-cancellation.md#cancelling-an-awaited-async-run)
  is a research effort in its own right, on a binding whose whole thesis is to
  stay thin and *not* reimplement hard concurrency logic.
- The anyio ecosystem is not actually shut out — anyio-on-asyncio works — so
  the excluded slice is specifically the trio-family loops, a smaller segment.

The path if this is ever revisited: port the pure-Python readiness helpers to
anyio primitives first (cheap, and it makes `wait_for_port` / `wait_until`
loop-agnostic), then evaluate a loop-agnostic compiled bridge once
`pyo3-async-runtimes` grows a trio backend or a concrete demand signal appears.

## Interpreter shutdown and the async bridge

Completed async operations are handed back on the **event-loop thread**. The
tokio task stores a type-erased outcome in Rust memory and wakes one shared
socket per event loop; `loop.sock_recv(...)` receives the wakeup and performs
the Python value conversion plus `Future.set_result()` / `set_exception()` on
the loop itself. Repeated stream steps reuse that same dispatcher rather than
opening a socket for every `__anext__`.
`sock_recv` is supported by the standard selector and Windows proactor loops,
uvloop, and anyio's asyncio backend, so this safety property does not narrow
the supported event-loop set.

This design matters at process exit. The previous upstream completion path
entered Python from a detached tokio blocking thread and called
`loop.call_soon_threadsafe(...)`. The awaiting coroutine could resume while
that foreign thread was still returning through Python frames; a short-lived
program whose last act was the `await` could then begin `Py_FinalizeEx`
underneath it and intermittently die from SIGSEGV after all useful work had
finished. The socket handoff never enters Python from the completing runtime
thread, so once the await resumes there is no bridge thread left inside the
interpreter. Ordinary interpreter finalization is safe; no `os._exit` or
post-await delay is required.

Cancellation keeps the same contract: cancelling the backing asyncio Future
signals the tokio work to drop and cancels the private socket receive, which
preserves `asyncio.CancelledError` and process-tree teardown.

## The readiness helpers

The readiness helpers ([`wait_for_port`](streaming.md#readiness-probes),
`wait_for_line`, `wait_for_path`, `wait_until`) are pure Python but built on
asyncio primitives, so they follow the same rule as the rest of the surface:
they need a running asyncio loop (asyncio, uvloop, or anyio-on-asyncio). In
particular `wait_for_line` consumes a `RunningProcess` stream, which is itself
asyncio-bridged — so there is no configuration in which the streaming surface is
asyncio-only while the helpers are not.

`sample_stats` (see [Process groups](process-groups.md#live-monitoring)) is the
same story: pure Python built on `asyncio.sleep`, needing the same running
asyncio loop as everything else here.

---

Next: [Timeouts & cancellation](timeouts-and-cancellation.md) ·
[Streaming & interactive I/O](streaming.md) ·
[Platform support](platforms.md) ·
[Cookbook](cookbook.md)

[`pyo3-async-runtimes`]: https://github.com/PyO3/pyo3-async-runtimes
[uvloop]: https://github.com/MagicStack/uvloop
[anyio]: https://anyio.readthedocs.io/
[trio]: https://trio.readthedocs.io/
[`trio-asyncio`]: https://github.com/python-trio/trio-asyncio

---

# Platform support & caveats

processkit's guarantee is strongest on Windows and weakest on macOS. This is
inherent to what each OS offers, and it is documented here rather than hidden.
`ProcessGroup.mechanism` tells you which mechanism is active at runtime:
`"job_object"`, `"cgroup_v2"`, or `"process_group"`.

## Teardown (the no-orphan guarantee)

| | Mechanism | When the `with` / `async with` block exits | If the Python process is hard-killed (`SIGKILL`, `os._exit`) |
|---|---|---|---|
| **Windows** | Job Object | Whole tree reaped (kernel-enforced) | **Still reaped** — `KILL_ON_JOB_CLOSE` fires when the last handle closes |
| **Linux** | cgroup v2 (else process group) | Whole tree reaped | No whole-tree cleanup. Opt-in `kill_on_parent_death()` kills the direct child only; grandchildren survive |
| **macOS / BSD** | process group | Tree reaped, *except* children that called `setsid()` | No automatic cleanup; the platform has no parent-death signal equivalent |

The takeaway: the `with` / `async with` exit path (and ordinary GC) reaps the
tree on every platform. Whole-tree cleanup after a hard kill of the parent is a
Windows-only property. `Command.kill_on_parent_death_scope()` reports the
actual abrupt-death capability as `"whole_tree"`, `"direct_child_only"`, or
`"unsupported"`. Lean on the context managers; don't rely on `__del__` or
`atexit`, which don't run on `SIGKILL` / `os._exit`.

Cancelling an awaited run (`task.cancel()`, `asyncio.wait_for`,
`asyncio.timeout`) reaps the run's tree on every platform — the dropped future
tears it down.

## Resource limits (`ProcessGroup(max_memory=…, max_processes=…, cpu_quota=…)`)

| | Support |
|---|---|
| **Windows** | Job Object enforces memory / active-process / CPU-rate caps |
| **Linux** | cgroup v2 — **only when this process runs at the cgroup-v2 root**. Under a container, a systemd session/scope/service, or any non-root cgroup, the kernel's "no internal processes" rule forbids it and `ResourceLimit` is raised |
| **macOS / BSD** | No whole-tree limit primitive — requesting any limit raises `ResourceLimit` (a fail-fast, never a silently-unbounded group) |

If you need limits inside a container, run the process at the container's cgroup
root (the create-leaf / migrate-self / enable-controllers dance), or use a
runtime that grants cgroup delegation. For the symptom-first version of this
failure, see [Troubleshooting](troubleshooting.md#resourcelimit-under-docker-systemd-or-a-non-root-cgroup).

## Signals, suspend/resume, stats

| | `signal()` / `suspend()` / `resume()` | `stats()` |
|---|---|---|
| **Windows** | Only `kill` is deliverable — it terminates the job; **every other name, including `term`, raises `Unsupported`**. suspend/resume freeze/thaw the job | Memory + process count via the OS process APIs |
| **Linux** | Real signals to the cgroup/process group; freeze via cgroup or `SIGSTOP`/`SIGCONT` | cgroup + `/proc` |
| **macOS / BSD** | Real signals to the process group | Process count only; CPU / peak-memory are `None` (no whole-tree kernel accounting) |

Operations a platform can't perform raise `Unsupported` — catch it if you target
multiple platforms.

## Multiprocessing: use `spawn` or `forkserver`, not `fork`

processkit runs a tokio runtime with background worker threads, started lazily
the first time you call any verb. A bare POSIX `fork()` copies that runtime into
the child **without** its worker threads — `fork()` carries only the calling
thread across — and any lock a worker held at fork time stays locked forever in
the child. Driving the copied runtime there (any further processkit call) would
deadlock or panic with no recovery. This is the standard "don't `fork()` a
multi-threaded process" hazard; processkit is not special here, but its runtime
makes the process multi-threaded as soon as you use it.

**What processkit does about it.** Rather than hang, a processkit verb called
from a process that `fork()`ed *after* the runtime was initialized fails fast
with a clear `ProcessError` (it detects the PID change and refuses before
touching the dead runtime). Nothing is spawned, so nothing is orphaned. It does
**not** transparently rebuild the runtime in the child: the managed runtime lives
in a process-global that cannot be soundly reset, so a clean refusal is the safe
contract.

**What you should do.** Choose a fork-free start method for `multiprocessing` /
`concurrent.futures.ProcessPoolExecutor` whenever the workers use processkit:

```python
import multiprocessing as mp

ctx = mp.get_context("spawn")  # or "forkserver"
with ctx.Pool() as pool:
    ...
```

`spawn` and `forkserver` start each worker from a fresh interpreter, so every
worker initializes its own runtime cleanly. On macOS and Windows `spawn` is
already the default; on Linux the default is still `fork` for `multiprocessing`
below Python 3.14, so set the context explicitly there. If you must call
`os.fork()` directly, do it **before** the first processkit call in the parent —
a child that forks before the runtime is initialized simply builds its own.

## Python build

- Distributed as **abi3 wheels for CPython 3.10+** (one wheel per OS/arch runs on
  every supported minor version, 3.14 included).
- **Free-threaded CPython (PEP 703) is supported.** The extension declares
  `gil_used = false`, so importing it on a free-threaded build does *not* re-enable
  the GIL. Because the limited API (abi3) isn't available on free-threaded builds,
  this ships as a **version-specific wheel** for CPython 3.14t (where free-threading
  is officially supported, per PEP 779) alongside the abi3 GIL wheel. The full test
  suite runs on the free-threaded interpreter in CI. The binding holds no
  unsynchronized shared state, so calling it from many threads is memory-safe —
  PyO3's per-object borrow checking still serializes *mutating* calls on a single
  shared handle (a concurrent mutate raises rather than racing), so give each
  thread its own `Command` / `RunningProcess` / runner as you would any object.

## Wheel availability

Published on [PyPI](https://pypi.org/project/processkit-py/) with prebuilt
wheels covering:

| Platform | Architectures |
|---|---|
| **Linux** (manylinux, glibc) | x86_64, aarch64 |
| **Linux** (musllinux, musl — Alpine) | x86_64, aarch64 |
| **macOS** | arm64 (Apple Silicon), x86_64 (Intel) |
| **Windows** | x64, arm64 |

Each row ships both the abi3 GIL wheel (CPython 3.10+) and the free-threaded
cp314t wheel, with an sdist alongside for source builds anywhere. The Windows
arm64 and Linux aarch64 wheels are built natively on GitHub's ARM runners (the
free-for-public-repos `windows-11-arm` and `ubuntu-24.04-arm`). The Intel macOS
wheel is cross-compiled from the arm64 (Apple Silicon) runner — GitHub retired
the free Intel `macos-13` runner, but Rust cross-compiles darwin-x86_64
trivially. Not prebuilt: 32-bit targets (incl. 32-bit musl, which has no Rust
toolchain) — there, `pip install processkit-py` builds from the sdist, which
needs a [Rust toolchain](https://rustup.rs/).

---

# Troubleshooting & FAQ

[‹ docs index](./)

Use this page to map a symptom to the guide that explains the full platform or
runtime contract. The entries are deliberately short; follow the links for the
complete behavior and examples.

## `ResourceLimit` under Docker, systemd, or a non-root cgroup

On Linux, whole-tree limits require a real cgroup-v2 root; the kernel's "no
internal processes" rule rejects the setup from a container, systemd
session/scope/service, or another non-root cgroup. Check whether the process is
at the cgroup root or has delegated controllers, then see [Resource limits: the
sandbox](process-groups.md#resource-limits-the-sandbox) and [Platform support](platforms.md#resource-limits-processgroupmax_memory-max_processes-cpu_quota).

## `Unsupported` from `signal("term")` on Windows

Windows Job Objects can terminate the whole job, but they do not deliver POSIX
signals, so only `signal("kill")` is supported and `"term"` raises
`Unsupported`. Use a platform-appropriate shutdown path; see [Signalling the
whole tree](process-groups.md#signalling-the-whole-tree) and [Platform
support](platforms.md#signals-suspendresume-stats).

## A child escaped a POSIX process group with `setsid()`

`ProcessGroup.mechanism` honestly reports `"process_group"` when Linux cannot
use cgroup-v2 delegation and falls back to a POSIX process group. A child that
calls `setsid()` or `setpgid()` leaves that group, so `killpg` cannot reach it;
see [Tearing down](process-groups.md#tearing-down) for the escape boundary and
the stronger backends.

## `a`-prefixed verbs report no running asyncio event loop

The async surface is asyncio-native, so `aoutput()`, `arun()`, and related
verbs need an active asyncio loop; native trio, curio, and anyio on trio do not
provide one. Use asyncio/uvloop or the synchronous surface as appropriate; see
[Async runtimes & event loops](event-loops.md#support-at-a-glance).

## `record_replay_runner` cassette not found

The pytest fixture replays by default, so a missing cassette means it has not
been recorded at the expected fixture path. Run pytest once with
`--processkit-record` and configure `processkit_cassette_dir` if the cassette
must persist; see [Record/replay cassettes:
RecordReplayRunner](testing.md#recordreplay-cassettes-recordreplayrunner).

## Privilege drop sets `uid` but not `gid` and `groups`

Dropping only `uid` leaves the inherited supplementary groups in place, which
can retain privileges you meant to remove. Set `groups`, then `gid`, then `uid`;
see [Sandboxing untrusted tools](sandboxing.md) and [Privileges and spawn
flags](commands.md#privileges-and-spawn-flags).

## The parent is hard-killed with `SIGKILL` or `os._exit`

Normal context-manager teardown cannot run after a hard kill; only Windows Job
Objects retain a kernel-enforced whole-tree cleanup when their last handle
closes. Linux and macOS/BSD are best-effort in that case, so design the child
lifetime accordingly; see [Tearing down](process-groups.md#tearing-down) and
[Platform support](platforms.md#teardown-the-no-orphan-guarantee).

---

# API reference

The complete, per-symbol reference for the public `processkit` surface —
every class, function, protocol, type alias, and exception exported by the
package, plus the `processkit.testing` submodule.

It is generated from the type stub (`processkit/_processkit.pyi`) and the
docstrings, the same source your IDE and `mypy` read, so it cannot drift from
the real API. The narrative [guides](./) explain how the pieces compose;
this page is the exhaustive index. Both surfaces are covered together: the
synchronous verbs and their `a`-prefixed asyncio twins.


## Building & running commands

Construct a command and run it — capturing everything, or checking for success — synchronously or with the `a`-prefixed asyncio twins. `CliClient` binds a program to reusable defaults; `Pipeline` chains commands shell-free; `RunningProcess` is the live handle a started child hands back.

### `Command`

```text
Command(program: StrPath, args: Args | None = ...)
```

A command builder. Builder methods return a new `Command`.

Its `a`-verbs return custom awaitables, rather than coroutine objects:
await them directly, or pass one to ``asyncio.ensure_future(...)`` when a
Task/Future is required.

#### `arg`

```text
def arg(arg: StrPath) -> Command
```

#### `args`

```text
def args(args: Args) -> Command
```

#### `cwd`

```text
def cwd(path: StrPath) -> Command
```

#### `prefer_local`

```text
def prefer_local(dir: StrPath) -> Command
```

Search this directory before ``PATH`` when resolving a bare-name
program. Repeated calls accumulate in priority order, path-form
programs are unchanged, and the child's own ``PATH`` is not rewritten.

#### `env`

```text
def env(key: str, value: str) -> Command
```

#### `envs`

```text
def envs(vars: Mapping[str, str]) -> Command
```

#### `env_remove`

```text
def env_remove(key: str) -> Command
```

#### `env_clear`

```text
def env_clear() -> Command
```

#### `inherit_env`

```text
def inherit_env(names: Sequence[str]) -> Command
```

#### `stdin_bytes`

```text
def stdin_bytes(data: ReadableBuffer) -> Command
```

#### `stdin_text`

```text
def stdin_text(text: str) -> Command
```

#### `stdin_file`

```text
def stdin_file(path: StrPath) -> Command
```

#### `keep_stdin_open`

```text
def keep_stdin_open() -> Command
```

#### `inherit_stdin`

```text
def inherit_stdin() -> Command
```

Give the child this process's **own** stdin — it reads directly from
whatever the parent's stdin is connected to (a terminal, a file, a pipe)
instead of a crate-managed pipe. The stdin counterpart of
``stdout("inherit")``: the child *shares* the parent's stream. Reach for
it when a child must talk to the real terminal — ``git commit`` opening
``$EDITOR``, a tool prompting for a password/confirmation, or forwarding
the parent's piped stdin straight through. There is no writer to
``RunningProcess.take_stdin()`` (it raises, as for a non-kept-open run);
stdout/stderr are unaffected, so ``run()``/``output()`` still return the
child's stdout. Mutually exclusive with a *mediated* stdin — a configured
``stdin_bytes()``/``stdin_text()``/``stdin_file()`` source or
``keep_stdin_open()``: the conflict is rejected as a ``ProcessError`` at
**launch** (from the run/output verb), not when you build the ``Command``;
the live ``Runner`` and the test doubles reject it identically. Drop the
other stdin knob to resolve it.

#### `timeout`

```text
def timeout(seconds: float) -> Command
```

#### `idle_timeout`

```text
def idle_timeout(seconds: float) -> Command
```

Set an idle (inactivity) timeout: tear the child down if it produces
no stdout/stderr *line* for ``seconds`` (finite, ``> 0`` — else
``ValueError``, exactly like ``timeout``). For the "hung tool" case a
plain ``timeout`` handles poorly: a legitimately long job keeps emitting
progress, so you bound its *silence* instead of guessing a generous
wall-clock ceiling. Composes with ``timeout``; last write wins against an
earlier ``idle_timeout``.

When it fires, the streaming iterator raises ``IdleTimeout`` (a
``ProcessError`` sibling of ``Timeout`` carrying ``idle_timeout_seconds``)
— a *distinct* signal, deliberately not the wall-clock ``timed_out`` /
``Timeout``, so the two timeout classes stay tellable apart and the
captured ``timed_out`` contract is untouched (an idle-timeout never sets
it).

Enforced on the streaming/interactive surface only — ``start()`` /
``astart()`` + ``stdout_lines()`` / ``output_events()`` — since idle
monitoring rides that per-line channel. The one-shot capture verbs
(``output``/``run``/``exit_code``/``probe`` and their ``a``-twins), the
``Pipeline``, and ``Supervisor`` run entirely inside the crate, which has
no native idle-timeout to enforce, so they do **not** honor it (that needs
upstream support). Monitoring rides the streaming verbs, which the crate
gates on stdout being piped: under
``stdout_file``/``stdout("inherit")``/``stdout("null")`` both
``stdout_lines()`` and ``output_events()`` raise ``ProcessError`` ("stdout
is not piped") at setup, so an idle-timeout on a redirected stdout is
diagnosed there — never silently un-enforced. ``stderr_file`` leaves
stdout piped, so idle monitoring keeps working on the stdout channel.

#### `timeout_grace`

```text
def timeout_grace(seconds: float) -> Command
```

#### `timeout_signal`

```text
def timeout_signal(name: SignalName | int) -> Command
```

The signal sent first on a graceful timeout (default ``"term"``): a
name (``term``/``kill``/``int``/``hup``/``quit``/``usr1``/``usr2``) or a
raw platform signal number (Unix only). A raw number is validated as a
real, deliverable signal — on Unix ``1..=SIGRTMAX`` (``0``, the existence
probe that delivers nothing, negatives, and out-of-range raise
``ValueError``); on Windows a raw number raises ``Unsupported`` (only
``"kill"`` is deliverable there). A ``bool`` raises ``TypeError`` — it is
an ``int`` subtype that would otherwise silently become raw signal
``1``/``0``.

#### `no_timeout`

```text
def no_timeout() -> Command
```

#### `timeout_opt`

```text
def timeout_opt(seconds: float | None) -> Command
```

#### `cancel_on`

```text
def cancel_on(token: CancellationToken) -> Command
```

#### `success_codes`

```text
def success_codes(codes: Sequence[int]) -> Command
```

#### `retry`

```text
def retry(
    retry_if: RetryIf,
    *,
    max_retries: int | None = ...,
    initial_backoff: float | None = ...,
    multiplier: float | None = ...,
    max_backoff: float | None = ...,
    jitter: bool | None = ...,
) -> Command
```

#### `retry_never`

```text
def retry_never() -> Command
```

#### `stdout`

```text
def stdout(mode: Literal['pipe', 'inherit', 'null']) -> Command
```

#### `stderr`

```text
def stderr(mode: Literal['pipe', 'inherit', 'null']) -> Command
```

#### `encoding`

```text
def encoding(label: str) -> Command
```

#### `stdout_encoding`

```text
def stdout_encoding(label: str) -> Command
```

#### `stderr_encoding`

```text
def stderr_encoding(label: str) -> Command
```

#### `line_terminator`

```text
def line_terminator(mode: LineTerminatorName) -> Command
```

Choose where the line pump splits **both** streams into lines.
``"newline"`` (the default) splits on ``\n`` only; ``"carriage_return"``
also splits on a bare ``\r`` (one not immediately followed by ``\n``),
delivered live — for ``curl``/``pip``/``apt``-style ``\r``-redrawn
progress output that would otherwise pile up into a single line until
EOF. A ``\r\n`` pair still counts as one terminator. Shared by
``stdout_lines()``/``output_events()``, the per-line handlers
(``on_stdout_line``/``on_stderr_line``), ``stdout_tee``/``stderr_tee``,
and ``output_string`` alike; set both streams here or independently
with ``stdout_line_terminator``/``stderr_line_terminator``. Unknown
preset raises ``ValueError``.

#### `stdout_line_terminator`

```text
def stdout_line_terminator(mode: LineTerminatorName) -> Command
```

Choose where the line pump splits **stdout** into lines (see
``line_terminator``); stderr framing is left untouched.

#### `stderr_line_terminator`

```text
def stderr_line_terminator(mode: LineTerminatorName) -> Command
```

Choose where the line pump splits **stderr** into lines (see
``line_terminator``); stdout framing is left untouched. Handy when
progress output lands on stderr while stdout stays newline-structured.

#### `stdout_tee`

```text
def stdout_tee(sink: StrPath | SupportsWrite, *, append: bool = ...) -> Command
```

Tee every decoded stdout line (line + ``\n``) to ``sink`` as it is
produced, while the run *also* keeps capturing the full output (the sink
does not steal from ``ProcessResult.stdout``).

``sink`` is either a **file path** (``str`` / ``os.PathLike[str]``) or a
**Python writer** — any object with a callable ``write()`` (an
``io.StringIO``, ``sys.stderr``, a text-mode file, a logger wrapper),
picked apart by whether it exposes ``write`` (neither ``str`` nor
``pathlib.Path`` does).

- *File path:* teed as raw UTF-8 bytes, opened **at build time** (not at
  run) — created if absent and truncated, or append mode when
  ``append=True``; an unopenable path raises the matching ``OSError``
  subclass right here.
- *Writer:* each decoded line (then ``"\n"``) is passed to ``write()``
  as a ``str`` (a text sink — a binary writer whose ``write(str)`` raises
  ``TypeError`` is the wrong object here). Every write is dispatched to a
  blocking thread and awaited on the pump, so a slow ``write()`` applies
  backpressure without blocking the event loop; the object is **not**
  closed for you. ``append`` is meaningless for a writer — passing
  ``append=True`` with one raises ``ValueError``.

A write error disables the tee for the rest of the run (a ``tracing``
warning under ``enable_logging()``) while the run and its captured result
continue unaffected; a writer's ``write()`` exception is additionally
reported via ``sys.unraisablehook``. Inert unless stdout is piped through
the line pump — a no-op under ``stdout("inherit")`` / ``stdout("null")``
and under ``output_bytes()`` (raw capture).

#### `stderr_tee`

```text
def stderr_tee(sink: StrPath | SupportsWrite, *, append: bool = ...) -> Command
```

Tee every decoded stderr line to ``sink``. Same contract as
``stdout_tee`` — a file path (opened at build time, truncate by default
or ``append``) or a Python writer object with a callable ``write()`` (fed
each decoded line as a ``str`` via the same blocking-pool async-write
bridge, never closed for you), coexisting with capture, inert unless
stderr is piped through the line pump.

#### `stdout_file`

```text
def stdout_file(path: StrPath, *, append: bool = ...) -> Command
```

Redirect the child's stdout **straight to a file**, opened at spawn
time — the child writes to the file's own descriptor, with no
parent-side pump, tee, or capture buffer. The direct-redirect cousin of
``stdout_tee`` (which instead *also* captures and mirrors each line): a
``>`` / ``>>`` shell redirect, minus the shell.

The binding folds the crate's three spellings (``stdout_file`` /
``stdout_file_append`` / ``stdout_file_truncate``) into one ``append=``
kwarg, mirroring the sibling ``stdout_tee(sink, *, append=False)`` rather
than a 1:1 copy of the core's convenience aliases. ``append=False`` (the
default) **creates or truncates** the file on every spawn; ``append=True``
**creates or appends** — the mode for a shared log across ``Supervisor``
incarnations / ``retry()`` attempts (each re-run appends to the one file
with no separator).

**Opened at spawn, not now (unlike ``stdout_tee``).** Only the path is
stored; the file is opened when the command launches, so a not-yet-
existing path is not a build-time error and each re-run / retry reopens
it. An unopenable path (a missing parent directory, a permission denial)
surfaces from the run verb at launch, not from this call.

**No capture — use a non-capturing verb.** With stdout on the file there
is no pipe to read, so the capture/streaming verbs (``output()`` /
``run()`` / ``output_bytes()`` / their ``a``-twins, and ``start()`` +
``stdout_lines()`` / ``output_events()``) raise ``ProcessError`` ("stdout
is not piped … so the capture verbs have nothing to read") instead of
returning empty output — drive it with ``exit_code()`` / ``probe()``. A
later ``stdout("pipe"/"inherit"/"null")`` **clears** the redirect and
restores the ordinary stdio mode.

#### `stderr_file`

```text
def stderr_file(path: StrPath, *, append: bool = ...) -> Command
```

Redirect the child's stderr **straight to a file**, opened at spawn
time. Same contract as ``stdout_file`` — a child-owned descriptor with no
parent-side pump/tee/buffer, the path opened lazily at launch (a missing
path is not a build-time error), ``append=False`` truncating on every
spawn while ``append=True`` appends (the shared-``Supervisor``-log mode),
and a later ``stderr("pipe"/"inherit"/"null")`` clearing the redirect.

Unlike ``stdout_file``, this does **not** disable the capture verbs: only
a non-piped *stdout* gates them, so ``output()`` / ``run()`` keep working
and return the child's stdout while stderr is diverted to the file and
``result.stderr`` comes back empty.

#### `on_stdout_line`

```text
def on_stdout_line(callback: Callable[[str], None]) -> Command
```

Call ``callback`` with every decoded stdout line as it is produced —
the way to give the **synchronous** surface (``.output()``/``.run()``)
live progress observation during an otherwise-blocking call, without
losing the full capture: ``callback`` observes the same decoded lines
that land in ``ProcessResult.stdout``, it does not replace them. Also
fires on the async verbs and on a streamed run (``start()``/
``astart()`` + ``stdout_lines()``/``output_events()``) — one callback,
every path.

``callback`` is infallible: an exception raised inside it is reported
via ``sys.unraisablehook`` rather than propagated — it never derails
the run or alters the captured result. At most one handler per stream
— a repeat call replaces the previous one (builder semantics, like
``timeout()``); compose inside one callable to fan out.

Inert under ``stdout("inherit")``/``stdout("null")`` (no pump runs)
and under ``output_bytes()`` (stdout is captured raw there, bypassing
the line pump).

#### `on_stderr_line`

```text
def on_stderr_line(callback: Callable[[str], None]) -> Command
```

Call ``callback`` with every decoded stderr line as it is produced.
Same contract as ``on_stdout_line`` — full capture unaffected, fires on
sync/async/streamed paths alike, infallible (a raising callback goes to
``sys.unraisablehook``, never propagates), at most one handler per
stream.

Inert under ``stderr("inherit")``/``stderr("null")``. Unlike
``on_stdout_line``, **not** silenced by ``output_bytes()``: that verb
only bypasses the *stdout* line pump for its raw-bytes capture —
stderr still decodes through the line pump exactly as under
``output()``, so this callback still fires.

#### `kill_on_parent_death`

```text
def kill_on_parent_death() -> Command
```

Add best-effort hardening for abrupt owner death, beyond ordinary
kill-on-drop teardown. Windows already reaps the whole Job Object tree;
Linux arms ``PR_SET_PDEATHSIG`` for the direct child only (grandchildren
survive); macOS/BSD have no equivalent and treat this as a no-op. Use
``kill_on_parent_death_scope()`` to report the platform's actual reach.

#### `kill_on_parent_death_scope`

```text
def kill_on_parent_death_scope() -> str
```

The scope of parent-death cleanup this platform actually achieves
when the owner dies **abruptly** (a ``SIGKILL`` or crash, where graceful
``Drop`` teardown never runs), as a stable string:

- ``"whole_tree"`` — Windows: the kernel closes the Job Object handle on
  owner death and kill-on-close reaps the direct child and every
  descendant.
- ``"direct_child_only"`` — Linux: ``PR_SET_PDEATHSIG`` reaches only the
  direct child; grandchildren survive the owner's abrupt death.
- ``"unsupported"`` — macOS/BSD: no ``pdeathsig`` equivalent, so an
  abrupt owner death triggers no cleanup at all.

An honest capability report, not a request — read it to state the real
reach of ``kill_on_parent_death()`` (best-effort on Unix) instead of
overpromising a whole-tree guarantee the OS cannot keep. It covers only
the abrupt-death path: ordinary graceful teardown still kills the whole
tree on every platform regardless.

A ``staticmethod`` — the scope is fixed per target at build time and does
not depend on instance state or on whether ``kill_on_parent_death()`` was
called, so call it on the class (``Command.kill_on_parent_death_scope()``)
or on any instance for the same answer.

#### `create_no_window`

```text
def create_no_window() -> Command
```

#### `windows_graceful_ctrl_break`

```text
def windows_graceful_ctrl_break() -> Command
```

Windows: opt in to a **graceful teardown** — at a graceful timeout
(``timeout_grace``) or a ``ProcessGroup`` shutdown, send the direct
child a console ``CTRL_BREAK`` before the grace window, so a console
child (a CLI, Node, Python, or Go service that installs a ``CTRL_BREAK``
handler) can flush and exit cleanly ahead of the hard
``TerminateJobObject`` fallback. Without it Windows has no soft-signal
tier and a graceful timeout collapses straight to an atomic Job Object
kill; any survivor past the grace is still hard-killed.

**Boundaries.** *Console-only* — a child spawned ``create_no_window()``
(or otherwise detached) shares no console, never receives the event, and
rides the grace to the hard kill; a GUI/service parent with no console
of its own can't deliver it either. It is ``CTRL_BREAK``, **not**
``CTRL_C``, and ``timeout_signal`` does not apply. Only the **direct
child** is addressed — its own descendants get it via the shared
console/group, but an ``adopt``ed process does not. A harmless **no-op
outside Windows** (Unix's graceful tier already sends a real signal) —
unlike the POSIX-only ``uid``/``gid``/``groups``/``setsid``/``umask``,
which raise ``Unsupported`` off-platform rather than silently
no-op'ing.

#### `uid`

```text
def uid(uid: int) -> Command
```

#### `gid`

```text
def gid(gid: int) -> Command
```

#### `groups`

```text
def groups(gids: Sequence[int]) -> Command
```

#### `setsid`

```text
def setsid() -> Command
```

#### `umask`

```text
def umask(mask: int) -> Command
```

#### `priority`

```text
def priority(level: Priority) -> Command
```

#### `output_limit`

```text
def output_limit(
    *,
    max_bytes: int | None = ...,
    max_lines: int | None = ...,
    on_overflow: Literal['drop_oldest', 'drop_newest', 'error'] = ...,
) -> Command
```

#### `output`

```text
def output() -> ProcessResult
```

#### `output_bytes`

```text
def output_bytes() -> BytesResult
```

#### `run`

```text
def run() -> str
```

#### `exit_code`

```text
def exit_code() -> int
```

#### `probe`

```text
def probe() -> bool
```

#### `resolve_program`

```text
def resolve_program() -> str
```

Resolve this command's ``program`` to a concrete executable path
**without launching it** — a spawn-free, side-effect-free preflight
("is this tool installed?"). Reuses the same PATH/PATHEXT/execute-bit
lookup a real run performs — a bare name against this command's
``prefer_local()`` directories (priority order) then the effective
``PATH``, a path-form program directly — honoring a relocated child
``PATH`` (``env()``/``env_remove()``/``env_clear()``/``inherit_env()``),
so the result is exactly what a spawn of this same command would find.
Returns the resolved **absolute** path; raises ``ProcessNotFound`` (also
a ``FileNotFoundError``, with a ``searched`` diagnostic) on a miss. No
``a``-prefixed async twin — the probe is synchronous and needs no
runtime.

#### `aoutput`

```text
def aoutput() -> Awaitable[ProcessResult]
```

#### `aoutput_bytes`

```text
def aoutput_bytes() -> Awaitable[BytesResult]
```

#### `arun`

```text
def arun() -> Awaitable[str]
```

#### `aexit_code`

```text
def aexit_code() -> Awaitable[int]
```

#### `aprobe`

```text
def aprobe() -> Awaitable[bool]
```

#### `start`

```text
def start() -> RunningProcess
```

#### `astart`

```text
def astart() -> Awaitable[RunningProcess]
```

#### `unchecked_in_pipe`

```text
def unchecked_in_pipe() -> Command
```

#### `program`

```text
program: str
```

#### `arguments`

```text
arguments: list[str]
```

#### `command_line`

```text
def command_line() -> str
```

#### `pipe`

```text
def pipe(other: Command) -> Pipeline
```

### `CliClient`

```text
CliClient(
    program: StrPath,
    *,
    default_timeout: float | None = ...,
    default_env: Mapping[str, str] | None = ...,
    default_env_remove: Sequence[str] | None = ...,
    default_env_fn: Mapping[str, Callable[[], str]] | None = ...,
    default_retry_if: RetryIf | None = ...,
    default_max_retries: int | None = ...,
    default_initial_backoff: float | None = ...,
    default_multiplier: float | None = ...,
    default_max_backoff: float | None = ...,
    default_jitter: bool | None = ...,
    default_cancel_on: CancellationToken | None = ...,
    runner: RunnerLike | None = ...,
)
```

A program bound to default timeout/env/retry, run with the real
`Runner` by default or an injected `runner=` (a `ScriptedRunner` and
friends, for testable code with no real spawns). The verbs take just the
per-call arguments.

#### `command`

```text
def command(args: Args) -> Command
```

A `Command` for `program <args>`, the client's defaults pre-applied
— chain more builders, then pass it to a verb below instead of a plain
arg list. An explicit setting on it always wins over the default.

#### `run`

```text
def run(call: Args | Command) -> str
```

#### `run_json`

```text
def run_json(call: Args | Command) -> Any
```

Run like ``run`` (require a zero exit) and parse the stdout as JSON,
returning the decoded object (a ``dict``/``list``/``str``/number/
``bool``/``None``) — the ``run(...)`` + ``json.loads(...)`` +
error-attribution boilerplate the many CLIs that emit machine JSON
otherwise force on every caller. A non-zero exit raises ``NonZeroExit``
(like ``run``); stdout that does not parse raises ``InvalidJson`` (a
``ProcessError`` carrying the client's ``program`` and a bounded stdout
fragment), never a bare ``json.JSONDecodeError``. Returns ``Any``: JSON
admits any of those shapes, so narrow the result yourself.

#### `output`

```text
def output(call: Args | Command) -> ProcessResult
```

#### `output_bytes`

```text
def output_bytes(call: Args | Command) -> BytesResult
```

#### `exit_code`

```text
def exit_code(call: Args | Command) -> int
```

#### `probe`

```text
def probe(call: Args | Command) -> bool
```

#### `resolve_program`

```text
def resolve_program() -> str
```

Resolve this client's ``program`` to a concrete executable path
**without spawning it** — the client-level preflight ("is this tool
installed?"), with no side effects. Applies the client's defaults (so a
``default_env``/``default_env_fn`` that relocates ``PATH`` is honored
as at launch), then resolves via the same PATH/PATHEXT/execute-bit logic
a real run uses. Returns the resolved **absolute** path; a
``default_env_fn`` that raises or returns a non-``str`` aborts it
fail-closed (like the run verbs), and a miss raises ``ProcessNotFound``
(also a ``FileNotFoundError``, with a ``searched`` diagnostic). No
``a``-prefixed async twin — the probe is synchronous.

#### `arun`

```text
def arun(call: Args | Command) -> Awaitable[str]
```

#### `arun_json`

```text
def arun_json(call: Args | Command) -> Awaitable[Any]
```

Async counterpart of ``run_json()`` — await it for the decoded JSON
object. Same contract: a non-zero exit propagates ``NonZeroExit``,
stdout that does not parse propagates ``InvalidJson`` (never a bare
``json.JSONDecodeError``), both out of the ``await``.

#### `aoutput`

```text
def aoutput(call: Args | Command) -> Awaitable[ProcessResult]
```

#### `aoutput_bytes`

```text
def aoutput_bytes(call: Args | Command) -> Awaitable[BytesResult]
```

#### `aexit_code`

```text
def aexit_code(call: Args | Command) -> Awaitable[int]
```

#### `aprobe`

```text
def aprobe(call: Args | Command) -> Awaitable[bool]
```

### `Pipeline`

```text
class Pipeline
```

A shell-free pipeline `a | b | c`. Each stage runs in its own
kill-on-drop sub-group; checked failure, chain timeout, or cancellation fans
teardown across every sub-group, while outcome attribution follows pipefail
semantics.

By design, no `start`/`astart` — see `Command.pipe()`'s stub/binding
comment: a pipeline is a whole-chain verb, with no natural "handle to a
live chain" to hand back. Stream an individual stage by `start()`ing that
one `Command` directly instead.

#### `pipe`

```text
def pipe(other: Command) -> Pipeline
```

#### `timeout`

```text
def timeout(seconds: float) -> Pipeline
```

#### `cancel_on`

```text
def cancel_on(token: CancellationToken) -> Pipeline
```

#### `output`

```text
def output() -> ProcessResult
```

#### `output_bytes`

```text
def output_bytes() -> BytesResult
```

#### `run`

```text
def run() -> str
```

#### `exit_code`

```text
def exit_code() -> int
```

#### `probe`

```text
def probe() -> bool
```

#### `aoutput`

```text
def aoutput() -> Awaitable[ProcessResult]
```

#### `aoutput_bytes`

```text
def aoutput_bytes() -> Awaitable[BytesResult]
```

#### `arun`

```text
def arun() -> Awaitable[str]
```

#### `aexit_code`

```text
def aexit_code() -> Awaitable[int]
```

#### `aprobe`

```text
def aprobe() -> Awaitable[bool]
```

### `RunningProcess`

```text
class RunningProcess
```

A handle to a started process: stream output, write stdin, wait for exit.

Usable as a (async) context manager — exiting the block tears the process
down (a hard kill of the whole private tree for a standalone
``start()``/``astart()`` handle). ``stdout_lines()`` / ``output_events()`` /
``take_stdin()`` / ``kill()`` are *synchronous* setup calls; the
iterator / handle they return is what you await.

Every consuming verb — ``outcome``/``finish``/``output``/``output_bytes``/
``profile``/``shutdown`` — comes in a sync/async pair, like everywhere else
in this library: the bare name blocks the calling thread (via the same
interruptible driver as ``Command.output()``), the ``a``-prefixed twin is a
coroutine (``outcome``/``aoutcome`` rather than the unusable ``wait``/
``await``, since ``await`` is a reserved word). Either member of a pair
**consumes** the handle — afterwards it is spent (``pid`` and the other
getters return ``None``, and every consuming verb raises). Use whichever
matches your calling code, regardless of whether the handle came from
``start()`` or ``astart()``.

#### `pid`

```text
pid: int | None
```

#### `elapsed_seconds`

```text
elapsed_seconds: float | None
```

#### `cpu_time_seconds`

```text
cpu_time_seconds: float | None
```

#### `peak_memory_bytes`

```text
peak_memory_bytes: int | None
```

#### `stdout_line_count`

```text
stdout_line_count: int | None
```

#### `stderr_line_count`

```text
stderr_line_count: int | None
```

#### `owns_group`

```text
owns_group: bool | None
```

#### `stdout_lines`

```text
def stdout_lines() -> StdoutLines
```

#### `output_events`

```text
def output_events() -> OutputEvents
```

Interleaved stdout+stderr lines as an async iterator (call once).

Consumes both pipes, so pick this **or** ``stdout_lines()``. Report the
run afterwards with ``finish()``/``afinish()`` (outcome + stderr) or
``outcome()``/``aoutcome()`` — they report it whether you iterate to the
end or ``break`` out early. ``output()``/``output_bytes()``/``profile()``
raise once this stream has taken the run over, which it does as soon as
it observes the child exit: its stdout was streamed away and its stderr
was delivered as events, so they have nothing left to capture or
sample. Teardown is unaffected — leaving the handle's context-manager
block (or dropping it) hard-kills the whole tree even after the stream
has taken the run over, which is what stops a grandchild still holding
the pipe from outliving the block.

#### `take_stdin`

```text
def take_stdin() -> ProcessStdin
```

The writable stdin handle. Raises `ProcessError` if stdin was not kept
open (build the `Command` with ``keep_stdin_open()``) or was already
taken — so a missing setup fails here, not with a later `AttributeError`.

#### `kill`

```text
def kill() -> None
```

Begin tearing the tree down without waiting (like
``subprocess.Popen.kill()``: fire-and-forget).

#### `outcome`

```text
def outcome() -> Outcome
```

#### `aoutcome`

```text
def aoutcome() -> Awaitable[Outcome]
```

#### `finish`

```text
def finish() -> Finished
```

#### `afinish`

```text
def afinish() -> Awaitable[Finished]
```

#### `output`

```text
def output() -> ProcessResult
```

Wait for exit and capture the full `ProcessResult`; consumes the handle.

Raises `ProcessError` once an ``output_events()`` stream has taken this
run over — which that stream does as soon as it observes the child exit,
whether or not you iterated to the end: it consumed stdout, delivered
stderr as events and completed the run, so there is nothing left to
capture. Read such a run with ``finish()``/``afinish()`` or
``outcome()``/``aoutcome()`` instead. Stopping the iteration while the
child is still running leaves the run with this handle, and this still
does what it did before the processkit 3.0 migration: returns empty
captures with a real outcome. Which of the two a given ``break`` gets
depends on the child's timing, so after streaming events prefer a
finisher.

#### `aoutput`

```text
def aoutput() -> Awaitable[ProcessResult]
```

Async counterpart of `output` — same ``output_events()`` restriction.

#### `output_bytes`

```text
def output_bytes() -> BytesResult
```

Wait for exit and capture raw stdout as a `BytesResult`; consumes the
handle. Raises `ProcessError` once an ``output_events()`` stream has
taken the run over, under the same condition and for the same reason as
`output`.

#### `aoutput_bytes`

```text
def aoutput_bytes() -> Awaitable[BytesResult]
```

Async counterpart of `output_bytes` — same ``output_events()`` restriction.

#### `profile`

```text
def profile(every_seconds: float) -> RunProfile
```

Wait for exit while sampling resource usage every ``every_seconds``,
returning a `RunProfile`; consumes the handle.

Raises `ProcessError` once an ``output_events()`` stream has taken this
run over — which that stream does as soon as it observes the child exit,
whether or not you iterated to the end: the run is over, so there is no
live run left to sample. Use ``finish()``/``afinish()`` or ``outcome()``/
``aoutcome()`` for its outcome. Stopping the iteration while the child is
still running leaves the run with this handle, and this still profiles
the rest of it.

#### `aprofile`

```text
def aprofile(every_seconds: float) -> Awaitable[RunProfile]
```

Async counterpart of `profile` — same ``output_events()`` restriction.

#### `shutdown`

```text
def shutdown(grace_seconds: float) -> Outcome
```

Graceful teardown (signal -> wait ``grace_seconds`` -> hard kill),
returning the `Outcome`; consumes the handle. Only for a standalone
``start()``/``astart()`` handle — a handle from ``ProcessGroup.start()``
raises `Unsupported`; tear such a child down via the group (or `kill()`).
Named to match ``ProcessGroup.shutdown()``/``ashutdown()``.

After an ``output_events()`` stream has taken the run over the child has
already exited, so there is nothing to signal: this reports that run's
real outcome, waiting for its output to finish draining just as
``finish()`` does, rather than escalating against surviving
grandchildren. Leave the handle's context-manager block instead when a
hard bound matters more than the outcome.

#### `ashutdown`

```text
def ashutdown(grace_seconds: float) -> Awaitable[Outcome]
```

Async counterpart of `shutdown`.

## Program resolution

Resolve a program to its concrete executable path *without* launching it — a spawn-free preflight ("is this tool installed?") that reuses the same PATH/PATHEXT/execute-bit lookup a real run performs, so it never disagrees with what a spawn would find. The module-level `which` searches the process `PATH`; `Command.resolve_program()` and `CliClient.resolve_program()` additionally honor a `prefer_local` directory and a relocated child `PATH`. A miss raises `ProcessNotFound`.

### `which`

```text
def which(program: StrPath) -> str
```

## Results & outcomes

What a finished (or streamed) run reports back. A non-zero exit, a timeout, and a signal-kill are all *data* on these types — never raised by the capturing verbs.

### `ProcessResult`

```text
class ProcessResult
```

The captured result of a finished run. A non-zero exit, a timeout, and a
signal-kill are all reported as data here — never raised by `output()`.

Value semantics: `==`/`hash()` compare every field (program/stdout/stderr/
outcome/success codes — not the incidental `duration_seconds`/`truncated`).
**Not** picklable: equality also spans the configured `timeout` and accepted
`success_codes`, which processkit exposes no accessor to read back, so a
pickled result could not reconstruct them and would compare unequal to its
original for any command that set `.timeout(...)`/`.success_codes(...)`;
pickling raises `TypeError`. Pickle `result.outcome` (an `Outcome`, which
round-trips exactly — e.g. to return it from a
`concurrent.futures.ProcessPoolExecutor` worker), or persist
`result.stdout`/`.stderr`/`.code` yourself, to cross a process boundary.

#### `stdout`

```text
stdout: str
```

#### `stderr`

```text
stderr: str
```

#### `code`

```text
code: int | None
```

#### `is_success`

```text
is_success: bool
```

#### `timed_out`

```text
timed_out: bool
```

#### `signal`

```text
signal: int | None
```

#### `program`

```text
program: str
```

#### `duration_seconds`

```text
duration_seconds: float
```

#### `truncated`

```text
truncated: bool
```

#### `combined`

```text
combined: str
```

#### `diagnostic`

```text
diagnostic: str | None
```

The best human-facing message: stderr if it carries text, otherwise
stdout, otherwise ``None`` if both are blank — the same preference
order as ``NonZeroExit``/``Timeout``/``Signalled.diagnostic``.

#### `outcome`

```text
outcome: Outcome
```

The full run outcome (``code`` / ``signal`` / ``timed_out``), the
same value ``RunProfile.outcome`` and the checking-verb exceptions
expose.

#### `ensure_success`

```text
def ensure_success() -> ProcessResult
```

Raise the same exception a checking verb would if this result's
exit isn't in ``success_codes``; returns ``self`` unchanged otherwise,
so it composes: ``cmd.output().ensure_success().stdout``.

### `BytesResult`

```text
class BytesResult
```

The captured result of a run with raw-bytes stdout (`Command.output_bytes()`);
stderr stays decoded text. A non-zero exit, a timeout, and a signal-kill are
all data here, never raised.

Value semantics: `==`/`hash()` compare every field, same as `ProcessResult`.
**Not** picklable — raw stdout may not be valid UTF-8 and processkit has no
way to reconstruct one from arbitrary bytes outside a real run; pickling
raises `TypeError`. Pickle a `ProcessResult` (`Command.output()`) instead,
or persist the fields you need yourself.

#### `stdout`

```text
stdout: bytes
```

#### `stderr`

```text
stderr: str
```

#### `code`

```text
code: int | None
```

#### `is_success`

```text
is_success: bool
```

#### `timed_out`

```text
timed_out: bool
```

#### `signal`

```text
signal: int | None
```

#### `program`

```text
program: str
```

#### `duration_seconds`

```text
duration_seconds: float
```

#### `truncated`

```text
truncated: bool
```

Whether captured output was truncated by an ``output_limit(...)`` cap
— the line-captured stderr under any cap, and (since processkit 2.1.0)
the raw stdout too when an ``output_limit(max_bytes=...)`` byte ceiling
bounds it to a head/tail. A ``max_lines`` cap never truncates raw stdout
(bytes have no line count); only a ``max_bytes`` cap does.

#### `diagnostic`

```text
diagnostic: str | None
```

See ``ProcessResult.diagnostic``. Raw stdout is lossily decoded to
text for this message when stderr is blank.

#### `outcome`

```text
outcome: Outcome
```

See ``ProcessResult.outcome``.

#### `ensure_success`

```text
def ensure_success() -> BytesResult
```

See ``ProcessResult.ensure_success()``.

### `Outcome`

```text
class Outcome
```

How a process ended.

There is no `is_success` here on purpose: an `Outcome` carries no
`success_codes` context, so it cannot give the command's own success verdict
the way `ProcessResult.is_success` does. Use `exited_zero` for the literal
"exit code 0" test, or compare `code` against your accepted set.

Value semantics: `==`/`hash()` compare `code`/`signal`/`timed_out`
(equivalently, which variant this is and its payload); picklable.

#### `code`

```text
code: int | None
```

#### `signal`

```text
signal: int | None
```

#### `timed_out`

```text
timed_out: bool
```

#### `exited_zero`

```text
exited_zero: bool
```

### `Finished`

```text
class Finished
```

A process's outcome plus captured stderr (stdout was streamed).

Mirrors `Outcome`'s `code`, `exited_zero`, `timed_out`, and `signal`
directly (in addition to the nested `outcome`), so callers don't need to
reach through `.outcome` for fields they already use on `Outcome`. Like
`Outcome`, it exposes `exited_zero` (literal "exit code 0"), not an
`is_success` that would falsely imply `success_codes` were considered.

Value semantics: `==`/`hash()` compare `outcome`/`stderr`; picklable.

#### `outcome`

```text
outcome: Outcome
```

#### `stderr`

```text
stderr: str
```

#### `code`

```text
code: int | None
```

#### `exited_zero`

```text
exited_zero: bool
```

#### `timed_out`

```text
timed_out: bool
```

#### `signal`

```text
signal: int | None
```

### `RunProfile`

```text
class RunProfile
```

A resource-usage profile sampled across a run (`RunningProcess.profile`),
plus the run's `outcome` — `profile()` is a superset of `outcome()`.

Value semantics: `==`/`hash()` compare every field (`outcome`/
`duration_seconds`/`cpu_time_seconds`/`peak_memory_bytes`/`samples`; all
exact underneath, though two are exposed here as `float`). **Not**
picklable — it reports live OS resource-sampling telemetry that processkit
has no way to reconstruct outside an actual monitored run; pickling raises
`TypeError`.

#### `code`

```text
code: int | None
```

#### `signal`

```text
signal: int | None
```

#### `timed_out`

```text
timed_out: bool
```

#### `outcome`

```text
outcome: Outcome
```

#### `duration_seconds`

```text
duration_seconds: float
```

#### `cpu_time_seconds`

```text
cpu_time_seconds: float | None
```

#### `peak_memory_bytes`

```text
peak_memory_bytes: int | None
```

#### `samples`

```text
samples: int
```

#### `avg_cpu_cores`

```text
avg_cpu_cores: float | None
```

## Streaming & interactive I/O

The live handles a started `RunningProcess` hands out: async iterators over its output (line by line, or as interleaved stdout/stderr events) and a writable stdin.

### `StdoutLines`

```text
class StdoutLines
```

Async iterator over a process's stdout, line by line.

### `OutputEvents`

```text
class OutputEvents
```

Async iterator over stdout + stderr as interleaved `OutputEvent`s.

Yields **output lines only**. The underlying core stream is the child's whole
lifecycle (it also reports process start and exit), but those non-line events
are filtered out here rather than surfaced as an `OutputEvent` with an empty
``text`` — which would be indistinguishable from a real blank output line.
Process start is `RunningProcess.pid`; the exit is what the finisher you call
afterwards returns.

Draining this iterator to its end also drives the run to completion, so the
documented order — iterate fully, then ``await proc.afinish()`` (or
``aoutcome()``) — terminates. The finisher then reports that same run.

### `OutputEvent`

```text
class OutputEvent
```

One captured line and the stream it came from.

Value semantics: `==`/`hash()` compare `is_stderr`/`text`; picklable.

#### `stream`

```text
stream: Literal['stdout', 'stderr']
```

#### `is_stderr`

```text
is_stderr: bool
```

#### `text`

```text
text: str
```

### `ProcessStdin`

```text
class ProcessStdin
```

A writable handle to a running process's stdin (all methods awaitable).

#### `write`

```text
def write(data: ReadableBuffer) -> Awaitable[None]
```

#### `write_line`

```text
def write_line(line: str) -> Awaitable[None]
```

#### `send_control`

```text
def send_control(control: str) -> Awaitable[None]
```

Write one mapped control byte, e.g. ``"c"`` -> Ctrl-C (``\x03``).

This writes a byte to the child's stdin pipe, not a terminal signal;
real SIGINT/SIGTSTP delivery requires a pseudoterminal.

#### `flush`

```text
def flush() -> Awaitable[None]
```

#### `close`

```text
def close() -> Awaitable[None]
```

## Process groups

Kill-on-drop containment for a whole process tree — start children into it, signal or suspend the group, and reap the entire tree (grandchildren included) on exit. `MemberInfo` is the enriched per-member snapshot `members_info()` returns; `sample_stats` turns a one-shot `stats()` snapshot into a periodic async series for live monitoring.

### `ProcessGroup`

```text
ProcessGroup(
    *,
    max_memory: int | None = ...,
    max_processes: int | None = ...,
    cpu_quota: float | None = ...,
    shutdown_grace: float | None = ...,
    escalate_to_kill: bool | None = ...,
)
```

A kill-on-drop container for a process tree; use as a (async) context
manager. Also a `ProcessRunner` in its own right (see `_RunnerVerbs`):
its run verbs run `command` as a *shared* member of this group (not a
standalone tree) — the same verb surface `Runner`/`ScriptedRunner`/…
expose (not an `extract_runner` target, though — see `runner.rs`).

#### `mechanism`

```text
mechanism: Literal['job_object', 'cgroup_v2', 'process_group', 'unknown']
```

#### `members`

```text
def members() -> list[int]
```

#### `members_info`

```text
def members_info() -> list[MemberInfo]
```

An enriched, point-in-time snapshot of the group's members — the same
set as ``members()``, but each pid carried in a `MemberInfo` alongside
best-effort ``ppid``/``exe_name``/``start_time``. Synchronous only (the
crate offers no async twin). See `MemberInfo` for the per-field platform
matrix and the ``start_time`` opacity/pid-reuse note.

#### `signal`

```text
def signal(name: SignalName | int) -> None
```

Send a signal to every process in the tree: a name
(``term``/``kill``/``int``/``hup``/``quit``/``usr1``/``usr2``) or a raw
platform signal number (Unix only). On Windows a Job Object has no POSIX
signals, so only ``"kill"`` is deliverable and any other name/number
raises ``Unsupported``. A raw number is validated as a real, deliverable
signal (``1..=SIGRTMAX`` on Unix); ``0`` (the existence probe), negatives,
and out-of-range values raise ``ValueError`` instead of a silent no-op,
and a ``bool`` raises ``TypeError``.

#### `suspend`

```text
def suspend() -> None
```

#### `resume`

```text
def resume() -> None
```

#### `kill_all`

```text
def kill_all() -> None
```

#### `stats`

```text
def stats() -> ProcessGroupStats
```

#### `shutdown`

```text
def shutdown() -> None
```

#### `ashutdown`

```text
def ashutdown() -> Awaitable[None]
```

### `ProcessGroupStats`

```text
class ProcessGroupStats
```

A snapshot of a `ProcessGroup`'s resource usage.

#### `active_process_count`

```text
active_process_count: int
```

#### `peak_memory_bytes`

```text
peak_memory_bytes: int | None
```

#### `total_cpu_time_seconds`

```text
total_cpu_time_seconds: float | None
```

### `MemberInfo`

```text
class MemberInfo
```

An enriched, point-in-time snapshot of one member of a `ProcessGroup`'s
tree — its pid plus best-effort metadata.

The metadata-carrying companion to a bare pid from `ProcessGroup.members()`.
*Which* members appear follows the same platform matrix as ``members()`` (the
whole tree on Windows and the Linux cgroup backend; the tracked group leaders
on the POSIX process-group fallback). Every field beyond ``pid`` is
independently ``None`` wherever the platform can't report it — never a
fabricated value — and a member that exits mid-snapshot is silently omitted,
not invented.

The raw command line / environment is **deliberately never** carried, on any
platform: an argv routinely holds secrets, and redaction is the consumer's
policy to own.

Field availability (``None`` where the platform can't report it): ``ppid``,
``exe_name``, and ``start_time`` are populated on Windows, Linux, and macOS,
and are always ``None`` on the BSDs (no wired-up per-process reader).

#### `pid`

```text
pid: int
```

The member's process id — always present. Point-in-time, like a pid
from ``members()``: pair it with ``start_time`` to tell a recycled number
apart from the original process.

#### `ppid`

```text
ppid: int | None
```

The member's parent process id, or ``None`` where unreadable (always
``None`` on the BSDs).

#### `exe_name`

```text
exe_name: str | None
```

The member's short image **base name** — never a full path, and never a
command line (the crate never exposes argv/env). ``None`` where unreadable
(always ``None`` on the BSDs).

#### `start_time`

```text
start_time: int | None
```

An **opaque per-process identity token**, or ``None`` where unreadable —
**not** a wall-clock timestamp. Its unit and epoch are platform-specific
(Windows creation ``FILETIME``, 100ns intervals since 1601; Linux
``/proc/<pid>/stat`` field 22, clock ticks since boot; macOS microseconds
since the Unix epoch; always ``None`` on the BSDs), so do not interpret it
or compare it across platforms. Its sole use is pairing with ``pid``: two
snapshots whose ``pid`` **and** ``start_time`` both match name the same
process instance, telling a recycled pid apart from the original.

### `sample_stats`

```text
async def sample_stats(
    group: ProcessGroup,
    every: float,
) -> AsyncIterator[ProcessGroupStats]
```

Sample ``group.stats()`` on an interval, forever, as an async series of
`ProcessGroupStats` snapshots — a pure-Python analogue of the crate's
`ProcessGroup::sample_stats` (its `StatsSampler` borrows the group by
lifetime and has no FFI-safe equivalent here; this is plain Python built
directly on the already-public `group.stats()`, living alongside the
readiness helpers above for the same reason).

``async for snapshot in sample_stats(group, every): ...`` — the first
snapshot is taken immediately (no initial sleep), then one every ``every``
seconds, for as long as you keep consuming. There is no overall deadline;
stop by ``break``ing out of the loop or otherwise abandoning/closing the
generator yourself.

**Fused, and louder than the crate's stream.** The crate's `StatsSampler`
swallows the error on the first failed sample and just ends the series
silently — a caller has to separately call `stats()` to learn why. This
generator instead lets `group.stats()`'s own exception (a `ProcessError` —
e.g. "ProcessGroup is already closed" once the group has torn down, or an
`Unsupported`/OS-error-derived failure from the platform's resource query)
propagate out of the ``async for`` untouched — the underlying cause is
never hidden behind a quiet end-of-series. That still fuses the series:
once this generator function raises, it is exhausted by Python's own
async-generator protocol, so a further ``__anext__`` (another loop
iteration, a second ``async for`` over the same object) raises
`StopAsyncIteration` rather than calling `group.stats()` again or
replaying the same error. If the group is already closed/invalid *before
the first snapshot* (e.g. iteration starts only after `group.shutdown()`
already ran), that same exception surfaces on the very first ``async
for`` step, not silently as an empty series.

``every`` is validated up front: NaN and negative values raise
`ValueError` (the shared convention with the readiness helpers'
``timeout``/``interval``). Unlike the crate — which clamps a zero period
to 1 ms because `tokio` panics on a zero-duration interval — ``every=0``
is accepted here as-is: `asyncio.sleep(0)` has no such restriction, so it
means "sample as fast as the event loop allows," with no artificial floor.

## Supervision

Keep a command alive: restart it per a policy, with backoff and jitter, until a stop condition is met.

### `Supervisor`

```text
Supervisor(
    command: Command,
    *,
    restart: Literal['always', 'on_crash', 'never'] | None = ...,
    max_restarts: int | None = ...,
    backoff_initial: float | None = ...,
    backoff_factor: float | None = ...,
    max_backoff: float | None = ...,
    jitter: bool | None = ...,
    stop_when: Callable[[ProcessResult], bool] | None = ...,
    give_up_when: Callable[[ProcessResult | ProcessError], bool] | None = ...,
    storm_pause: float | None = ...,
    failure_threshold: float | None = ...,
    failure_decay: float | None = ...,
    capture_max_bytes: int | None = ...,
    capture_max_lines: int | None = ...,
    capture_on_overflow: Literal['drop_oldest', 'drop_newest', 'error'] | None = ...,
    health_check: Callable[[], bool] | None = ...,
    health_check_interval: float | None = ...,
    health_check_failures: int | None = ...,
    runner: RunnerLike | None = ...,
)
```

Keep a command alive: restart per policy with backoff until a stop condition.

#### `run`

```text
def run() -> SupervisionOutcome
```

#### `arun`

```text
def arun() -> Awaitable[SupervisionOutcome]
```

### `SupervisionOutcome`

```text
class SupervisionOutcome
```

The result of a `Supervisor.run()`.

Value semantics: `==`/`hash()` compare every field (`final_result` via
`ProcessResult`'s own comparison, plus
`restarts`/`stopped`/`storm_pauses`/`liveness_kills`).
**Not** picklable: its identity includes `final_result` (a `ProcessResult`),
which cannot be faithfully reconstructed from a pickle (its `timeout`/
`success_codes` have no accessor to read back), so pickling raises
`TypeError`. Read the fields you need, or pickle `final_result.outcome` (an
`Outcome`, which round-trips exactly), to cross a process boundary.

#### `final_result`

```text
final_result: ProcessResult
```

#### `restarts`

```text
restarts: int
```

#### `stopped`

```text
stopped: Literal['policy_satisfied', 'predicate', 'restarts_exhausted', 'gave_up', 'unhealthy', 'unknown']
```

#### `storm_pauses`

```text
storm_pauses: int
```

#### `liveness_kills`

```text
liveness_kills: int
```

## Cancellation

A portable cancel switch, wired into a run via `Command.cancel_on()`, `Pipeline.cancel_on()`, or `CliClient`'s `default_cancel_on=`.

### `CancellationToken`

```text
class CancellationToken
```

A cancel switch: fire it to tear down every run wired to it via
`Command.cancel_on()` / `CliClient`'s `default_cancel_on=` /
`Pipeline.cancel_on()` — surfacing `Cancelled`. Cheap to clone/share:
every clone refers to the same underlying state, so cancelling any clone
cancels every run wired to it. A cancelled token stays cancelled forever.

`child_token()` derives a separate, scoped token: it is cancelled
automatically when this one is, but cancelling it back does NOT
propagate to this token or to its other children — cancellation only
flows parent-to-child, never child-to-parent or between siblings.

#### `cancel`

```text
def cancel() -> None
```

#### `is_cancelled`

```text
def is_cancelled() -> bool
```

#### `child_token`

```text
def child_token() -> CancellationToken
```

A new token that is cancelled automatically when this one is, but
can also be cancelled independently — cancelling the child does not
affect this token or its other children.

## Batch execution

Run many commands with bounded concurrency, each result — or a `ProcessError` for a spawn/I/O failure — in its own slot. The `output_all` family is *collect-all* (every result in input order once the whole batch finishes); `aoutput_as_completed` and its `_bytes` twin instead stream each `(index, result)` pair *as it finishes*, for progress and early reaction on a large fan-out.

### `output_all`

```text
def output_all(
    commands: Sequence[Command],
    *,
    concurrency: int | None = ...,
    runner: RunnerLike | None = ...,
) -> list[ProcessResult | ProcessError]
```

Run a collect-all batch in input order. With ``concurrency=None``, use
the process-available CPU count (CPU affinity/cgroup-aware), falling back to
``4`` if it cannot be determined. A non-positive value raises `ValueError`.

### `output_all_bytes`

```text
def output_all_bytes(
    commands: Sequence[Command],
    *,
    concurrency: int | None = ...,
    runner: RunnerLike | None = ...,
) -> list[BytesResult | ProcessError]
```

Raw-bytes `output_all` with the same concurrency default and validation.

### `aoutput_all`

```text
def aoutput_all(
    commands: Sequence[Command],
    *,
    concurrency: int | None = ...,
    runner: RunnerLike | None = ...,
) -> Awaitable[list[ProcessResult | ProcessError]]
```

Async `output_all` with the same concurrency default and validation.

### `aoutput_all_bytes`

```text
def aoutput_all_bytes(
    commands: Sequence[Command],
    *,
    concurrency: int | None = ...,
    runner: RunnerLike | None = ...,
) -> Awaitable[list[BytesResult | ProcessError]]
```

Async raw-bytes batch with the same concurrency default and validation.

### `aoutput_as_completed`

```text
def aoutput_as_completed(
    commands: Sequence[Command],
    *,
    concurrency: int | None = None,
) -> AsyncIterator[tuple[int, ProcessResult | ProcessError]]
```

Run ``commands`` with bounded concurrency, yielding each ``(original
index, ProcessResult | ProcessError)`` pair **as that command finishes** —
the streaming, pure-Python counterpart to the compiled `aoutput_all`.

Where `aoutput_all` is *collect-all* (nothing is visible until the whole
batch is done), this is an async iterator — ``async for index, result in
aoutput_as_completed(commands, concurrency=8): ...`` — that hands each
result back the moment its command completes, so a large fan-out reports
progress and lets you react to early finishers instead of blocking on the
slowest command in the batch.

**Completion order, not input order.** Pairs arrive in the order their
commands *finish*, which is generally not the input order; the ``index`` (a
command's position in ``commands``) is what re-associates a result with the
command that produced it. Every command is yielded exactly once, and the
iterator is exhausted once all of them have been.

**Errors are per-slot data, not a series-ending raise** (aligned with
`output_all`): a command that fails to *spawn* — or hits an I/O error, or is
cancelled through its own `CancellationToken` — yields its `ProcessError` in
its own pair, and never short-circuits the others. A non-zero exit, a
timeout, and a signal-kill are, as everywhere in this library, *data* on a
`ProcessResult`, not errors at all.

**Hard concurrency cap.** At most ``concurrency`` commands are ever live at
once (an `asyncio.Semaphore` gates each `Command.aoutput()`), so fanning out
hundreds of commands can't exhaust file descriptors or the process table —
the same bound `aoutput_all` gives, held *while* streaming. ``concurrency``
defaults to the process-available CPU count (CPU affinity/cgroup-aware on
Python 3.13+), falling back to `os.cpu_count()` and then ``4``; this matches
the batch family. A non-positive value raises `ValueError` rather than being
silently clamped.

**No orphans on cancellation or early exit.** Cancelling the task consuming
this iterator — or simply ``break``ing out of the ``async for`` early — tears
down every command still in flight: each `Command.aoutput()` reaps its whole
process subtree (grandchildren included) on cancellation, and this iterator
drives that teardown for *all* live slots before it finishes unwinding. No
started child is left orphaned, whether the batch ran to completion, was
abandoned partway, or was cancelled outright.

Built directly on `Command.aoutput()`; unlike the compiled `aoutput_all`
family it takes no ``runner=`` double — the streaming layer is deliberately
kept minimal, so for a hermetic batch that doesn't need streaming reach for
`aoutput_all(..., runner=...)` instead. For raw ``bytes`` output (no UTF-8
decode) use the twin `aoutput_as_completed_bytes`.

### `aoutput_as_completed_bytes`

```text
def aoutput_as_completed_bytes(
    commands: Sequence[Command],
    *,
    concurrency: int | None = None,
) -> AsyncIterator[tuple[int, BytesResult | ProcessError]]
```

The raw-``bytes`` twin of `aoutput_as_completed`: the identical streaming,
concurrency-cap, per-slot-error, and no-orphan-on-cancellation contract, but
each finished command yields a `BytesResult` — raw-``bytes`` stdout for
non-UTF-8 or binary output, while stderr stays decoded text — in place of a
text `ProcessResult`, mirroring how `aoutput_all_bytes` relates to
`aoutput_all`.
With ``concurrency=None``, it uses the same process-available CPU-count
default (and fallbacks) as every other batch entry point. See
`aoutput_as_completed` for the full contract.

## Readiness helpers

Asyncio helpers that wait for a condition — a matching output line, an open TCP port, an HTTP endpoint answering with an expected status, a filesystem path or Unix-domain socket, or any polled predicate — bounded by a deadline.

### `wait_until`

```text
async def wait_until(
    predicate: Callable[[], bool | Awaitable[bool]],
    *,
    timeout: float,
    interval: float = 0.05,
) -> None
```

Poll ``predicate`` until it returns true, or ``timeout`` seconds elapse.

(Named ``wait_until``, not ``wait_for`` — the latter would collide with
``asyncio.wait_for``, whose semantics differ: it bounds one *awaitable*,
not a *polled predicate*.)

``predicate`` may be synchronous or return an awaitable. Polls every
``interval`` seconds; raises `WaitTimeout` (also a `TimeoutError`) if the
deadline passes first. A synchronous ``predicate`` runs on the event loop,
so keep it non-blocking — use an async ``predicate`` for anything that does
I/O. If ``predicate``'s awaitable is already a `asyncio.Future`/`asyncio.Task`
you own, note it is never cancelled by this helper on timeout — only
abandoned, so cancel or await it yourself afterwards if that matters.

``timeout<=0`` contract (shared with `wait_for_port` / `wait_for_line`):
at ``timeout=0``, ``predicate`` is still evaluated (at least once) before
any deadline check, so an already-true predicate succeeds instead of
failing before it was ever checked. A **negative** ``timeout`` is rejected
outright — raises `ValueError`, same as NaN — rather than being treated as
"expired" or silently accepted.

### `wait_for_line`

```text
async def wait_for_line(
    lines: AsyncIterator[Any],
    predicate: str | Callable[[Any], bool],
    *,
    timeout: float,
) -> Any
```

Consume from an async iterator until ``predicate`` matches an item.

``predicate`` is either a callable (``predicate(item) -> bool``) or, for a
`str`-yielding iterator only, a plain `str` — a shorthand for "the item
contains this substring" (``predicate in item``). Not just for
`StdoutLines`: any async iterator works (e.g. `OutputEvents`, with a
callable predicate over its `OutputEvent` items).

Returns the matching item. Raises `WaitTimeout` (also a `TimeoutError`,
carrying ``timeout_seconds``) if nothing matches within ``timeout``
seconds, or propagates whatever ``predicate`` or the iterator itself
raised (a `ProcessError` if the stream ends first) untouched — never
masked behind the timeout. Items read before the match are consumed;
iteration may continue afterward **only when a match was found** — on a
`WaitTimeout`, exactly how far the iterator advanced past the last
inspected item is unspecified (cancellation of the internal scan races the
iterator's own advancement), so don't rely on its position after a
timeout.

``timeout<=0`` contract (shared with `wait_until` / `wait_for_port`): at
``timeout=0``, the iterator is still scanned (at least one tick), so an
item that already matches (already sitting in the iterator) succeeds
instead of failing before it was ever inspected. A **negative** ``timeout``
is rejected outright — raises `ValueError`, same as NaN — rather than being
treated as "expired" or silently accepted.

### `wait_for_port`

```text
async def wait_for_port(
    host: str,
    port: int,
    *,
    timeout: float,
    interval: float = 0.05,
) -> None
```

Wait until a TCP connection to ``(host, port)`` succeeds.

Polls every ``interval`` seconds until the port accepts a connection or
``timeout`` seconds elapse, in which case `WaitTimeout` (also a
`TimeoutError`) is raised — carrying ``host``/``port`` — chained from the
last connection attempt's exception (e.g. a DNS failure survives as the
cause instead of being silently dropped).

``timeout<=0`` contract (shared with `wait_until` / `wait_for_line`): at
``timeout=0``, a connection attempt is still made (at least one), so an
already-ready port succeeds instead of failing before a connection was
ever tried — this first attempt is not cut short by the already-expired
deadline. It IS bounded, though: to a short, fixed event-loop tick (or a
smaller caller-supplied ``interval``), not left uncapped — an
unresolvable/blackhole address would
otherwise be free to block on the OS's own (much longer, or absent)
connect/DNS timeout well past the caller's requested deadline. A
**negative** ``timeout`` is rejected outright — raises `ValueError`, same
as NaN — rather than being treated as "expired" or silently accepted.

### `wait_for_http`

```text
async def wait_for_http(
    host: str,
    port: int,
    path: str = '/',
    *,
    timeout: float,
    interval: float = 0.05,
    expected_status: Container[int] | Callable[[int], bool] | None = None,
) -> None
```

Wait until an HTTP ``GET`` of ``http://host:port/path`` answers with an
acceptable status code.

A stronger readiness signal than `wait_for_port`: a server often *accepts*
TCP connections while still warming up and answering ``503``, so a bare port
probe reports ready too early. This one performs a minimal HTTP/1.1 ``GET``
(hand-rolled over `asyncio.open_connection` — no `http.client` / `urllib` /
third-party dependency) every ``interval`` seconds and succeeds only once the
response's status code is accepted.

``expected_status`` decides what "accepted" means: either a container tested
with ``in`` or a predicate ``Callable[[int], bool]`` for arbitrary logic
(e.g. ``lambda c: c == 204``). The default (``None``) accepts any 2xx code —
equivalent to passing ``range(200, 300)``. The whole request/response is
bounded by the deadline, so a server that accepts the connection but never
answers can't outlive ``timeout``.

On failure the deadline raises `WaitTimeout` (also a `TimeoutError`),
carrying ``host`` / ``port`` / ``path`` and chained (as ``__cause__``) from
the last attempt's failure — a connection error (e.g. a refused connect or a
DNS failure) or a `ProcessError` recording the last unexpected status code —
so the evidence for *why* it never became ready survives.

``timeout<=0`` contract (shared with `wait_until` / `wait_for_port` /
`wait_for_line` / `wait_for_path`): at ``timeout=0`` one request attempt is
still made (at least one), so an already-ready endpoint succeeds instead of
failing before it was ever probed; that first attempt is bounded to a short,
fixed event-loop tick (or a smaller caller-supplied ``interval``), never left
uncapped. A **negative** ``timeout`` is rejected outright — raises
`ValueError`, same as NaN — as is a non-positive ``interval``.

``host`` and ``path`` are validated up front, before any connection is
attempted (fail-fast, not "after one retry cycle"): an IPv6 literal
``host`` (e.g. ``"::1"``) is bracketed in the ``Host`` header per RFC
9112/3986 (``Host: [::1]:8080``, never the ambiguous ``Host: ::1:8080``);
a ``path`` containing whitespace or a control character (including
CR/LF — which could otherwise inject extra request/header lines from an
untrusted ``path``) raises `ValueError`; and a ``host``/``path`` with a
character that can't be encoded as latin-1 (required for the request
line) raises `ValueError` instead of a raw `UnicodeEncodeError`.

### `wait_for_path`

```text
async def wait_for_path(
    path: StrPath,
    *,
    timeout: float,
    interval: float = 0.05,
) -> None
```

Wait until ``path`` exists on the filesystem.

Polls every ``interval`` seconds until ``path.exists()`` returns true or
``timeout`` seconds elapse, in which case `WaitTimeout` (also a
`TimeoutError`) is raised, carrying ``path``. A unix-socket, a pid file, or
any other marker file a daemon creates once ready are all typical uses. For
a Unix-domain socket that must actually accept connections, use
`wait_for_unix_socket`; for a TCP port or an arbitrary predicate, see
`wait_for_port` / `wait_until` instead (`wait_until(lambda: path.exists(),
...)` is exactly what this helper does, named for readability and given the
same `WaitTimeout` discipline as its siblings).

``timeout<=0`` contract (shared with `wait_until` / `wait_for_port` /
`wait_for_line`): at ``timeout=0``, ``path`` is still checked (at least
once) before any deadline check, so an already-existing path succeeds
instead of failing before it was ever checked. A **negative** ``timeout``
is rejected outright — raises `ValueError`, same as NaN — rather than
being treated as "expired" or silently accepted.

### `wait_for_unix_socket`

```text
async def wait_for_unix_socket(
    path: StrPath,
    *,
    timeout: float,
    interval: float = 0.05,
) -> None
```

Wait until a Unix-domain socket at ``path`` accepts a connection.

Unlike `wait_for_path`, this proves that the socket has started accepting
connections, rather than only that its filesystem entry exists. Polls every
``interval`` seconds until a connection succeeds or ``timeout`` seconds
elapse, in which case `WaitTimeout` (also a `TimeoutError`) is raised,
carrying ``path`` and chained from the last connection failure.

Platforms lacking Unix-domain-socket support — no ``socket.AF_UNIX`` or no
``asyncio.open_unix_connection`` (asyncio binds the latter only when the
former existed at import) — raise `Unsupported` instead of silently
downgrading to a filesystem-existence check. At ``timeout=0`` one bounded
connection attempt still runs, so an already-ready socket succeeds; negative
and NaN timeouts are rejected with `ValueError`.

### `WaitTimeout`

```text
WaitTimeout(
    message: str,
    *,
    timeout_seconds: float,
    host: str | None = None,
    port: int | None = None,
    path: StrPath | None = None,
)
```

A readiness helper (`wait_until` / `wait_for_line` / `wait_for_port` /
`wait_for_http` / `wait_for_path` / `wait_for_unix_socket`) didn't succeed within its deadline.

Also a builtin `TimeoutError`, so `except TimeoutError` catches it too —
the same convention a run's own `.timeout()` uses (see `Timeout`). Always
carries `timeout_seconds`; `wait_for_port` and `wait_for_http` additionally
set `host` / `port` (and `wait_for_http` also `path`), while `wait_for_path`
and `wait_for_unix_socket` set `path` (all `None` for `wait_until` /
`wait_for_line`, which have none of these). `wait_for_port` /
`wait_for_http`, and `wait_for_unix_socket` also chain the last attempt's
failure as `__cause__` (a connection error, or — for `wait_for_http` — the
last unexpected status code).

#### `timeout_seconds`

```text
timeout_seconds = timeout_seconds
```

#### `host`

```text
host = host
```

#### `port`

```text
port = port
```

#### `path`

```text
path = path
```

## Observability

Opt-in bridging of the core's per-run `tracing` events to Python `logging`.

### `enable_logging`

```text
def enable_logging() -> bool
```

## The runner seam

The dependency-injection seam: annotate your code against a protocol, inject the real `Runner` in production and a test double (see the Testing section) in tests. `ProcessRunner` is the capture/check verbs; `StreamingRunner` adds `start`/`astart`.

### `ProcessRunner`

```text
class ProcessRunner
```

The capture/check run verbs as a structural type: `output`/`output_bytes`/
`run`/`exit_code`/`probe` and their `a`-prefixed async twins — no streaming.

Every built-in runner satisfies this (and the wider `StreamingRunner`).
Prefer this narrower protocol when your own code only calls these verbs —
a hand-rolled double then only needs to implement five verbs (times two
for the async twins), not the full runner surface.
`CliClient` also satisfies `ProcessRunner`: each capture/check verb accepts
either per-call `Args` (which it combines with its bound program) or a
`Command` (whose explicit settings win over client defaults). It is not a
`StreamingRunner`, because it has no `start`/`astart` verbs.

#### `output`

```text
def output(command: Command, /) -> ProcessResult
```

#### `output_bytes`

```text
def output_bytes(command: Command, /) -> BytesResult
```

#### `run`

```text
def run(command: Command, /) -> str
```

#### `exit_code`

```text
def exit_code(command: Command, /) -> int
```

#### `probe`

```text
def probe(command: Command, /) -> bool
```

#### `aoutput`

```text
def aoutput(command: Command, /) -> Awaitable[ProcessResult]
```

#### `aoutput_bytes`

```text
def aoutput_bytes(command: Command, /) -> Awaitable[BytesResult]
```

#### `arun`

```text
def arun(command: Command, /) -> Awaitable[str]
```

#### `aexit_code`

```text
def aexit_code(command: Command, /) -> Awaitable[int]
```

#### `aprobe`

```text
def aprobe(command: Command, /) -> Awaitable[bool]
```

### `StreamingRunner`

```text
class StreamingRunner
```

`ProcessRunner` plus `start`/`astart` — the full runner verb surface,
for code that also needs a live `RunningProcess` handle to stream.

`Runner`, `ScriptedRunner`, `RecordReplayRunner`, `RecordingRunner`, and
`DryRunRunner` all satisfy it. A hand-rolled double can implement the
capture/check verbs easily, but `start`/`astart` must return a
`RunningProcess`, which has no public constructor — and the built-in
runners are `@final`, so a fully-conforming custom runner in practice means
*wrapping* one (delegating `start`/`astart` to it; use `ScriptedRunner`
for streaming doubles).

#### `start`

```text
def start(command: Command, /) -> RunningProcess
```

#### `astart`

```text
def astart(command: Command, /) -> Awaitable[RunningProcess]
```

### `Runner`

```text
class Runner
```

The real process runner — inject it for testable code.

## Exceptions

Every error raised by the package descends from `ProcessError`, so a single `except ProcessError` catches them all. `Timeout`, `ProcessNotFound`, and `PermissionDenied` also subclass a builtin (`TimeoutError` / `FileNotFoundError` / `PermissionError`, each itself an `OSError`), so the stdlib `except` clauses catch them too.

### `ProcessError`

```text
class ProcessError
```

Base class for every error raised by this package.

### `NonZeroExit`

```text
class NonZeroExit
```

`run()` / `exit_code()` got a non-zero exit.

#### `program`

```text
program: str
```

#### `code`

```text
code: int
```

#### `stdout`

```text
stdout: str
```

#### `stderr`

```text
stderr: str
```

#### `stdout_bytes`

```text
stdout_bytes: bytes | None
```

#### `diagnostic`

```text
diagnostic: str | None
```

### `Timeout`

```text
class Timeout
```

A run exceeded its configured timeout.

Also a builtin `TimeoutError`, so `except TimeoutError` catches it too —
and since `TimeoutError` is itself an `OSError` subclass (as of Python
3.3), `except OSError` catches it as well (the same is true of
`ProcessNotFound`/`FileNotFoundError` and
`PermissionDenied`/`PermissionError` below — all three dual-base
exceptions are transitively `OSError`).

#### `program`

```text
program: str
```

#### `timeout_seconds`

```text
timeout_seconds: float | None
```

#### `stdout`

```text
stdout: str
```

#### `stderr`

```text
stderr: str
```

#### `stdout_bytes`

```text
stdout_bytes: bytes | None
```

#### `diagnostic`

```text
diagnostic: str | None
```

### `IdleTimeout`

```text
class IdleTimeout
```

A run produced no stdout/stderr line for its `Command.idle_timeout(...)`
window (the child went silent) and was killed.

A deliberate *sibling* of `Timeout`, not a subclass: an idle (inactivity)
timeout is a distinct condition from a wall-clock `timeout()` expiry — "the
child went silent" vs "the run took too long overall" — so `except
IdleTimeout` does not swallow a wall-clock `Timeout` and vice-versa, while
`except ProcessError` still catches both. Raised from the streaming
iterators (`stdout_lines()` / `output_events()`) on the handle from
`start()`/`astart()`; the one-shot capture verbs do not enforce
`idle_timeout` (see its docstring).

#### `idle_timeout_seconds`

```text
idle_timeout_seconds: float
```

### `Signalled`

```text
class Signalled
```

A run was killed by a signal.

#### `program`

```text
program: str
```

#### `signal`

```text
signal: int | None
```

#### `stdout`

```text
stdout: str
```

#### `stderr`

```text
stderr: str
```

#### `stdout_bytes`

```text
stdout_bytes: bytes | None
```

#### `diagnostic`

```text
diagnostic: str | None
```

### `ProcessNotFound`

```text
class ProcessNotFound
```

The program could not be found / spawned.

Also a builtin `FileNotFoundError` (what `subprocess` raises), so
`except FileNotFoundError` catches it too.

#### `program`

```text
program: str
```

#### `searched`

```text
searched: str | None
```

### `PermissionDenied`

```text
class PermissionDenied
```

The program could not be spawned because of insufficient permissions
(e.g. a non-executable file), or a permission-denied OS error surfaced
from elsewhere in the run (e.g. a group signal the OS refused).

Also a builtin `PermissionError`, so `except PermissionError` catches it too.

#### `program`

```text
program: str | None
```

### `ResourceLimit`

```text
class ResourceLimit
```

A resource limit (memory / processes / CPU) was invalid or could not be
enforced by the active containment mechanism. The reason is the exception
message (``str(exc)``); it carries no extra structured field.

### `Unsupported`

```text
class Unsupported
```

The operation is not supported on this platform.

#### `operation`

```text
operation: str
```

### `OutputTooLarge`

```text
class OutputTooLarge
```

Captured output hit an `output_limit(..., on_overflow="error")` ceiling.

``total_bytes`` (and the ``max_bytes`` ceiling it crossed) count **raw bytes
read from the child's output pipe** — line terminators and invalid-UTF-8
bytes included — not the bytes of the decoded text on
``ProcessResult.stdout``, so ``total_bytes`` can exceed
``len(stdout.encode())`` for the same run. ``total_lines`` is the line count
of the line-captured output.

#### `program`

```text
program: str
```

#### `max_lines`

```text
max_lines: int | None
```

#### `max_bytes`

```text
max_bytes: int | None
```

#### `total_lines`

```text
total_lines: int
```

#### `total_bytes`

```text
total_bytes: int
```

### `Cancelled`

```text
class Cancelled
```

The run was deliberately cancelled via a `CancellationToken` wired
with `Command.cancel_on()` / `CliClient`'s `default_cancel_on=` /
`Pipeline.cancel_on()`. Terminal — never retried by `Command.retry()` or
restarted by `Supervisor` (the token stays cancelled forever, so another
attempt could only fail the same way).

#### `program`

```text
program: str
```

### `InvalidJson`

```text
class InvalidJson
```

`CliClient.run_json()` / `arun_json()` ran the command successfully (a
zero exit, like `run`) but its stdout did not parse as JSON.

A `ProcessError` subclass raised in place of a bare `json.JSONDecodeError`,
so the failure is attributed (which program, and what the parser reported in
`str(exc)`) and a single `except ProcessError` still catches it. A deliberate
*sibling* of `NonZeroExit`, not a subclass: the run itself succeeded — only
its output *shape* is wrong — so `except InvalidJson` isolates a bad-payload
failure without also catching a genuine non-zero exit.

#### `program`

```text
program: str
```

#### `stdout`

```text
stdout: str
```

## Type aliases

Exported so your own wrappers can annotate against the same types the API accepts.

### `Args`

```text
Args = list[str] | list[Path] | list[os.PathLike[str]] | tuple[StrPath, ...]
```

### `LineTerminatorName`

```text
LineTerminatorName = Literal['newline', 'carriage_return']
```

### `Priority`

```text
Priority = Literal['idle', 'below_normal', 'normal', 'above_normal', 'high']
```

### `ReadableBuffer`

```text
ReadableBuffer = bytes | bytearray | memoryview
```

### `RetryIf`

```text
RetryIf = Literal['transient', 'transient_or_timeout']
```

### `SignalName`

```text
SignalName = Literal['term', 'kill', 'int', 'hup', 'quit', 'usr1', 'usr2']
```

### `StrPath`

```text
StrPath = str | os.PathLike[str]
```

## Testing

Runner test doubles, in the `processkit.testing` submodule. Inject one in tests — all satisfy the `ProcessRunner` protocol — so the code under test spawns no real processes.

### `ScriptedRunner`

```text
class ScriptedRunner
```

A scripted test double for `Runner`.

#### `on`

```text
def on(prefix: Args, reply: Reply) -> None
```

#### `on_sequence`

```text
def on_sequence(prefix: Args, replies: Sequence[Reply]) -> None
```

#### `when`

```text
def when(predicate: Callable[[Command], bool], reply: Reply) -> None
```

#### `fallback`

```text
def fallback(reply: Reply) -> None
```

### `RecordReplayRunner`

```text
class RecordReplayRunner
```

Records real runs to a cassette file (`record`) and replays them without
spawning (`replay`); shares the `Runner` run-verb surface.

#### `record`

```text
def record(path: StrPath) -> RecordReplayRunner
```

#### `replay`

```text
def replay(path: StrPath) -> RecordReplayRunner
```

#### `save`

```text
def save() -> None
```

### `RecordingRunner`

```text
class RecordingRunner
```

A recording test double: replies to every command with a canned `Reply`
and records each call, so a test can assert on what its code ran. Shares the
`Runner` run-verb surface; inspect calls with `calls()` / `only_call()`.

#### `replying`

```text
def replying(reply: Reply) -> RecordingRunner
```

#### `new`

```text
def new(inner: RunnerLike) -> RecordingRunner
```

#### `calls`

```text
def calls() -> list[Invocation]
```

#### `only_call`

```text
def only_call() -> Invocation
```

### `DryRunRunner`

```text
class DryRunRunner
```

A dry-run test double: never spawns a process. Every verb renders the
command to its display-quoted line (like `Command.command_line()`) and
returns a synthetic successful result — the seam behind a tool's own
`--dry-run`/`--echo` mode. Shares the `Runner` run-verb surface; inspect the
rendered lines with `commands()` / `only_command()`, or stream them live
with `on_invocation()`.

#### `on_invocation`

```text
def on_invocation(callback: Callable[[str], None]) -> None
```

#### `commands`

```text
def commands() -> list[str]
```

#### `only_command`

```text
def only_command() -> str
```

### `Reply`

```text
class Reply
```

A canned reply for a `ScriptedRunner` rule.

#### `ok`

```text
def ok(stdout: str) -> Reply
```

#### `fail`

```text
def fail(code: int, stderr: str) -> Reply
```

#### `timeout`

```text
def timeout() -> Reply
```

#### `signalled`

```text
def signalled(signal: int | None = ...) -> Reply
```

#### `pending`

```text
def pending() -> Reply
```

#### `lines`

```text
def lines(lines: Sequence[str]) -> Reply
```

#### `with_stdout`

```text
def with_stdout(stdout: str) -> Reply
```

#### `with_stderr`

```text
def with_stderr(stderr: str) -> Reply
```

#### `with_line_delay`

```text
def with_line_delay(seconds: float) -> Reply
```

### `Invocation`

```text
class Invocation
```

One call captured by a `RecordingRunner`: the program, args, cwd, env
overrides, and whether stdin was supplied. Values are inspectable for
assertions; the `repr` stays redacted (program, arg count, cwd, env names,
has_stdin — never argv or env values).

#### `program`

```text
program: str
```

#### `args`

```text
args: list[str]
```

#### `cwd`

```text
cwd: str | None
```

#### `env`

```text
env: dict[str, str | None]
```

#### `env_is`

```text
def env_is(name: str, value: str) -> bool
```

#### `has_env`

```text
def has_env(name: str) -> bool
```

#### `has_stdin`

```text
has_stdin: bool
```

#### `has_flag`

```text
def has_flag(flag: str) -> bool
```

---

# Architecture

This page is for contributors: how the binding is put together, where the line
between "binding" and "crate" runs, and the conventions that keep the two
layers — and the three parallel views of the public API — in sync. The user-
facing guides (linked from the [docs home](./)) explain *what* the
library does; this page explains *how the code that implements it is
organized*.

## Two layers, one boundary

**processkit-py is a thin PyO3 binding to the [`processkit`](https://crates.io/crates/processkit)
Rust crate — not a reimplementation.** Concretely:

```text
┌───────────────────────────────────────────────────────────────────┐
│  Python package (src/processkit/)                                 │
│  __init__.py facade · _aio.py · _protocols.py · _types.py         │
├───────────────────────────────────────────────────────────────────┤
│  Binding crate (src/*.rs) — cdylib `_processkit`                  │
│  pyclasses/verbs, error mapping, runtime driving — thin glue only │
├───────────────────────────────────────────────────────────────────┤
│  `processkit` crate (crates.io, pinned exact version)             │
│  ALL platform logic: Windows Job Objects, Linux cgroup v2,        │
│  POSIX process groups, race-free spawn, async-throughout (tokio)  │
└───────────────────────────────────────────────────────────────────┘
```

Everything that decides *how a process tree is actually contained and torn
down on a given OS* — Job Object completion ports on Windows, cgroup v2 on
Linux, process-group fallbacks, the race-free spawn sequencing — lives in the
`processkit` crate (see its own docs at [docs.rs/processkit](https://docs.rs/processkit)).
The binding crate (`src/*.rs`, compiled to the cdylib `_processkit`) never
reimplements any of that; it exists solely to:

- expose the crate's types as PyO3 pyclasses with a Python-shaped verb surface
  (kwargs, `str`/`os.PathLike`, sync **and** async pairs),
- drive the crate's `async`-throughout futures to completion from Python's
  sync and async worlds (the `runtime.rs` trio, below),
- map the crate's single `processkit::Error` onto a typed Python exception
  hierarchy (`errors.rs`'s `map_err`, below),
- and re-export a small amount of pure-Python convenience on top (`src/processkit/`,
  further below) that composes on the compiled surface instead of touching the
  OS itself.

If you find yourself teaching the binding crate a new fact about an OS
mechanism, that fact almost certainly belongs upstream in `processkit`
instead — bump the pinned crate version and bind the new capability, don't
duplicate it here.

## The Rust module map

`src/lib.rs` is the `#[pymodule(gil_used = false)]` entry point. It declares no
logic of its own beyond calling each module's `register(m)` — registration is
delegated so that adding a new pyclass or function touches only its own
module, not this central list:

```rust
mod batch;       mod cancellation; mod cli;      mod command;
mod convert;     mod errors;       mod group;    mod logging;
mod result;      mod runner;       mod running;  mod runtime;
mod supervisor;
```

| Module | Owns |
|---|---|
| `command.rs` | The `Command` builder and shell-free `Pipeline`. |
| `runner.rs` | The runner seam: `Runner`, the `ScriptedRunner`/`RecordReplayRunner`/`RecordingRunner`/`DryRunRunner` test doubles, the `Reply` builder, and the `runner_pymethods!` macro (below). |
| `running.rs` | The async streaming/interactive handles: `RunningProcess`, `ProcessStdin`, `StdoutLines`, `OutputEvents`. |
| `group.rs` | The `ProcessGroup` containment container and its `ProcessGroupStats`. |
| `supervisor.rs` | The `Supervisor` (restart/backoff) and its `SupervisionOutcome`. |
| `cli.rs` | `CliClient` — a program plus default timeout/env/retry, with verbs that take just per-call args. |
| `batch.rs` | Module-level batch execution: many `Command`s with bounded concurrency. |
| `result.rs` | The captured-result value types: `ProcessResult`, `BytesResult`, `Outcome`, `OutputEvent`, `Finished`, `RunProfile`. |
| `cancellation.rs` | `CancellationToken`, a portable cancel switch shared by `Command`/`CliClient`/`Pipeline`. |
| `logging.rs` | Opt-in bridge forwarding the crate's `tracing` events to Python's `logging`. |
| `convert.rs` | Small converters from Python-facing strings/numbers to crate types (durations, encodings, retry policy). |
| `errors.rs` | The exception hierarchy and `map_err` — the single crate-error → Python-exception funnel (below). |
| `runtime.rs` | The single tokio runtime and the interruptible blocking driver (`block_on` / `drive_async` / `block_on_interruptible`, below). |

This table is a map, not a promise: consult each module's own doc comment for
the authoritative, current description.

`gil_used = false` opts the module into PEP 703 free-threaded CPython (on a
free-threaded build, importing it does not force the GIL back on). This is
sound only because the binding holds no unsynchronized shared state — see
`lib.rs`'s own comment for the itemized reasons (the tokio runtime is a
managed singleton, exception caches use `PyOnceLock`, stream handles are
`Arc<Mutex<…>>`, the stateful pyclasses that carry consumable/reconfigurable
state — `ProcessGroup`, `RunningProcess`, `ScriptedRunner`, `DryRunRunner` —
are `#[pyclass(frozen)]` with an interior `std::sync::Mutex` that serializes
cross-thread access, and the remaining immutable pyclasses lean on PyO3's own
per-object borrow checking). Keep that invariant in mind before adding any new
shared mutable state to a pyclass.

## The call flow: Python → crate → typed exception

Every consuming verb (`output`, `run`, `exit_code`, `probe`, `start`, and their
`a`-prefixed async twins) funnels through the same shape:

```text
Python call
  │
  ▼
PyO3 pyclass method (#[pymethods], e.g. PyCommand::output / Runner::aoutput)
  │
  ▼
crate future (processkit::Command::output_string(&cmd), etc. — async-throughout)
  │
  ▼
runtime.rs: block_on(...) [sync verbs]  or  drive_async(...) [async verbs]
  │                                            │
  │  block_on_interruptible: GIL released,      │  PyLazyFuture starts work on
  │  polls the future on a fixed tick so a       │  tokio; completion writes to a
  │  blocked Ctrl+C still raises on the main      │  socket and loop.sock_recv
  │  thread; a reentrant call from inside the     │  resolves the Python Future on
  │  runtime (e.g. a Supervisor stop_when         │  the event-loop thread
  │  thread; a reentrant call from inside the     │
  │  runtime (e.g. a Supervisor stop_when         │
  │  callback) is rejected with a clear error      │
  │  instead of panicking tokio                    │
  ▼                                            ▼
Result<T, processkit::Error>
  │
  ▼
errors.rs: map_err(error) -> PyErr   (the ONLY place a crate Error becomes a PyErr)
  │
  ▼
Typed Python exception (ProcessError subclass, or a dual-base one like
Timeout/ProcessNotFound/PermissionDenied that also inherits a builtin)
```

Two invariants worth internalizing when adding a new verb:

- **`runtime.rs` is the only place a future is driven** (`block_on`,
  `drive_async`, and the lower-level `block_on_interruptible` that both build
  on). A new verb should call one of these three, never hand-roll its own
  `rt().block_on(...)` — that's how the reentrancy guard and the Ctrl+C
  polling stay uniform across the whole surface.
- **`map_err` is the only funnel from `processkit::Error` to `PyErr`.** It
  picks the exception class from the error's own accessors
  (`is_timeout()`/`is_not_found()`/`is_permission_denied()`, falling back to a
  match on the enum variant for the rest) and attaches the structured fields
  (`code`, `stdout`, `stderr`, `program`, `signal`, `timeout_seconds`,
  `diagnostic`, output-cap counters) via `setattr`. A new crate error variant
  is covered automatically as long as it exposes the right accessor; no other
  module should construct a `ProcessError` subclass by hand from a crate error.

## Conventions

- **Per-module `register(m)`.** Every `src/*.rs` module exposes
  `pub(crate) fn register(m: &Bound<'_, PyModule>) -> PyResult<()>` that adds
  its own classes/functions (and, for `errors.rs`, the whole exception
  hierarchy). `lib.rs` only calls each `register`; it never lists an
  individual class or function itself. Adding a pyclass or function means
  adding it to its module's `register`, nothing in `lib.rs`.
- **`runner_pymethods!` (`runner.rs`).** PyO3's `multiple-pymethods` feature is
  off, so a pyclass may have only one `#[pymethods]` impl. Five runner
  pyclasses (`Runner`, `ScriptedRunner`, `RecordReplayRunner`,
  `RecordingRunner`, `DryRunRunner`) each need the identical twelve-verb
  surface (`output`/`output_bytes`/`run`/`exit_code`/`probe`/`start`, times
  their `a`-prefixed async twins) forwarding to the generic `runner_*` helper
  functions over `ProcessRunner`. The macro splices that shared block together
  with each type's own unique members (constructor, builders, `__repr__`)
  passed in as a token tree, so the run-verb surface has a single source of
  truth instead of five hand-copied blocks that could drift.
- **Config struct → kwargs, not a mirror pyclass.** When the crate exposes a
  builder/options struct (e.g. `ProcessGroupOptions`), the binding does *not*
  create a matching Python class for it. Instead the pyclass constructor takes
  the options as `#[pyo3(signature = (*, field=None, ...))]` keyword
  arguments, builds the crate's options struct from defaults, and applies only
  what was actually passed (see `PyProcessGroup::new` in `group.rs`). This
  keeps the Python surface flat (`ProcessGroup(max_memory=..., cpu_quota=...)`)
  instead of forcing callers to construct and thread through a second object.
- **Sync/async verb parity (the `a`-prefix).** Every consuming verb ships as a
  pair: a blocking one (`output`, `run`, `start`, …) and an `a`-prefixed
  asyncio one (`aoutput`, `arun`, `astart`, …) that accepts the identical
  arguments and returns the identical wrapped type. This holds across
  `Command`/`Pipeline`, every runner (real and test doubles), `RunningProcess`,
  `ProcessGroup`, `Supervisor`, and `CliClient`. A new verb should ship both
  halves together, wired through `block_on`/`drive_async` respectively.
  `drive_async` returns a **lazy** awaitable (`PyLazyFuture`): it schedules
  nothing until the first `await`, so an `a`-verb built but never awaited starts
  no work and, when dropped, releases what it captured (and tears down a process
  it already owns). Once awaited it delegates to a real `asyncio.Future` for
  cancellation. Tokio stores the completed outcome in Rust memory and wakes
  one shared per-loop `sock_recv` dispatcher; value conversion and Future
  resolution happen on the event-loop thread, so completion never attaches to
  Python from a foreign runtime thread and repeated stream steps do not allocate
  a socket each.
- **`gil_used = false`.** See the free-threading note above — a deliberate,
  narrowly-justified opt-in, not a default to imitate carelessly in a module
  that *does* need shared mutable state outside PyO3's own guarding.

## The pure-Python layer (`src/processkit/`)

Alongside the compiled `_processkit` extension, a small amount of hand-written
Python composes on top of it rather than adding more Rust surface:

- **`_aio.py`** — asyncio readiness helpers (`wait_until`, `wait_for_line`,
  `wait_for_port`, `wait_for_path`) and `WaitTimeout`. These compose on the
  already-compiled async surface (a `StdoutLines` iterator, a plain TCP
  connect) instead of bridging the crate's own probing methods, which keeps
  them simpler and usable against *any* server, not only one this package
  started. It also holds `sample_stats(group, every)`, a periodic
  `ProcessGroupStats` series built directly on `ProcessGroup.stats()` — the
  crate's own `StatsSampler` borrows the group by lifetime and has no
  FFI-safe equivalent, so this is plain Python for the same reason as the
  readiness helpers.
- **`_protocols.py`** — the `ProcessRunner` / `StreamingRunner` `Protocol`
  classes: the typed dependency-injection seam that lets code written against
  "a runner" accept the real `Runner`, any of the test doubles, or a
  hand-rolled double, all checked structurally by the type checker.
- **`_types.py`** — the public type aliases (`StrPath`, `Args`, `SignalName`,
  `RetryIf`, `ReadableBuffer`, `LineTerminatorName`, `Priority`) exported so
  callers can annotate their own wrappers with the same vocabulary the API
  uses.
- **`__init__.py`** — the facade. It re-exports the compiled classes/functions
  from `_processkit` together with the pure-Python helpers above, and its
  `__all__` list **is** the public surface: anything not listed there is not
  public, regardless of what's importable by digging into a submodule. The
  test-double runners (`ScriptedRunner`, `RecordReplayRunner`,
  `RecordingRunner`, `DryRunRunner`, `Reply`, `Invocation`) are deliberately
  excluded from the top-level `__all__` and re-exported instead from
  `processkit.testing`, so the production surface and the testing surface
  stay visibly separate.

## Guarding against drift: the stub/runtime/surface triangle

The compiled module (`_processkit`), the hand-written type stub
(`src/processkit/_processkit.pyi`), and the package's `__all__` re-exports are
three independent, hand-maintained mirrors of one surface. Nothing keeps them
in sync automatically — a renamed method, a new pyclass, or a dropped kwarg
default can drift silently in any one of them. Two independent mechanisms
catch that:

1. **`tests/test_api_surface.py`** is an AST-based drift guard, run as part of
   the normal test suite. It parses `_processkit.pyi` and compares it against
   the compiled module at runtime: every compiled class/function must be
   stubbed (and vice versa), every class's members must match (name, and
   property-vs-method kind), every `__all__` must be sorted/unique/importable
   and cover every compiled export and every shim module's own `__all__`, the
   (async) context-manager dunders must be declared where promised, and every
   exported exception must remain a `ProcessError` subclass. A dedicated test
   (`test_signature_parameters_match_the_stub`) additionally compares each
   callable's actual *parameter list* (name, kind, whether it has a default)
   against the stub's — catching a renamed/reordered kwarg or a dropped
   default that the name-only checks can't see.
2. **`stubtest` (mypy.stubtest)**, run in CI's `typecheck` job
   (`uv run python -m mypy.stubtest processkit --ignore-disjoint-bases
   --allowlist stubtest-allowlist.txt`), checks the stub against the compiled
   module from the opposite direction — signature *shape* (parameter
   names/kinds/defaults) and member existence both ways, at a level
   `test_api_surface.py`'s hand-written checks don't reach. `stubtest-allowlist.txt`
   suppresses only the small set of irreducible false positives this pairing
   produces (PyO3's `__new__`-only construction vs. the stub's `__init__`
   form, module-level `Literal` aliases stubtest doesn't recognize as such,
   and the compiled module's own auto-generated `__all__`) — every entry there
   documents *why* it's a false positive, not a real gap, and an unused entry
   fails CI (`--ignore-disjoint-bases` is passed but
   `--ignore-unused-allowlist` is not), so a stale suppression surfaces on its
   own.

**When you add a new pyclass, method, property, or module-level function:**
add it to the `#[pymethods]`/`#[pyfunction]` in Rust, add the matching
declaration to `_processkit.pyi`, and re-export it (top-level `__init__.py`
for production surface, `processkit/testing.py` for a test double) if it's
meant to be public. Run `uv run pytest tests/test_api_surface.py` and
`uv run python -m mypy.stubtest processkit --ignore-disjoint-bases --allowlist
stubtest-allowlist.txt` locally (both also run in CI) before opening a pull
request — they will fail loudly, and specifically, if any of the three views
disagree.

## Rust unit tests (`cargo test`) vs. the Python suite (`tests/`)

The binding has two independent levels of test coverage, split by what they
can exercise without a live Python interpreter:

- **Rust `#[cfg(test)]` modules** (`src/convert.rs`, `src/supervisor.rs`) cover
  the crate's *pure, PyO3-free helpers* — string/number parsing
  (`parse_priority`, `parse_signal`/`parse_signal_name`, `parse_overflow_mode`,
  `parse_line_terminator`, `parse_restart_policy`, `stop_reason_str`) and
  boundary-value validation (`positive_duration`/`nonnegative_duration`'s
  NaN/infinite/negative/overflowing-`Duration` cases,
  `build_output_buffer_policy`'s cap combinations). These are cheap to write
  and run per-case (every named preset, every alias, the unknown-name
  rejection), which the Python suite can only reach indirectly and rarely
  exhaustively. `cargo test` runs them **without** the `extension-module`
  feature — the crate is deliberately structured (see the `[features]` comment
  in `Cargo.toml`) so `cargo test`/`cargo check` work without ever linking as a
  Python extension; the handful of these tests that do need the GIL (e.g.
  `parse_signal`'s `Bound<'_, PyAny>` argument) call `Python::initialize()`
  first, since nothing else brings up the interpreter in a plain test binary.
  `cargo test` runs in CI's `rust-lint` job alongside `cargo fmt`/`cargo
  clippy`.
- **`tests/` (pytest, `uv run pytest`)** covers everything that needs PyO3, the
  GIL, or a real child process/event loop: the compiled classes' behavior
  (`Command`, `Pipeline`, `ProcessGroup`, `Supervisor`, `CliClient`, the runner
  doubles), the sync/async verb pairs, exception mapping, the stdout/stderr
  capture and tee pipeline, and the stub/runtime/surface drift guards above.
  This is also where a parsing helper's *observable* behavior through the
  Python-facing API is covered end-to-end (e.g. `Command.priority("bogus")`
  raising `ValueError`), even though the exhaustive boundary-value cases for
  the helper itself live in the Rust tests instead.

When adding a new pure helper to `convert.rs`/`supervisor.rs`, prefer a Rust
`#[cfg(test)]` case for its boundary values; reach for a Python test only for
behavior that's actually observable through the compiled API (an exception's
type/message, a builder's resulting policy) rather than the helper's internals
directly.