Async-and-sync child-process management for Python with a kernel-backed no-orphan guarantee: every process you start — and everything it spawns — lives in a kill-on-exit container (a Windows Job Object, a Linux cgroup v2, or a POSIX process group). Subject to the documented POSIX escape caveats, normal completion, errors, timeouts, cancellation, and context-manager exit reap descendants as a unit. Abrupt owner-death coverage is platform-specific and reported explicitly.

Beyond spawning a subprocess: run-and-capture, line streaming, interactive stdin, opt-in pseudo-terminals, shell-free pipelines, readiness probes, timeouts & cancellation, supervision with restart/backoff, resource-limited sandboxes, and a mockable runner seam for subprocess-free tests — each in a synchronous and an asyncio-native form.

from processkit import Command

# Require success and get trimmed stdout; a failure raises a typed exception.
version = Command("python", ["--version"]).run()
print(version)

Why processkit?

subprocess and asyncio.subprocess reach (at most) the direct child. The processes it spawned — a build tool's compiler children, the real payload behind a wrapper (cmd /c …, sh -c …), 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 — a Job Object on Windows, a cgroup v2 on Linux (with a process-group fallback), a POSIX process group on macOS/BSD — so teardown is a kernel operation over the whole tree, not a best-effort signal to one pid:

• Nothing escapes silently. Exiting a with / async with block reaps every descendant, grandchildren included. Where a mechanism has a genuine weakness (a setsid child can escape a POSIX process group), ProcessGroup.mechanism reports the active backend instead of pretending — never a silent downgrade.
• Sync and async, first-class. The run-&-capture verbs, pipelines, and supervision each exist as a plain synchronous call and an a-prefixed asyncio coroutine, sharing one set of types. The inherently-streaming surfaces — live line streaming, interactive stdin, readiness probes — are asyncio-native (awaited on a started process), not duplicated as blocking calls.
• Honest results. A non-zero exit is data (ProcessResult) until you ask for success; a timeout is captured in the result; a cancellation is always an error; every platform divergence raises Unsupported or is documented. Raised exceptions carry structured fields and alias the stdlib's (Timeout is a TimeoutError, ProcessNotFound a FileNotFoundError, PermissionDenied a PermissionError).
• Testable. One runner seam swaps the real spawner for scripted doubles or record/replay cassettes — no subprocess in your tests.

How it compares

The first column is the differentiator: a child's descendants are contained and reaped as a unit (Job Object / cgroup v2 / process group), not just the direct child.

Stable API. The public API has been stable since 1.0 and follows Semantic Versioning: breaking changes land only in a new major version, so 1.x upgrades are backward-compatible. See CHANGELOG.md, and ROADMAP.md for how it was built.

Guides

New here? Start with the Cookbook — short task-to-snippet recipes for everything the package does — then read Running commands end to end (it's the vocabulary every other guide builds on). Coming from the standard library? Coming from subprocess maps your existing subprocess / asyncio.subprocess patterns onto their processkit equivalents. Reach for the rest as the need arises, and keep Platform support handy before you ship: it collects every per-OS caveat in one place.

Packaging

Unlike the Rust crate's compile-time feature flags, the Python wheel ships one surface with everything enabled — resource limits, signals/stats, record/replay, and opt-in logging are all present in every published wheel. There is nothing to opt into at install time:

pip install processkit-py   # import name: processkit

On a platform without a prebuilt wheel, build from source (uv run maturin develop) — see the README.

Distributed as abi3 wheels for CPython 3.10+ (one wheel per OS/arch runs on every supported minor version, 3.14 included), plus a version-specific free-threaded wheel for CPython 3.14t (PEP 703). See Platform support for the wheel matrix and the free-threaded note.

The 60-second tour

import asyncio
from processkit import Command, ProcessGroup

# One-shot, sync: capture everything. A non-zero exit is data, not an exception.
head = Command("git", ["rev-parse", "HEAD"]).output()
print("HEAD =", head.stdout.strip(), head.code)

# Success-checking: a non-zero exit / timeout / signal-kill becomes a typed exception.
version = Command("python", ["--version"]).run()


async def main():
    # Asyncio: the same verbs with an `a` prefix; cancelling reaps the whole tree.
    result = await Command("git", ["status", "--short"]).aoutput()

    # Stream a child's stdout; the context manager reaps the tree on exit.
    async with await Command("my-build", ["--watch"]).astart() as proc:
        async for line in proc.stdout_lines():
            print(line)

    # A managed PTY for tools that buffer behind pipes or require a tty.
    async with await Command("interactive-tool").pty(cols=120, rows=40).astart() as pty_proc:
        pty_proc.resize_pty(160, 50)

    # Containment: anything started in the group dies with it (grandchildren too).
    async with ProcessGroup() as group:
        await group.astart(Command("dev-server"))
    # async-with exit reaps the whole tree


asyncio.run(main())

API reference

The API reference is the complete, per-symbol index of the public surface — every class, function, protocol, type alias, and exception, plus the processkit.testing submodule. It is generated straight from the type stubs and docstrings (the same source your IDE and mypy read), so it never drifts from the real API. These guides are the narrative layer on top — they explain how the pieces compose, with the platform fine print collected in Platform support. The underlying algorithms (the OS containment mechanisms, race-free spawn) live in the processkit Rust crate.

For LLM assistants

This site publishes two machine-readable files following the llmstxt.org convention, so an AI assistant can pull an accurate, current slice of the docs instead of guessing from a stale summary:

• llms.txt — an annotated index of every guide, in reading order, each linked with a one-line description drawn from the guide itself.
• llms-full.txt — the full text of all guides concatenated, for loading the whole documentation set into an assistant's context at once.

Both are generated straight from these guides (scripts/gen_llms_txt.py, drift-guarded like the API reference) and served as static assets of this book.

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, Process groups, Streaming & interactive I/O, Pipelines, Timeouts & cancellation, Supervision, Testing your code, and Platform support.

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 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:

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:

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:

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

Check whether a command succeeds

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:

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

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

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

# 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

# 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

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):

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:

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.

Stream output line by line (async)

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:

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

Clean ANSI/VT escapes from PTY output

TTY-sensitive tools may emit colors, cursor movement, and OSC metadata into a PTY's merged output. Sanitize the captured and streamed text before logging, parsing, or asserting on it:

from processkit import Command

result = Command("colorful-tool", ["status"]).pty().sanitize_vt().output()
print(result.stdout)  # plain text; terminal escape sequences removed

In pipe mode, use stdout_sanitize_vt() or stderr_sanitize_vt() to clean only one captured stream. Sanitization runs after decoding and line splitting. A simultaneous stdout_tee() still receives the original escape-laden decoded lines, so it can remain a faithful terminal log while result.stdout is clean.

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:

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:

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 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:

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 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:

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)

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:

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:

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

If another library already started the process, adopt it by pid instead of recreating the launch:

import subprocess

from processkit import ProcessGroup, Unsupported

external = subprocess.Popen(["dev-server"])
try:
    with ProcessGroup() as group:
        try:
            group.adopt_external(external.pid)
        except Unsupported:
            raise RuntimeError("pid-only adoption is unsupported on this platform")
        print(external.pid in group.members())
finally:
    if external.poll() is None:
        external.kill()
    external.wait()

The group now covers the adopted process for signalling and teardown, but it does not reap it or expose an exit status. The original parent still calls wait(). A process's future descendants are included by Windows Job Objects and Linux cgroup v2; the POSIX process-group fallback normally tracks only the adopted process. See Existing processes and containment for pid identity, platform errors, and the host-level alternative.

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:

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

Wait for a server to be ready

from processkit import (
    Command,
    ProcessGroup,
    wait_until,
    wait_for_named_pipe,
    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.stderr_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 Windows named pipe (a busy server is ready too):
    # await wait_for_named_pipe(r"\\.\pipe\my-server", 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:

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.

On Windows, use the symmetric named-pipe probe for services that publish a pipe instead of a Unix socket:

from processkit import wait_for_named_pipe

await wait_for_named_pipe(r"\\.\pipe\my-daemon", timeout=10, interval=0.05)

An occupied pipe counts as ready because its server is live; on non-Windows platforms this probe raises Unsupported.

Build a shell-free pipeline

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). If Pipeline.timeout() fires, its capture verbs retain the best-effort stdout and stderr already emitted by the last stage before the deadline. A pipeline has no output_limit cap of its own — bound a flooding pipeline with timeout(). That whole-chain timeout is distinct from a per-stage Command.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):

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) 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:

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 breaking 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.

Runnable version: examples/08_batch_as_completed.py.

Wrap a CLI tool

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

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. For a one-off call, Command.run_json() / arun_json() decode it directly; CliClient.run_json() / arun_json() add reusable program defaults. Both run like run (requiring a zero exit) and hand back the already-parsed object, so you skip the run(...) + json.loads(...) + error-mapping boilerplate:

from processkit import CliClient, Command, InvalidJson

version = Command("tool", ["version", "--json"]).run_json()

gh = CliClient("gh")
try:
    pr = gh.run_json(["pr", "view", "42", "--json", "title,state"])
except InvalidJson as exc:
    # `run_json()`/`arun_json()` always attach `.stdout` (unlike the streaming
    # `RunningProcess.stdout_json_lines()` case, where it is `None`) — narrow the
    # type before slicing it.
    stdout = exc.stdout or ""
    raise SystemExit(f"{exc.program} did not return JSON: {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.

Stream NDJSON output line by line

An agent/LLM tool or a build tool with a streaming --json mode that emits one object per line — stdout_json_lines() is stdout_lines()'s typed twin: same one-shot setup call, but each item is already the decoded object, no manual json.loads() loop:

from processkit import Command, InvalidJson

proc = await Command("agent-tool", ["--emit", "ndjson"]).astart()
async for event in proc.stdout_json_lines():
    print(event["type"])
finished = await proc.afinish()

A malformed line raises InvalidJson (its message already reports the NDJSON line number and a bounded fragment of that line, plus the real column/byte offset for a genuine JSON syntax error — see Streaming NDJSON output for the rare non-syntax case that has no parser position to report) and the stream continues with the next line rather than ending.

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:

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 stats, no spawn), and raise ProcessNotFound (also a FileNotFoundError) with a searched diagnostic on a miss — the exact error a real run would raise.

Run a tty-sensitive or pipe-buffered tool

Some CLIs block-buffer output when connected to a pipe, suppress interactive features, or refuse to run without a tty. Give the child a managed PTY while keeping process-tree teardown:

from processkit import Command

command = Command("interactive-tool").pty(cols=120, rows=40).keep_stdin_open()
with command.start() as proc:
    stdin = proc.take_stdin()
    proc.resize_pty(160, 50)

Read proc.stdout_lines() for the merged terminal stream. In async code, await stdin.send_control("c") delivers a real terminal Ctrl-C. Do not combine PTY mode with inherited/null/file-redirected stdio; the builder rejects those conflicts before launch.

Keep a service alive (supervision)

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 aborts supervision and is re-raised to the caller; it is never silently interpreted 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:

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):

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:

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

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

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:

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 for the full picture.

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

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:

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:

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:

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.

See what processkit runs (logging)

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

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.

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 for the full set.

Running a command (sync)

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: 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.

Shell pipelines, without the shell

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

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

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

See Pipelines 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:

# 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.

Exceptions

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

# 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.

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:

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 and Resource limits.

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 · Cookbook

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
• Picking a verb
• Program, arguments, working directory
• Local program search
• Environment and sandboxing
• Standard input
• Redirecting stdout and stderr
• Text decoding
• Bounding captured output
• Timeouts
• Privileges and spawn flags
• CPU affinity
• I/O scheduling priority
• Detached launch: the deliberate containment opt-out
• Pseudo-terminal mode
• Results
• Errors
• 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. 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().)

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.

Picking a verb

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, arun_json, aexit_code, aprobe, astart.

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

run_json() is the checked JSON twin of run(): it requires a zero exit and passes stdout through json.loads, returning ordinary Python dictionaries, lists, scalars, booleans, or None. A successful command with malformed JSON raises InvalidJson (a ProcessError) carrying the program and a bounded stdout fragment; a process failure remains the same NonZeroExit/Timeout/Signalled that run() would raise.

metadata = Command("tool", ["metadata", "--json"]).run_json()
metadata = await Command("tool", ["metadata", "--json"]).arun_json()

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.)

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.

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

Overriding argv[0]

arg0(value) overrides the child's argv[0] independently of program — for a multicall binary (BusyBox/Toybox) or a login-shell convention (-bash):

Command("busybox").arg0("ls").args(["-l"]).run()  # busybox dispatches on argv[0]

Program lookup, prefer_local, preflight, spawn diagnostics, and containment all keep using program; only the argument vector delivered to the child changes. Read back what's configured with configured_arg0 (None if unset); a repeat call is last-write-wins.

arg0 is an argv value like any other, so the redaction above covers it too: repr() renders it as arg0: Some("<redacted>"), never the configured string. configured_arg0 and command_line() (busybox [argv0=ls] -l) stay the opt-in ways to read the real value.

Unix only. On a non-Unix platform a run raises Unsupported rather than silently passing the executable name instead — configured_arg0 stays observable there even though a run can never use it.

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.

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:

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:

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 stats); 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():

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:

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:

# 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):

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:

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.

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:

# 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).

Redirecting stdout and stderr

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

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.

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.

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.

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

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.

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 watched output 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).

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()/stderr_lines()/output_events()/lifecycle_events()). The stdout-only iterator resets its window only for stdout lines; the other three use the merged event stream, so either piped stream counts as activity. 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.

Retrying a run

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:

# 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.

Per-process resource limits

rlimit(resource, soft, hard) sets a POSIX setrlimit(2) limit for the child, installed after fork and before exec — before it has run any of its own code:

# Cap the child's open-file-descriptor count and CPU time, and disable core
# dumps for a process that may handle secrets.
(Command("worker").rlimit("no_file", 256, 256).rlimit("cpu", 30, 30).rlimit("core", 0, 0).run())

resource is one of the RlimitResourceName presets: "cpu" (seconds), "core" (bytes), "data" (bytes), "file_size" (bytes), "no_file" (a count), "stack" (bytes) — an unknown name raises ValueError immediately. soft/hard use each resource's native unit; soft must not exceed hard — an invalid pair is a predictable error before the child is ever spawned, never a silent correction. Calls for different resources accumulate; a repeated call for the same resource is last-write-wins. Descendants inherit the values but may lower them further (and raise a lowered soft value back up to hard).

rlimit is POSIX-only: on Windows the run raises Unsupported, the same platform-honest contract as uid / gid / groups / setsid above. It complements the whole-tree caps on ProcessGroup — max_memory=... / max_processes=... / cpu_quota=..., see Sandboxing untrusted tools — with a finer, per-command knob that also works where a cgroup limit isn't available: a non-root cgroup, or macOS/BSD, which have no cgroup equivalent at all.

CPU affinity

cpu_affinity(cpus) pins the child to logical CPU indices; descendants inherit the mask unless they later change their own affinity:

worker = Command("indexer").cpu_affinity([0, 2])

Linux applies the set before exec; Windows applies it while the child is still suspended, before it joins the Job Object and resumes. The sequence must be non-empty and representable by the platform. Duplicates are removed, stored in ascending order, and repeated calls are last-write-wins. macOS and BSD reject a configured affinity at launch with Unsupported rather than silently ignoring it. This is distinct from cpu_quota: affinity chooses which cores may run the tree, while a quota caps total CPU time.

I/O scheduling priority

On Linux, io_priority asks the kernel to lower or raise the child's disk-I/O scheduling class independently of CPU priority:

# Best-effort levels run from 0 (highest) to 7 (lowest).
background = Command("indexer").io_priority("best_effort", level=7)
idle_only = Command("cleanup").io_priority("idle")

"best_effort" and "real_time" require level=0..7; "idle" accepts no level. Real-time I/O can require elevated privilege. The setting is Linux-only: building the command remains portable, but launching it on Windows, macOS, or BSD raises Unsupported rather than silently running with ordinary I/O priority. Last write wins, like CPU priority().

Detached launch: the deliberate containment opt-out

spawn_detached() is the one API that intentionally inverts processkit's no-orphan guarantee. It creates a child outside this library's per-run container and returns a separate DetachedChild carrying only its spawn-time pid:

child = (
    Command("self-updater", ["--apply"])
    .stdout_file("updater.log", append=True)
    .stderr_file("updater.log", append=True)
    .spawn_detached()
)
print(child.pid)

The module-level process_info(child.pid) and process_is_alive(child.pid, saved_start_time) helpers can inspect whether that pid still names the same process instance without taking ownership. Save MemberInfo.start_time when available to reject pid reuse; neither helper adds wait or kill semantics.

Dropping child does not kill or reap the process. There is deliberately no kill, wait, timeout, capture, or interactive-stdin method: after launch, processkit no longer owns it. Stdio is null by default; file redirects are the only supported output destination, because an ownerless pipe can fill and deadlock the child.

Use this only for a daemon, updater, or handoff helper that must outlive its launcher. Prefer start() or a one-shot verb everywhere else. Any setting that needs an owner or output pump — timeouts (including idle_timeout), retries, cancellation, PTY, open stdin, capture callbacks/limits/tees, inherited stdio, or parent-death cleanup — is rejected with Unsupported, never ignored. "Detached" means outside processkit's container, not outside a surrounding CI job, Windows Job Object, cgroup, service, or container imposed by the host. Runnable version: examples/09_spawn_detached.py.

Pseudo-terminal mode

Pipe mode remains the default. Opt into a real pseudo-terminal when a program buffers output behind a pipe, requires a tty, or needs terminal control semantics:

from processkit import Command

command = Command("interactive-tool").pty(cols=120, rows=40).keep_stdin_open()
with command.start() as proc:
    proc.resize_pty(160, 50)

PTY mode is supported on Windows and POSIX. The terminal has one merged output stream: both child stdout and stderr arrive through the existing stdout capture/streaming APIs, while the stderr capture is empty. With keep_stdin_open(), take_stdin() writes to the terminal master and send_control("c") is interpreted by the terminal as Ctrl-C, rather than merely writing byte 0x03 to a pipe.

Provide cols and rows together and use positive values. PTY mode owns the child's stdio, so it cannot be combined with inherit_stdin(), inherited/null stdout or stderr, or stdout_file() / stderr_file(); the conflicting builder call raises before a child can spawn. Use non-interactive flags instead when a PTY is unnecessary — for example ssh -o BatchMode=yes or GIT_TERMINAL_PROMPT=0. Deeper: Streaming & interactive I/O. Runnable version: examples/06_interactive_pty.py.

Results

The capturing verbs hand back a ProcessResult:

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.

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).

# 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:

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.

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.

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:

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

Deeper: Pipelines.

Next: Streaming & interactive I/O · Process groups · Timeouts & cancellation · Supervision · Testing your code · Cookbook · Platform support

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). 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
• Spawning into the group
• Existing processes and containment
• Tearing down
• Signalling the whole tree
• Suspending and resuming
• Inspecting members
• Resource limits: the sandbox
• Stats
• 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):

from processkit import ProcessGroup, host_containment

host = host_containment()  # no group creation or process spawn
print(host.mechanism, host.soft_stop_scope, host.parent_death_cleanup)

with ProcessGroup() as group:
    print(group.mechanism)  # "job_object" | "cgroup_v2" | "process_group" | "unknown"
    print(group.soft_stop_scope)  # "whole_tree" | "opt_in_members" | "none"

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. FreeBSD's ProcessReaper is currently reported as "unknown" because the binding preserves unrecognized variants of the crate's non-exhaustive mechanism enum. See Platform support for the per-OS matrix; the short version is Windows strongest, macOS weakest.

host_containment() predicts the host-level mechanism and maximum graceful stop reach before a group exists, plus abrupt parent-death cleanup and the underlying Rust crate version. A real group's soft_stop_scope is more specific: on Windows it can narrow from host-level "opt_in_members" to "none" when the current membership has no console-CTRL or windowed process. On Unix it is "whole_tree".

Tune the teardown timing at construction:

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.

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:

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

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.

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.

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:

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

Existing processes and containment

A ProcessGroup can establish containment in two ways: processkit can create a root through the group's start() / astart() / runner verbs, or an already running process can be enrolled with adopt_external(pid). The latter is for a process started by subprocess, asyncio.create_subprocess_exec() / asyncio.create_subprocess_shell(), another library, an outside supervisor, or a pidfile.

import subprocess

from processkit import ProcessGroup, Unsupported

external = subprocess.Popen(["my-service"])
try:
    with ProcessGroup() as group:
        try:
            group.adopt_external(external.pid)
        except Unsupported:
            raise RuntimeError("pid-only adoption is unsupported on this platform")
        assert external.pid in group.members()
    # The group's teardown has killed the adopted process; its real parent
    # still owns completion observation and must reap it.
finally:
    if external.poll() is None:
        external.kill()
    external.wait()

pid is an address, not a process handle. During the call, the crate captures its own identity anchor for the process currently named by that number. Later pid reuse is therefore rejected by the group's probes, signals, and teardown. The crate cannot check the earlier race between the caller reading the pid and passing it to adopt_external(), so look the number up as late as possible.

Adoption is containment and teardown only. It never reaps the adopted process, and this API exposes no completion handle or exit status for it. Use members() / members_info() to list it and the group's signal or teardown verbs to control it. The process's actual parent (the caller, an outside supervisor, or init after re-parenting) remains responsible for wait() and the exit status. On the process_group fallback, an adopted process that exits without being reaped can remain a zombie during the configured shutdown grace; only its parent can clear that state.

The containment boundary depends on group.mechanism:

• On Windows Job Objects and Linux cgroup v2, descendants spawned after the adoption inherit the job/cgroup. Descendants that were already spawned keep their original containment.
• On macOS and the Linux process_group fallback, a foreign process normally cannot be regrouped with setpgid, so adoption succeeds with individual tracking. Its future descendants are not included. This is Ok, not a silent failure.
• Linux cgroup-v2 membership is exclusive: adoption moves the process out of its previous cgroup, so that supervisor's limits and teardown no longer apply. Windows may nest a process already in another Job Object, but the kernel can reject the assignment depending on the existing jobs and call order; do not treat one host's result as a universal rule.
• FreeBSD and other BSDs return Unsupported because the crate cannot capture the identity anchor needed for safe pid-only tracking. The process is not tracked by a bare number.

pid=0 and the current process's own pid are rejected as invalid input. A number naming no process, including an already-reaped process, is rejected as a not-found I/O error. Through this binding both cases surface as ProcessError; ProcessNotFound remains reserved for a program that could not be located.

To observe a foreign process without taking ownership, use the module-level process_info() and process_is_alive() lookup helpers documented in Commands.

If adoption is unsupported or several independent launchers must live under one operational umbrella, run the entire supervisor inside a host-managed container, Job Object, or cgroup. That outer boundary belongs to the deployment environment, not to this library.

Tearing down

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

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

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.

Use stop() when teardown telemetry matters:

with ProcessGroup() as group:
    group.start(Command("my-service"))
    report = group.stop(5.0, escalate=True)

print(report.soft_signal, report.attempted_signal)
print(report.members_before, report.members_after)
print(report.drained_within_grace, report.escalated, report.elapsed_seconds)

soft_signal is "sent", "unsupported", "failed", or the forward-compatible "unknown"; attempted_signal names the signal when one was attempted. Unlike shutdown(), stop() does not close the group, so the same object can contain a later child tree. Either member count is None when the platform membership query itself failed. On the POSIX process-group fallback, members_after can temporarily include a killed but not-yet-reaped zombie; atomic Job Object and cgroup membership drop it at exit.

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.

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), 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.

Deeper: keeping a service alive across crashes is Supervision.

Signalling the whole tree

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

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:

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:

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:

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):

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 enforced by the same kernel object that contains the tree:

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"))

update_limits(*, max_memory=None, max_processes=None, cpu_quota=None) changes those caps without recreating the group or restarting its children. It is a full replacement, not a merge: every call describes all three axes, and an omitted axis becomes unbounded. Reissuing the complete desired set is therefore idempotent when the previous update was attempted. update_limits() can return ProcessError("busy") while another operation on the same group is in flight (including an incomplete await group.arun(...)). Wait for that operation to complete, then retry the complete desired set:

with ProcessGroup(max_memory=512 * 1024 * 1024) as group:
    group.start(Command("worker"))
    group.update_limits(
        max_memory=1024 * 1024 * 1024,
        max_processes=64,
        cpu_quota=1.0,
    )
    group.update_limits(max_processes=32)  # memory and CPU are lifted

The method is synchronous; the core update does no asynchronous work. Invalid values and platform failures use the same typed ResourceLimit path as the constructor: the message distinguishes an invalid value, a mechanism without whole-tree accounting, and a capable mechanism that could not enforce the request.

Applying several OS caps is not atomic. A failure does not roll back writes that already succeeded, so the live container may hold a mix of old and new caps; retry the complete desired set or tear the group down. Every axis named by an update that reached the OS is nevertheless added to the sticky cap record, whether the call succeeds or fails. That record remains conservative and never supports a fabricated "not tripped" verdict for a possibly-applied cap.

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, construction or update_limits() raises ResourceLimit rather than handing you a silently-unbounded group:

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; pair limits with a locked-down Command (env_clear().inherit_env(["PATH"]), output_limit(...)) per the Cookbook. For a quick diagnosis of a ResourceLimit failure in those environments, see Troubleshooting.

Stats

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

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
    print(snap.io_read_bytes)  # int | None, cumulative
    print(snap.io_write_bytes)  # int | None, cumulative
    print(snap.peak_process_count)  # int | None, high-water mark

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.

The three additional fields retain the upstream containment mechanism's semantics rather than normalizing different operating systems into one measurement:

The I/O counters are cumulative: a member that has already exited remains in the total. They are not directly comparable between Windows and Linux. Windows counts bytes moved by read/write operations against any target, while Linux io.stat counts bytes that reached the block layer. Linux page-cache hits, pipes, sockets, and tmpfs traffic therefore do not have a Windows-equivalent meaning here; a write may also be accounted after the member that dirtied the page exits. An accounted zero is a real zero, while None means that the mechanism cannot provide that measurement — it is never substituted with 0.

peak_process_count is a kernel high-water mark, not the largest active_process_count observed by calls to stats(). On Linux it is a peak task count, so a multithreaded member contributes all of its threads. It is available only when the cgroup's pids controller is enabled (this binding enables it for a requested max_processes cap) and the kernel exposes pids.peak; otherwise it is None.

These are group counters, not per-run telemetry. RunningProcess.profile() and RunProfile remain unchanged: they describe the process started by one run, whereas group I/O counters and process peaks cannot be divided between multiple runs sharing one containment object.

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

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:

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.

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

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 (environment, output caps), Process groups (whole-tree resource limits), and Timeouts & cancellation (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 recipe
  • 1. Locked-down environment
  • 2. Bounded output
  • 3. Whole-tree resource limits
  • 4. A timeout
  • 5. Teardown
• Checklist: run an untrusted tool safely
• Never detach untrusted code
• 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) — 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.
• 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). 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). 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?) — 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).
• Runaway execution time. A timeout kills the whole tree at a deadline — see Timeouts & cancellation.
• 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. On POSIX you can additionally drop privileges — see 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:

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.

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.) Full treatment: 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.

For a long-lived sandbox, group.update_limits(max_memory=..., max_processes=..., cpu_quota=...) replaces the caps on the live kernel container without restarting its children. The call is a full replacement: an omitted axis is lifted, not retained. A failed multi-axis update is not rolled back and may have applied some axes, so retry the complete desired set; the crate's sticky cap record includes every requested capped axis even on that failure path. If another operation on the group is in flight, update_limits() raises ProcessError("busy"); wait for that operation to complete, then retry the complete desired set.

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.

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.

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.
• (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.
• Read the threat model above — this checklist buys resource and lifetime containment, not syscall/filesystem/network isolation.

Never detach untrusted code

Do not call Command.spawn_detached() for an untrusted tool. Detached launch is an explicit opt-out from processkit's lifetime containment: it has no owning handle, timeout, output bound, or teardown path, and dropping its pid-only DetachedChild does nothing. It exists for a trusted updater or daemon that is supposed to outlive the launcher. In a sandbox, that behavior is precisely the orphan escape this guide is designed to prevent; use a contained one-shot verb, start() context manager, or ProcessGroup instead.

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).

python examples/04_sandbox_resource_limits.py

Next: Process groups · Running commands · Timeouts & cancellation · Cookbook · Platform support

Streaming & interactive I/O

‹ docs index

The one-shot verbs in Running commands — 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
• Streaming stdout or stderr
• Streaming NDJSON output
• Tee output to a file
• Live per-line callbacks
• Interleaved stdout and stderr
• Full lifecycle event stream
• Interactive stdin
• Interactive PTY sessions
• Readiness probes
• Live introspection and per-run telemetry
• Deterministic teardown

Lifecycle

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):

(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.

Streaming stdout or stderr

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:

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 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.

When a service announces readiness on stderr, use stderr_lines() directly:

from processkit import Command, wait_for_line

proc = await Command("my-server").astart()
banner = await wait_for_line(proc.stderr_lines(), "listening", timeout=10)

stderr_lines() drains stdout in the background but yields only decoded stderr lines. It consumes the same one-shot output as stdout_lines(), output_events(), and lifecycle_events(), so choose one of those four for a handle; afterwards use finish()/afinish() or outcome()/aoutcome() to report the run. Because the current core adapter starts from the merged stream, stdout must remain piped.

Deeper: output buffering and capture limits apply to streamed runs too — Running commands.

Streaming NDJSON output

Some tools (agent/LLM CLIs, build tools with a --json streaming mode) emit one JSON object per line as they run. stdout_json_lines() is stdout_lines()'s typed twin: same synchronous setup call, same one-shot-stdout and consuming/streaming-conflict rules, but each item is already the decoded object instead of a raw str:

from processkit import Command, InvalidJson

proc = await Command("agent-tool", ["--emit", "ndjson"]).astart()

async for event in proc.stdout_json_lines():
    print(event["type"], event.get("message"))

finished = await proc.afinish()

No manual json.loads() loop, and a malformed line raises InvalidJson instead of a bare json.JSONDecodeError — the stream continues with the next line rather than ending, matching every other malformed-item case in this library:

stream = proc.stdout_json_lines()
while True:
    try:
        event = await anext(stream)
    except StopAsyncIteration:
        break
    except InvalidJson as exc:
        # str(exc) already reports the NDJSON line number and a bounded
        # fragment of that line — no need to reconstruct it yourself. For a
        # genuine JSON syntax error it also reports the real column/byte
        # offset; for the rare non-syntax decode failure (e.g. a bare integer
        # literal past Python's `sys.set_int_max_str_digits()` limit, which
        # has no parser position at all) it says so honestly instead of
        # inventing one.
        log.warning("skipping malformed line from %s: %s", exc.program, exc)
    else:
        handle(event)

InvalidJson.stdout is None here (unlike Command.run_json() / arun_json()'s bounded whole-payload fragment): a streamed run never buffers the whole payload before parsing, so there is nothing to attach under that name — the per-line diagnostic already lives in str(exc).

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():

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:

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()) — 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).

  An invalid integer count from write() — negative, zero before the buffer is empty, or larger than the remaining buffer — also disables the tee and is reported via sys.unraisablehook, making it visible on stderr even without logging. This report is separate from, and visible alongside, exception-based errors.

• 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().

Raw byte tee

stdout_tee() / stderr_tee() mirror decoded lines. stdout_raw_tee(sink) / stderr_raw_tee(sink) are their undecoded cousins: the raw pipe bytes, before any decoding or line splitting — for a caller that needs a byte-exact copy of exactly what the child wrote (a checksum/digest, a binary log, a protocol that isn't line-oriented):

from processkit import Command

result = Command("some-tool").stdout_raw_tee("out.raw").output()
# out.raw has the exact bytes the child wrote to stdout: non-UTF-8 bytes
# untouched, CRLF and a lone "\r" un-normalized, no fabricated final newline.

Same two sink forms as the decoded tee — a file path or a Python writer — but since the whole point is byte-exact fidelity, a writer here receives each chunk as bytes, so it must be a binary writer (io.BytesIO, a "wb" file), not a text one (sys.stderr, io.StringIO, whose write(bytes) would raise TypeError). It is independent of stdout_tee/stderr_tee/ on_stdout_line — all configured stdout sinks fire from the same pump — and requires that stream to be piped: a no-op under stdout("inherit") / stdout("null") / a stdout_file() redirect (no capture pump runs), and under output_bytes() too (its own return value already is the raw stdout, a separate raw drain with no line pump — reach for the raw tee alongside the line/streaming verbs instead). A write error disables it for the rest of the run, the same isolation as the decoded tee.

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:

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:

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:

  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). 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.

Full lifecycle event stream

Runnable version: examples/07_lifecycle_events.py.

For structured logging that needs the pid and terminal outcome in the same ordered channel as output, use lifecycle_events() instead:

proc = await Command("worker").astart()

async for event in proc.lifecycle_events():
    if event.kind == "started":
        print("started", event.pid)
    elif event.kind in {"stdout", "stderr"}:
        print(event.stream, event.text)
    elif event.kind == "exited":
        assert event.outcome is not None
        print("exit", event.outcome.code)

finished = await proc.afinish()

The sequence begins with started, contains zero or more stdout/stderr events, and ends with exited. Fields that do not apply to a kind are None. LifecycleEvent is an immutable value object: equality and hash() use its kind, pid, text, and outcome, and pickle round-trips preserve those fields for every event kind. The iterator and output_events() are two views over the same one-shot stream, so choose exactly one. Draining either iterator drives the run to completion; the following finish()/afinish() or outcome()/aoutcome() reports the same run. output_events() remains output-only for compatibility.

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():

# 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.

In the default pipe mode this is only a byte; it affects children that read and interpret it. Under Command.pty(), the same writer targets the terminal master, so send_control("c") receives real terminal handling (Ctrl-C / SIGINT on POSIX, and the corresponding ConPTY control input on Windows).

Interactive PTY sessions

Runnable version: examples/06_interactive_pty.py.

Use a PTY for programs that change buffering or interaction when stdout is not a terminal:

from processkit import Command

proc = await Command("interactive-tool").pty(cols=120, rows=40).keep_stdin_open().astart()
stdin = proc.take_stdin()
lines = proc.stdout_lines()  # merged terminal output: stdout plus stderr

await stdin.write_line("status")
print(await anext(lines))
proc.resize_pty(160, 50)
await stdin.send_control("c")
outcome = await proc.aoutcome()

The PTY has one merged terminal stream, exposed as stdout; stderr is empty. Existing line framing still applies, including line_terminator("carriage_return") for progress displays that redraw with bare \r. resize_pty(cols, rows) requires positive dimensions and raises ProcessError for a non-PTY or already-exited process.

Terminal-aware tools often fill that merged stream with ANSI colors, cursor movement, alternate-screen switches, and OSC titles or hyperlinks. Add sanitize_vt() when the consumer needs plain text for logging, parsing, or assertions:

result = Command("colorful-tool").pty().sanitize_vt().output()
assert "\x1b" not in result.stdout

sanitize_vt() targets both capture channels; stdout_sanitize_vt() and stderr_sanitize_vt() target one channel in ordinary pipe mode. The processing order is fixed and identical for stdout, stderr, and PTY's merged stdout: raw bytes are decoded with the configured encoding, decoded text is split using the configured line terminator, then each line is sanitized before entering the capture backlog. ProcessResult, run()/output(), and the streaming stdout_lines()/stderr_lines()/output_events() APIs therefore see clean text without changing line boundaries.

The sanitizer deliberately does not rewrite independent output paths. Per-line callbacks and decoded stdout_tee()/stderr_tee() sinks see the original decoded lines. output_bytes() preserves raw stdout bytes, but stderr remains line-decoded and is therefore sanitized when stderr sanitization is enabled. Direct stdout_file()/stderr_file() redirects preserve original bytes. It is also inert for an inherited or null stream because no capture pump runs. This makes it safe to keep a faithful terminal log in a tee while parsing the cleaned capture.

PTY mode is mutually exclusive with inherited, null, or file-redirected stdio. Conflicts are rejected while constructing the command. It preserves the same private-tree containment and context-manager teardown as an ordinary launch.

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() 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:

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.

Readiness probes

"Start a server, then use it" needs ready, not merely started. Seven free async helpers replace the arbitrary asyncio.sleep, each bounded by its own deadline (an eighth kind — waiting on an un-terminated prompt — is a handle method instead; see Waiting for a prompt below):

from processkit import (
    Command,
    wait_until,
    wait_for_named_pipe,
    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 Windows named-pipe server. A busy pipe is ready too: it proves the
#    server exists even when all pipe instances currently have clients:
await wait_for_named_pipe(r"\\.\pipe\my-server", timeout=10)

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

# 7. 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 seven probes are wait_for_line, wait_for_port, wait_for_http, wait_for_unix_socket, wait_for_named_pipe, wait_for_path, and wait_until. wait_for_named_pipe raises Unsupported outside Windows; wait_for_unix_socket raises it when the Unix connector is unavailable.
• wait_for_http requires the response status token to contain exactly three ASCII digits before applying the default or custom accepted-status predicate. A malformed status line remains a failed attempt even when, for example, expected_status={2000} would accept a loosely parsed integer.
• 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_named_pipe / wait_for_unix_socket set .path. wait_for_port / wait_for_http / wait_for_named_pipe / wait_for_unix_socket also chain the last failed attempt (a connection error, or — for wait_for_http — the last unexpected status or malformed status line) 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_named_pipe / wait_for_unix_socket / wait_until don't touch the process output 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_named_pipe / 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.

Waiting for a prompt (partial output)

Every probe above is line-shaped or endpoint-shaped. An interactive prompt is neither: Password: , (y/N) , a REPL >>> are written without a trailing newline and then blocked on, so they never become a line at all — wait_for_line cannot see them until the stream ends, which for a tool waiting on your answer is never. PTY sessions are made almost entirely of such prompts.

RunningProcess therefore carries its own probe over the live partial tail — the decoded output the pump has not yet split into a line — as the usual sync/async pair (plus a stderr twin for tools that prompt on stderr):

from processkit import Command

proc = await Command("unlock-tool").pty().keep_stdin_open().astart()

# 1. Wait for the un-terminated prompt itself (str = substring of the tail):
await proc.await_for_output("passphrase", timeout=10)

# 2. …answer it over the stdin writer the handle still owns…
stdin = proc.take_stdin()
await stdin.write_line(passphrase)

# 3. …and wait for whatever the tool prints next — a callable predicate here:
prompt = await proc.await_for_output(lambda tail: tail.endswith("$ "), timeout=10)

outcome = await proc.aoutcome()

wait_for_output / await_for_output watch stdout (which is also a PTY's single merged terminal stream); wait_for_stderr_output / await_for_stderr_output watch stderr. Their semantics:

• predicate is a str (substring of the tail) or a callable predicate(tail) -> bool, exactly like wait_for_line, and timeout is keyword-only seconds with the same ValueError on NaN/negative. The matching tail is returned.
• The deadline raises WaitTimeout like every other probe, and a failed probe never kills the child nor arms the run's own timeout() watchdog. If the stream ends before a match, it raises ProcessError immediately instead of waiting out the deadline — the same "stream ended" rule wait_for_line has.
• Non-consuming and repeatable: the tail is only peeked at, so a multi-turn dialog is a sequence of probe → answer turns, and pid / kill() / take_stdin() / context-manager teardown keep working throughout. Answer a prompt before waiting for the next one — a still-standing tail matches again.
• The tail is the whole current partial line, not just the newest fragment: a tool that prints two prompts with no newline between them yields both at once. Match with in / endswith, not equality.
• The tail is raw. Capture redaction and sanitize_vt() both run per completed line, so a terminal's escape sequences are still in there (ConPTY even renders the space in "Password: " as a cursor-move). Match a prompt's plain text — "Password:" — or strip inside a callable, and never assume the fragment is scrubbed.
• stdout and stderr are not symmetrical: the stderr twin raises ProcessError when stderr is not piped, which includes every pty() run (a PTY has one merged stream — use wait_for_output there) and any stderr("null") / stderr("inherit") / stderr_file(...) command.
• Probing installs stdout's one line pump, just like the crate's line probes. So bind stdout_lines() / stdout_json_lines() / output_events() / stderr_lines() / lifecycle_events() before your first probe if you want both — they then coexist, since the tail is a side channel that steals nothing from the iterator — while a stream opened after a probe raises ProcessError. finish() / outcome() / output() still report the run afterwards; output_bytes() does not (raw bytes are gone once stdout is decoded to lines).

Deeper: bounding the whole run (not just the wait) is Timeouts & cancellation.

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:

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
proc.stdout_bytes_seen  # int | None — raw pipe bytes, before decoding/line-splitting
proc.stderr_bytes_seen  # int | None — same, for stderr

stdout_bytes_seen / stderr_bytes_seen are the byte-counter siblings of stdout_line_count / stderr_line_count: monotonic counters of the raw pipe bytes read so far (including bytes an OutputBufferPolicy later discards), stable once the process and its pump have finished. They read 0 — not a sentinel — for a stream that is never pumped (a file redirect, stdout("null"), stdout("inherit")).

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:

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.

Deeper: whole-tree (grandchildren included) resource stats live on Process groups.

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), even if the block raises, without waiting on Python's GC:

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:

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 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.

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.

Next: Process groups · Timeouts & cancellation · Cookbook

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 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.

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):

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:

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 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.

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):

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.

Merging a stage's stderr into the pipe

Command.merge_stderr_in_pipe() is the shell-free equivalent of command 2>&1 | next — set on a non-final stage, it sends that stage's stderr into its own stdout pipe (cloned handles to the same anonymous-pipe writer, so the OS preserves write order), so the downstream stage reads both combined over its stdin:

merged = (
    Command("tool").merge_stderr_in_pipe()  # tool's stderr joins its stdout
    | Command("grep", ["WARN"])
).run()

It is opt-in per stage and a no-op outside a Pipeline or on the final stage — a pipeline only activates it on a stage with a downstream neighbor, so marking the last stage (or a standalone Command) has no effect.

Pipefail diagnostic trade-off. Once a stage's stderr enters the downstream pipe it is no longer available as that stage's own stderr capture: if pipefail attributes the chain's failure to this stage, ProcessResult.stderr is empty for it — the merged bytes may instead surface in the final stage's stdout, having passed through the rest of the pipeline.

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.

# 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:

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:

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 keeps the best-effort stdout and stderr already captured from the last stage before the deadline, using the same buffer policy as a normal capture. A pipeline has no output_limit cap of its own. A per-stage Command.timeout(...) is a separate mechanism: it 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; 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:

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 when you need incremental output, or run the pipeline inside a process group 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 · Process groups · Timeouts & cancellation · Cookbook · Platform support

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
• Idle (inactivity) timeout
• Graceful timeout
• Interrupting a blocked sync call (Ctrl+C)
• Cancelling an awaited async run
• Timeout vs. cancellation
• 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:

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:

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

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

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 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 watched output 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):

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()/stderr_lines()/output_events()/lifecycle_events() (piped stdout, the default). stdout_lines() watches stdout alone; the other three consume the merged event stream, so a line on either piped stream resets their window. 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.

Deeper: Running commands 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.

# 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:

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 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.

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:

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:

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:

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() 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:

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 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:

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.

Bounding pipelines & tuning group shutdown

• A pipeline bounds the whole chain with Pipeline.timeout(seconds). Its capture verbs retain best-effort stdout and stderr already captured by the last stage before the deadline, while its success verbs raise Timeout with that partial output attached. This whole-chain timeout is distinct from a per-stage Command.timeout(...).
• A ProcessGroup'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, ...) (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 — Supervisor(...) with a restart policy and backoff.

Next: Supervision · Streaming & interactive I/O · Async runtimes & event loops · Process groups · Cookbook

Supervision

‹ docs index

A timeout 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, 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
• Restart policies
• Backoff and jitter
• Stopping: the predicate
• Reading the outcome
• Liveness health checks
• Live supervision sessions
• 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:

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. 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), a timeout, or a signal-kill:

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:

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):

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; returning True ends supervision regardless of policy (stopped == "predicate"). The classic "exit 0 is done" under restart="always":

   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 re-raised from the supervisor's terminal verb, so supervision aborts rather than turning the failed check into a false verdict. If you must run a check, precompute it before the supervised run, or use the result handed to the predicate.
• A predicate that raises aborts supervision. The original Python exception is re-raised from run() / arun() or the session's terminal verb; a broken predicate is never silently interpreted as "don't stop".

Reading the outcome

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

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 (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.

Each real-run incarnation normally receives a fresh private ProcessGroup. Pass max_memory=, max_processes=, or cpu_quota= to Supervisor to create those groups with the matching whole-tree cap. A cap the active platform cannot enforce raises ResourceLimit before that incarnation is spawned. Resource caps cannot be combined with runner=: injected runners own their execution semantics, so silently wrapping one in a real process group would break the test-double boundary.

Resource-capped supervision is capture-only. run()/arun() retain their full result and restart semantics, but a start()/astart() session cannot expose a live child handle: status.pid stays None, while status.started_at still reports when the current incarnation began. stop() uses the session cancellation path rather than signalling the current child gracefully. The private limited group is still dropped and the whole incarnation tree is contained; only live pid introspection and graceful child signalling are unavailable.

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:

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.

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=.

Live supervision sessions

run() is ideal when the caller only needs the final outcome. Use start() when you need to observe or stop the keeper loop while it is running:

from processkit import Command, Supervisor

with Supervisor(Command("my-server"), restart="always").start() as session:
    status = session.status
    print(status.is_active, status.pid, status.restarts)
    print(status.started_at, status.is_storm_paused)
    outcome = session.stop(5.0)

status is an atomic snapshot. pid and started_at are None between incarnations, during backoff, and after completion. Capture-only runners, including resource-capped supervisors, report the current incarnation's started_at while keeping pid as None. wait() consumes the session and waits for its natural outcome; stop(grace_seconds) requests graceful termination and reports outcome.stopped == "stopped". Terminal verbs are one-shot.

The async twins are lazy awaitables:

session = await Supervisor(Command("my-server"), restart="always").astart()
async with session:
    print(session.status.pid)
    outcome = await session.astop(5.0)

Use await session.await_wait() to await natural completion. Exiting either context-manager form stops an open session with a one-second grace window; calling a terminal verb inside the block makes the later exit a no-op. Call stop() / astop() explicitly when that grace must be configured.

Sync vs async

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

# 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:

# 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 instead. To exercise restart/stop logic without spawning anything real, see Testing your code, and for the broader task-oriented recipes, the Cookbook.

Next: Timeouts & cancellation · Process groups · Cookbook

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 pytest plugin: ready-made fixtures
• Scripting replies: ScriptedRunner
• Scripted streaming: a live handle, no child
• Record/replay cassettes: RecordReplayRunner
• Asserting on calls: RecordingRunner
• Rendering commands without running: DryRunRunner
• 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:

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

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.

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. The runner fixtures yield the doubles below, so they satisfy the same ProcessRunner seam and spawn no real process. A companion fixture configures cassette redaction:

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 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.

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

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

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 unless a scrubber is configured. Override the processkit_cassette_scrubber fixture for cassettes kept in VCS (see Record/replay cassettes 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:

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.

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).
• 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.

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:

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 (visible on stderr) so a buggy predicate is noisy, not silently wrong.

Deeper: outcome semantics and the exception hierarchy — Running commands.

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:

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.

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.

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:

Only environment values are omitted automatically. program, args, cwd, stdout, and stderr are otherwise stored verbatim and can carry secrets. Use the opt-in scrub= hook for arguments, cwd, and captured output:

import os
from pathlib import Path


def scrub_cassette(field: str, text: str) -> str:
    if field in {"argument", "stdout", "stderr"}:
        return text.replace(os.environ["TOOL_TOKEN"], "<token>")
    if field == "cwd":
        return text.replace(str(Path.home()), "<home>")
    return text


rec = RecordReplayRunner.record("fixtures/tool.json", scrub=scrub_cassette)
# ... run commands, then rec.save()

# The same deterministic callback preserves redacted argument match keys.
rep = RecordReplayRunner.replay("fixtures/tool.json", scrub=scrub_cassette)

Record-mode callers still receive the real unsanitized result; only the stored entry is transformed. Replay returns the fixture-safe stored output. A scrubber exception or non-string result aborts the runner verb and uses a fail-closed placeholder internally, never the raw field. The program name is deliberately not scrubbed because it is the stable tool identity.

For the pytest fixture, override one companion fixture in conftest.py; the plugin applies it in both modes:

import pytest


@pytest.fixture
def processkit_cassette_scrubber():
    return scrub_cassette

save() writes the file owner-only (0600 on Unix) and refuses to follow a symlink, but still review a fixture 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.

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.

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, and to replay real output offline use 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.

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() (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 rather than derailing the run it was only observing.

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; when you also need the structured call record (cwd/env/stdin), use 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:

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:

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:

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:

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"
    # `run_json()`/`arun_json()` always attach `.stdout` (unlike the streaming
    # `RunningProcess.stdout_json_lines()` case, where it is `None`) — narrow the
    # type before indexing/membership-testing it.
    assert exc.stdout is not None
    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 → "Wrap a CLI tool".

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

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).

After pip install processkit-py, the same wrapper is also on PATH as the short processkit command — processkit run -- pytest -x and processkit doctor work exactly like their python -m processkit ... equivalents below, sharing the identical flag set and exit-code contract (both forms delegate to the same entry point). python -m processkit remains fully supported and is what the rest of this page uses throughout — reach for it explicitly when several interpreters are on the machine and the processkit command on PATH might not be the one you mean.

• Basic usage
• Flags
• --profile: machine-readable resource usage
• Exit codes
• supervise
• Resource limits: hard cap or best effort?
• doctor: preflight-diagnose the environment
• What you don't get here

Basic usage

python -m processkit run -- pytest -x
# or, once installed, the shorter console script (identical behavior):
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:

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 by default 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. Output-control flags below deliberately replace that default for the selected streams.

# 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

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. CLI environment layers have a fixed precedence: the inherited base (or --env-clear), then --inherit-env, then --env-file values in file/argument order, then explicit --env flags. A later entry in the same layer wins; explicit flags therefore override every file. Files are UTF-8 (an optional BOM is accepted); values are literal and may contain additional = characters. Missing/unreadable files and non-comment lines without = are usage errors with the file and line number, never tracebacks.

The resource and restart limits are parsed against the widths of their binding types before a ProcessGroup or Supervisor is constructed: --max-memory accepts 1..=2^64-1 (u64), while --max-processes and --max-restarts accept 1..=2^32-1 (u32). Values above those bounds are argparse usage errors; the other positive-integer flags keep their existing contracts.

--sanitize-vt: clean terminal output

Use --sanitize-vt for a color-sensitive tool whose output must become plain text for CI logs or downstream parsing. It is particularly useful with --pty, where stdout and stderr are one escape-heavy terminal stream:

python -m processkit run --pty --sanitize-vt -- colorful-tool status

The flag selects the managed line relay even without --pty: bytes are decoded, split into lines, sanitized through Command.sanitize_vt(), and then re-emitted to the wrapper's stdout/stderr. This preserves stream identity in pipe mode and uses stdout for PTY's merged stream. It is incompatible with --profile (which consumes the live handle through the profiling wait) and --stdout-file (which removes the stdout capture pipe).

--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). --profile exposes exactly that, for a CI step that wants a machine-readable resource accounting of a containerized run without writing any Python:

# 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:

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? 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.

# 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. It is also incompatible with --stdout-file, because streaming requires a piped stdout; redirect stderr instead when stdout activity alone is a useful silence signal.

--output-limit uses that same line pump without adding an idle deadline. Its on_overflow="error" policy counts raw bytes read from both pipes, kills the run when the cap is crossed, and reports the existing internal-failure exit code 125; output beyond the ceiling is never relayed. Direct --stdout-file redirection has no stdout pipe to monitor, so combining it with --output-limit is a usage error. --stderr-file remains compatible: stderr goes directly to the file and the ceiling applies to the still-captured stdout.

--pty also uses the line pump, but the source is a real pseudo-terminal: tools that require isatty() see a terminal and stdout/stderr arrive as one merged stream on the wrapper's stdout. Optional dimensions must be supplied together. PTY owns the child's stdio, so direct file redirects and --profile are usage errors; --output-limit, idle/wall-clock timeouts, resource limits, and --create-no-window keep their normal semantics. This CLI mode is intended for non-interactive TTY-sensitive tools and does not forward the wrapper's stdin; use Command.pty().keep_stdin_open() plus take_stdin() for interactive input. If the platform cannot allocate a PTY, the request fails through the existing Unsupported/internal-error path (125) rather than silently using pipes.

--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:

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.

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:

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.

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

Health checks are synchronous, bounded probes passed to Supervisor's health_check= hook. The first probe runs after one interval, giving the child a startup grace period; three consecutive failures use the binding's default threshold and force-kill the wedged incarnation. A restart policy then treats that kill like any other crash. With --restart never, a POSIX signal kill uses 128 + N; on a platform that reports no signal it uses the existing internal-failure code 120 with a health-check diagnostic.

Resource limits: hard cap or best effort?

The run and supervise forms of --max-memory / --max-processes / --cpu-quota need a real container — a Windows Job Object or a Linux cgroup-v2 root (see Process groups and Platform support). Inside an ordinary container, a systemd user session, or on macOS, the kernel refuses these caps outright.

Rather than fail the operation over a cap the environment can't grant, this CLI degrades: it prints a warning to stderr and runs (or supervises) 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:

python -m processkit doctor

processkit doctor
  graceful-stop scope     : whole_tree
  parent-death cleanup    : whole_tree
  processkit-rs version   : 3.1.0
  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):

processkit doctor
  graceful-stop scope     : opt_in_members
  parent-death cleanup    : direct_child_only
  processkit-rs version   : 3.1.0
  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:

The two entry-point-wide codes from 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:

{
  "mechanism": "cgroup_v2",
  "host_containment": {
    "mechanism": "cgroup_v2",
    "soft_stop_scope": "whole_tree",
    "parent_death_cleanup": "whole_tree",
    "crate_version": "3.1.0"
  },
  "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; all host_containment fields are strings; 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), advanced supervision callbacks such as stop_when and give_up_when (Supervision), line-by-line streaming (Streaming & interactive I/O), 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.

Next: Process groups · Timeouts & cancellation · Cookbook

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 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 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/ suite (pytest-benchmark based) turns that qualitative argument into reproducible numbers, answering the question 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) 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) at 1/10/50-way concurrency, i.e. how batched fan-out scales as concurrency grows.
• test_pty.py — merged PTY output relay over a bounded fixed-width workload. POSIX runs use the native pty path; Windows runs require ConPTY (Windows 10 1809 or newer).
• test_lifecycle_events.py — the complete lifecycle_events() stream, including start, mixed stdout/stderr output, and exit events.
• test_aoutput_as_completed.py — completion-order delivery from a bounded aoutput_as_completed() batch, including per-slot result handling.
• test_supervisor.py — a live Supervisor session that performs a fixed number of crash restarts and then waits for the restart policy to finish.

The PTY benchmark is intentionally platform-aware. The nightly benchmark job runs on Ubuntu, so its PTY result represents the POSIX implementation only; Windows ConPTY measurements are useful on their own host class and are not promised to be directly comparable with that nightly point. Unsupported PTY platforms skip the scenario instead of contributing a misleading measurement.

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:

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 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 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
• asyncio & uvloop
• anyio
• trio
• Why asyncio-native
• The readiness helpers

Support at a glance

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:

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):

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:

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.Futures 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.

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 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, 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) is the same story: pure Python built on asyncio.sleep, needing the same running asyncio loop as everything else here.

Next: Timeouts & cancellation · Streaming & interactive I/O · Platform support · Cookbook

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. host_containment() reports the expected mechanism, host-level soft-stop reach, abrupt parent-death cleanup, and Rust crate version without creating a group or spawning. ProcessGroup.mechanism and ProcessGroup.soft_stop_scope then report the actual mechanism and current per-group soft-stop reach at runtime.

Teardown (the no-orphan guarantee)

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=…))

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.

Signals, suspend/resume, stats

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

Pseudo-terminals (Command.pty())

PTY mode is opt-in and keeps the same containment backend as an ordinary launch. It is incompatible with inherited, null, or file-redirected stdio; conflicts fail while the command is built rather than becoming platform-specific no-ops.

The command-line wrapper exposes the non-interactive form as python -m processkit run --pty [--pty-cols N --pty-rows N] -- PROGRAM ...; its merged terminal stream is relayed on stdout. Interactive input and live resizing remain available through the Python RunningProcess API.

I/O scheduling priority (Command.io_priority())

This is independent of the cross-platform CPU Command.priority() setting.

CPU affinity (Command.cpu_affinity())

Affinity selects allowed logical CPUs; ProcessGroup(cpu_quota=...) separately limits aggregate CPU consumption.

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:

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 with prebuilt wheels covering:

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.

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 and Platform support.

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 and Platform support.

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 for the escape boundary and the stronger backends.

A process started by another library survives group teardown

Use group.adopt_external(process.pid) when the process is already running and you have its pid. Adoption places the process under the group's signal and teardown boundary, but it does not give processkit a child handle: the original parent must still call wait() for completion and exit status. If the platform returns Unsupported (FreeBSD and other BSDs), move creation to a Command / ProcessGroup, or put the whole supervisor under a host-managed containment boundary. See Existing processes and containment.

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.

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.

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 and 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 and Platform support.

Release notes

‹ docs index

All notable changes to processkit are documented in this file.

The format is based on Keep a Changelog, and this project adheres to Semantic Versioning.

Unreleased

Added

• Give LifecycleEvent immutable value semantics based on kind, pid, text, and outcome, including matching hashes and pickle round-trips for every lifecycle variant.

Changed

Fixed

• Include the repository's root commit when auto-generating notes for a first release; later releases remain bounded by the previous release tag.
• Report failures while looking up flush on CLI stdout or stderr as output loss at exit code 119, retaining the original error in emergency diagnostics and preventing interpreter finalization from retrying the failed stream.
• Keep processkit supervise setup best-effort when looking up or invoking reconfigure on a non-standard output stream fails; supervision continues without line buffering instead of aborting.
• Preserve a fresh CancelledError delivered while async completion-order iterators drain cancelled slots: both text and bytes iterators still settle every slot before re-raising it.
• Report non-AttributeError failures while looking up optional flush on Python tee writers through sys.unraisablehook; capture and mirroring continue, while missing or non-callable flush remains optional.
• Keep CLI output-loss termination at exit code 119 when even looking up sys.stderr.write fails; emergency diagnostics remain best-effort.
• Reject HTTP responses whose status token is not exactly three ASCII digits in wait_for_http, even when a custom expected_status would accept a loosely parsed short or long integer.
• Update the bundled ProcessKit-rs core to 3.3.4 so an unconfirmed timeout, cancellation, or pipeline teardown surfaces as ProcessError instead of a misleading terminal outcome; Windows ConPTY rejects unrepresentable sizes and rolls back failed startup, and Windows process-group enumeration errors no longer look like clean completion. This retains the restricted/legacy Linux cgroup thaw fix included since core 3.3.1.

1.5.0 - 2026-08-08

Added

• Add cumulative whole-tree I/O counters (ProcessGroupStats.io_read_bytes and io_write_bytes) and the optional kernel high-water mark ProcessGroupStats.peak_process_count. Availability and units remain mechanism-dependent: Windows Job Objects report transfer bytes, Linux cgroup v2 may report block-layer bytes and task counts, and unsupported measurements are exposed as None rather than a synthetic zero.
• Add ProcessGroup.adopt_external(pid) to bring an already-running external process under group signalling and teardown when only its pid is available. Adoption captures process identity during the call, never reaps the process or exposes its exit status, and documents the Windows, Linux cgroup, POSIX-fallback, and BSD support boundaries.
• Add Command.arg0()/configured_arg0, Command.merge_stderr_in_pipe(), Command.stdout_raw_tee()/stderr_raw_tee(), and RunningProcess.stdout_bytes_seen/stderr_bytes_seen — the last remaining small binding gaps against the processkit core: a Unix-only argv[0] override (raising Unsupported off-Unix), a per-stage 2>&1 |-equivalent pipeline marker, an undecoded byte-exact stdout/stderr tee alongside the existing decoded stdout_tee()/stderr_tee(), and raw pipe byte counters alongside the existing stdout_line_count/stderr_line_count.
• Add partial-tail readiness probes on RunningProcess — wait_for_output()/await_for_output() for stdout and wait_for_stderr_output()/await_for_stderr_output() for stderr — which match an un-terminated prompt (Password: , (y/N) , a REPL >>> ) that the line-oriented probes can never see, so a PTY or CLI dialog can wait for a prompt and answer it over take_stdin(). Non-consuming and repeatable, they raise WaitTimeout on their own deadline without ever killing the child.
• Add Command.sanitize_vt(), stdout_sanitize_vt(), and stderr_sanitize_vt() for clean captured and streaming terminal text, plus processkit run --sanitize-vt for ANSI-free CLI relay output.
• Add a processkit console-script entry point (processkit run -- pytest -x, processkit doctor), alongside the still-supported python -m processkit form, sharing the identical exit-code contract.
• Add ProcessGroup.update_limits(*, max_memory=None, max_processes=None, cpu_quota=None) for full-replacement, synchronous, dynamic adjustment of a live group's resource limits without recreating the group.
• Add Command.rlimit(resource, soft, hard) for POSIX per-process setrlimit(2) limits (RlimitResourceName: "cpu", "core", "data", "file_size", "no_file", "stack"), raising Unsupported off-POSIX.
• Add RunningProcess.stdout_json_lines() for built-in NDJSON line streaming — an async iterator that decodes each stdout line as a standalone JSON value, the streaming counterpart to Command.run_json()/arun_json().
• Add nightly benchmark coverage for PTY relay, lifecycle events, completion-order batches, and live Supervisor restart sessions, with platform-specific PTY handling documented.
• Add wait_for_named_pipe() readiness probing for Windows services, including busy-server detection and symmetric Unsupported behavior elsewhere.
• Add Linux/Windows child CPU affinity through Command.cpu_affinity(...) and the run/supervise CLI --cpu-affinity flag.
• Add reuse-safe process_info() / process_is_alive() helpers for bare pids returned by detached launches, supervision, and process groups.
• Add spawn-free host_containment() capability reports, the current ProcessGroup.soft_stop_scope graceful-stop reach, and the same host details to human-readable and JSON doctor reports.
• Add deterministic cassette secret scrubbing through RecordReplayRunner.record(..., scrub=) / replay(..., scrub=) and the overridable processkit_cassette_scrubber pytest fixture, backed by ProcessKit-rs 3.1.0's symmetric scrub hook.
• Add --pty with optional --pty-cols/--pty-rows to python -m processkit run, exposing a merged pseudo-terminal output stream for tools that require a TTY.
• Add runnable examples for managed PTY sessions, lifecycle events, completion-order batches, intentionally detached helpers, shell-free pipelines, and hermetic runner/cassette testing seams.
• Expand python -m processkit run with fail-loud captured-output limits, direct stdout/stderr file redirects, abrupt-parent-death cleanup, and CPU/I/O priority controls.
• Expand python -m processkit supervise with per-incarnation timeouts and resource caps, headless Windows launches, and proactive TCP or HTTP health checks. Supervisor now accepts the matching max_memory=, max_processes=, and cpu_quota= constructor options.
• Add RunningProcess.stderr_lines() for stderr-only line streaming and direct use with readiness helpers such as wait_for_line.
• Add Command.run_json() / arun_json() with the same typed JSON decoding and InvalidJson diagnostics as CliClient.
• Add a CLI Runner link immediately after the Rust version in the Pages navigation.
• Add opt-in pseudo-terminal launches with Command.pty(...) and live terminal resizing through RunningProcess.resize_pty(...). PTY output is a single merged stdout stream, and interactive stdin uses the existing writer API.
• Add live Supervisor.start() / astart() sessions with status snapshots, graceful stop, completion waits, and sync/async context management.
• Add ProcessGroup.stop() / astop() with a ShutdownReport describing the graceful signal, remaining members, elapsed time, and hard-kill escalation.
• Add RunningProcess.lifecycle_events() for one ordered stream containing the start pid, stdout/stderr lines, and final Outcome, while preserving the output-only output_events() contract.
• Add Linux disk-I/O scheduling controls through Command.io_priority(...); launching a configured command on another platform raises Unsupported.
• Add the deliberately uncontained Command.spawn_detached() escape hatch and pid-only DetachedChild for helpers that must outlive their launcher.
• Add repeatable --env-file PATH support to the run and supervise CLI commands for docker-style KEY=VALUE files with deterministic overrides.
• Add a generated Release notes page to the documentation site, kept in sync with this changelog by local and CI drift checks.

Changed

• Widen InvalidJson.stdout from str to str | None: it is None only when the exception comes from the new RunningProcess.stdout_json_lines() streaming iterator, which cannot buffer the full payload, while run_json()/arun_json() still always populate it. Typed consumers should narrow the type before use; the full diagnostic remains available through str(exc) regardless of the source.
• Bump the bundled ProcessKit-rs core to 3.3.0, preserving the existing Python API and feature set while bringing upstream fixes for merged-stderr pipe teardown, failed PTY-launch cleanup, pipeline pipefail attribution, process-identity-safe metrics, and cassette version validation. The upstream ProcessGroupStats statistics additions are exposed by the binding as documented under Added.
• Bump the bundled ProcessKit-rs core to 3.2.0, preserving the existing Python API and cancellation defaults while bringing upstream compatibility fixes for ConPTY, PTY EOF, readiness, pipelines, environment resolution, and supervision.
• Document and pin idle monitoring per iterator: stdout_lines() watches stdout activity, while merged-event and stderr-only streams count both pipes.
• Exercise Windows ARM64 in regular and nightly test matrices, and add a sharded nightly cargo-mutants signal for the Rust binding layer.

Fixed

• Reject CR/LF, control, and whitespace characters in a wait_for_http host before HTTP serialization, preventing header injection.
• Reject CLI resource and restart limit values above their binding widths with an argparse usage error before constructing a ProcessGroup or Supervisor.
• CLI duration, CPU, backoff, and health-check numeric options now reject non-finite values before constructing a command or supervisor.
• Keep Nightly hardening actionable: its mutation sandbox now includes the changelog required by release-note drift tests, first-run benchmark history can initialize its branch without runner-global git identity, the detached helper example waits for its process to release Windows resources, and the PTY alias/status tests no longer race short-lived incarnations. Resource-capped supervision documentation now also reflects the observable contract: status.pid is unavailable, while status.started_at identifies the current capture-only incarnation.
• Python writer objects used by decoded and raw output tees now retry partial integer write() counts to completion without truncating mirrored output; None and other non-integer return values remain supported and mean the full buffer was accepted, while invalid integer counts are reported via sys.unraisablehook.
• Close the completion hub's socket at the OS level when Python-level cleanup raises, preventing pending anyio-on-asyncio reader tasks and socket-resource warnings after an awaited operation is cancelled during loop shutdown.
• Deliver the output python -m processkit relays itself line by line when its own stdout or stderr is a pipe rather than a terminal: the run modes that re-emit the child's output (--idle-timeout, --output-limit, --pty) and the live tee of supervise. A piped reader — | grep, a log collector, a CI step — previously got those lines in ~8 KiB blocks or in one dump when the run ended, unlike the inherited-stream default; both paths now match the live output docs/cli.md describes.
• Block every direct Command spawn path under pytest's no_real_spawn marker, including JSON, async JSON, and deliberately detached launches.
• Preserve completion-hub rearm errors while still attempting every pending awaiter cancellation when secondary cleanup fails.
• Let python -m processkit supervise run when the parent interpreter has no stdout or stderr stream by omitting the unavailable live-output tee.
• Reject an empty key passed through the CLI's --env flag with the same usage error used for --env-file entries.
• Treat every accepted spelling of piped stdout/stderr consistently when combining Command.stdout() or stderr() with PTY mode.
• Prevent a concurrent lifecycle-event finisher from making a still-reportable RunningProcess appear consumed or letting context-manager teardown become a silent no-op.
• Stop Supervisor immediately on the first failing ScriptedRunner.when predicate, including under an unbounded restart policy.
• Accept bracketed IPv6 literals in wait_for_http and keep scoped IPv6 hosts from being percent-encoded twice between socket and Host header forms.
• Prevent the command-line wrapper's own intermediate output (doctor, idle streaming, diagnostics, and --profile) from producing a traceback when a pipe closes or another output write fails. A vanished receiver stays silent; other write failures use exit code 119.
• Keep async batch result conversion and CliClient.arun_json() parsing on the Python event-loop thread, and make worker-side error conversion safe during interpreter finalization.
• Let Outcome and Finished pickle payloads be restored without entering the Tokio runtime, including from supervisor callbacks and post-fork children.

1.4.2 - 2026-07-26

Added

• New python -m processkit exit code 119, shared by run, supervise, and doctor: the command finished, but the wrapper could not deliver its own buffered output (a final flush that failed with e.g. ENOSPC/EIO, or on a stream closed underneath the process). It is reported instead of the code the run would otherwise have returned — including the child's own — because that code would claim a complete, faithfully relayed run. A receiver that simply went away (BrokenPipeError, e.g. ... | head) is deliberately not this case and stays silent, as before. See "Exit codes" and "How the wrapper terminates" in docs/cli.md.

Changed

• Migrate the Rust core to processkit 3.0.0 (a breaking major release; Cargo.toml now requires 3, resolved to 3.0.1). The Python API is unchanged: OutputEvent, OutputEvents and RunningProcess.output_events() keep their names, signatures and meaning — the core's rename of those types (OutputEvent → ProcessEvent, OutputEvents → ProcessEvents, output_events() → events()) stays an internal detail of the binding, and its Error → Error/ErrorReason split changes nothing about the exception classes or their structured fields. The enabled feature set is unchanged; 3.0's optional new surface (the PTY launch mode, Linux I/O priority, PTY window-size control, the capture-redaction hook, the flat error classifier) is not adopted here.

  Two user-visible consequences, both confined to output_events():

  • The merged event stream became the child's whole lifecycle in the core, so it now also reports process start and exit. Those non-line events are filtered out rather than yielded as an OutputEvent with an empty text — which would be indistinguishable from a real blank line the child printed. async for ev in proc.output_events() therefore yields exactly what it always did: output lines. What the lifecycle events carry is already available: process start is RunningProcess.pid, and the exit is what the finisher you call afterwards returns.
  • The core now delivers that stream's terminal event only when the run is reaped, which means a consumer that drains the stream and then finishes would deadlock. The documented Python order — iterate fully, then await proc.afinish() (or aoutcome()) — is unaffected: the binding drives the run's completion itself once the child is observed to exit, so the iterator ends on its own and the finisher afterwards reports that same run. One deliberate narrowing comes with it — see the BREAKING entry below.

• BREAKING — output() / output_bytes() / profile() (and their a-twins) now raise a ProcessError naming output_events() once that stream has taken the run over, instead of returning the empty captures they used to: it consumed stdout, delivered stderr as events, and (since the 3.0 migration above) completed the run, so there is nothing left for them to capture or to sample. Use finish()/afinish() (outcome + stderr) or outcome()/aoutcome() instead — those report such a run either way. Code that called output() after output_events() and used the result got an empty stdout/stderr with a real outcome; it now has to read that outcome from a finisher.

  The stream takes the run over as soon as it observes the child exit. Iterating to the end always reaches that point, but an early break can too — out of a command that finished while you were reading it. Break out while the child is still running and nothing has been taken over: the old behaviour stands there (empty captures with a real outcome; profile() samples the rest of the run). Which side of that line a given break falls on is a matter of the child's timing, not of how the loop is written, so after streaming events reach for a finisher rather than a capture verb. See "Interleaved stdout and stderr" in docs/streaming.md.

• output_limit(max_bytes=...) under on_overflow="error" — and with it the total_bytes an OutputTooLarge reports — now counts the raw bytes read from the child's output pipe rather than the bytes of the decoded text, following the same change in the Rust core. Line terminators (\n, or both bytes of a CRLF) and bytes that are not valid UTF-8 are charged against the ceiling too, so a cap sized against decoded text raises marginally sooner: by one byte per line for ordinary UTF-8 output, and by more for CRLF or binary-ish output. The drop modes are unaffected — drop_oldest (the default) and drop_newest still bound the retained output by decoded line content, as does Supervisor's capture_max_bytes=, whose capture_on_overflow defaults to drop_oldest. Raw stdout captured by output_bytes() is never decoded, so its cap is unchanged in every mode. No API change; re-check any on_overflow="error" threshold you sized against decoded text. See "What max_bytes actually counts" in docs/commands.md.

• Ctrl+C that interrupts python -m processkit outside run/supervise's own guarded blocks — during startup, argument parsing, or doctor — now reports the documented 128 + SIGINT (130) with the same one-line processkit: interrupted message those paths print, instead of ending through the interpreter's own unhandled-KeyboardInterrupt path. This makes the Ctrl+C contract uniform across the entry point and platforms; for doctor it also keeps an interrupted run distinguishable from its valid 1 verdict ("containment enforced, limits not").

Fixed

• Fix an intermittent SIGSEGV at interpreter exit after a program's final processkit await. The async bridge no longer completes Python Futures from a detached tokio thread through call_soon_threadsafe; tokio stores the outcome and wakes one shared loop.sock_recv dispatcher per event loop, then the event-loop thread converts and resolves it. Repeated stream steps reuse that socket, and the dispatcher closes it after the loop becomes idle rather than leaving a pending receive behind at loop teardown. A short-lived script can now end immediately after any a-prefixed verb without racing Py_FinalizeEx. This also restores ordinary interpreter finalization for python -m processkit — including atexit hooks and finalizers — instead of the temporary os._exit workaround.

1.4.1 - 2026-07-24

Added

• Add CliClient.run_json(call) / arun_json(call): run a wrapped tool like run (requiring a zero exit) and return its stdout parsed as JSON — the run(...) + json.loads(...) + error-mapping boilerplate the many CLIs that emit machine JSON (gh, kubectl, docker, az, jj) otherwise force on every caller. Stdout that does not parse raises a new InvalidJson exception (a ProcessError carrying the client's program and a bounded stdout fragment, with the parser message in str(exc)) instead of a bare, unattributed json.JSONDecodeError; a non-zero exit still raises NonZeroExit as run does. Both verbs go through the same default_env_fn/when-capture pipeline and injectable runner= seam as the other CliClient verbs, so they are hermetically testable with a ScriptedRunner and no real process.
• Add Command.idle_timeout(seconds), an inactivity timeout that tears the child down if it produces no stdout/stderr line for that long — for the "hung tool" case a wall-clock timeout() handles poorly, where a legitimately long job keeps printing progress. It fires as a new, distinct IdleTimeout exception (a ProcessError sibling of Timeout, carrying idle_timeout_seconds), deliberately not the wall-clock timed_out/Timeout signal, so the two timeout classes stay tellable apart and the existing captured timed_out contract is untouched. Enforced on the streaming/interactive surface (start()/astart() + stdout_lines()/output_events()), where the binding drives the per-line output channel; a redirected/inherited stdout is diagnosed by the existing "stdout is not piped" error rather than silently un-watched. Scope note: the one-shot capture verbs (output/run/exit_code/probe and their a-twins), Pipeline, and Supervisor do not enforce it — processkit 2.3.x has no native idle-timeout to observe per-line activity mid-run through those paths, so honoring it there awaits upstream crate support; the setting is carried on the command regardless.
• Add python -m processkit run --idle-timeout SECONDS: kills the child and exits 123 (distinct from --timeout's 124) if it produces no output line for that long. Because idle monitoring rides the per-line channel, this flag pipes and re-emits the child's stdout/stderr (decoded, one line at a time) instead of inheriting them raw, and is incompatible with --profile. The flag is also present on supervise for parity but is a usage error there until upstream Supervisor idle-timeout support lands (its incarnations run through one-shot verbs the idle watchdog cannot observe).

Changed

Fixed

• wait_for_http now forms a correct HTTP/1.1 request line for edge-case host/path values: an IPv6 literal host (e.g. "::1") is bracketed in the Host header per RFC 9112/3986 (Host: [::1]:8080, not the previously ambiguous Host: ::1:8080), and a path containing whitespace, a control character (including CR/LF — previously a header-injection-shaped hazard), or a character outside latin-1 now raises ValueError up front, before any connection is attempted, instead of silently corrupting the request line or raising a raw UnicodeEncodeError.

1.4.0 - 2026-07-23

Added

• Add an actionlint CI gate for semantic GitHub Actions and shell-script checks
• Add the supervise CLI subcommand with restart-policy and backoff flags.
• Add wait_for_http(host, port, path="/", *, timeout, interval=0.05, expected_status=None), a readiness helper that polls an HTTP endpoint (a hand-rolled GET over asyncio streams, no new dependency) and succeeds only on an accepted status code (any 2xx by default; a set/range or a predicate overrides) — a stronger signal than wait_for_port for a server that accepts connections while still warming up
• Add aoutput_as_completed/aoutput_as_completed_bytes, the streaming counterpart to the aoutput_all family: an async iterator that yields each (index, result) pair as its command finishes rather than waiting for the whole batch, with the same hard concurrency cap and no-orphan teardown on cancellation or early exit
• Add python -m processkit run --profile [FILE], emitting a one-line JSON resource profile (duration, CPU time, peak memory, average CPU cores, sample count, exit code/signal, timed-out flag) after the child exits — to stderr if FILE is omitted, or written to FILE otherwise
• Add python -m processkit run --create-no-window, applying Command.create_no_window() to the child so the wrapper does not create a console window on Windows — a no-op outside Windows (same as the underlying binding method)
• Add Command.stdout_file() / stderr_file(), spawn-time direct-redirect sinks that send a stream straight to a file with no parent-side pump or capture in between (append=False, the default, truncates the file on each spawn; append=True appends — e.g. a shared log across Supervisor incarnations or retry() attempts). A file-redirected stdout makes output() / run() / output_bytes() (and their async twins) raise the usual "not piped" ProcessError, but exit_code() / probe() still work since they never touch the stdout pipe; a file-redirected stderr leaves output() working, with result.stderr coming back empty
• Add ProcessGroup.members_info() / MemberInfo, an enriched process-tree snapshot alongside members(): each pid comes with best-effort ppid/exe_name/start_time metadata (None wherever the platform can't report it). exe_name is a short image name, not a path, and start_time is an opaque, platform-specific identity token, not wall-clock — its sole use is pairing with pid across two snapshots to tell a recycled pid apart from the original
• Add Command.windows_graceful_ctrl_break(), an opt-in Windows-only graceful shutdown: at a graceful timeout (timeout_grace) or a group shutdown it sends the direct console child a CTRL_BREAK before the hard TerminateJobObject fallback, giving a child that handles it a chance to exit cleanly first. Console-only (inert under create_no_window / detached) and a harmless no-op outside Windows
• Add opt-in Supervisor liveness health checks via three new keyword-only constructor parameters: health_check (a synchronous () -> bool callable), health_check_interval (required alongside it), and health_check_failures. After health_check_failures consecutive probe failures the supervisor force-restarts the run; each force-restart increments the new SupervisionOutcome.liveness_kills counter, and the final such stop under restart="never" reports SupervisionOutcome.stopped == "unhealthy"
• Add Command.kill_on_parent_death_scope(), a read-only capability query reporting the scope of parent-death cleanup the current platform actually achieves when the owner dies abruptly, as a string: "whole_tree" on Windows (the Job Object reaps the whole tree on owner death), "direct_child_only" on Linux (PR_SET_PDEATHSIG reaches only the direct child; grandchildren survive), or "unsupported" on macOS/BSD (no pdeathsig equivalent). A static query fixed at build time — read it off the class or any instance, with no prior kill_on_parent_death() call — so a caller can state the real reach of the best-effort hardening instead of overpromising a whole-tree guarantee the OS cannot keep

Changed

• Refresh the GitHub Pages landing page from the README: its cover, status badges, no-orphan introduction, runnable example, and capability summary now appear before the guide index.
• Bump the processkit dependency to 2.3.1 (lockfile pinned via cargo update -p processkit --precise 2.3.1; the Cargo.toml requirement stays at the broad 2.3 range). 2.3.1 also added new upstream public surface (Command stdout/stderr file-redirect sinks, windows_graceful_ctrl_break, ProcessGroup::members_info/MemberInfo, Supervisor liveness health checks) that this binding has since adopted — see the Added entries above.
• Bump the processkit dependency to 2.3.2 (lockfile pinned via cargo update -p processkit --precise 2.3.2; the Cargo.toml requirement stays at the broad 2.3 range). 2.3.2 adds new upstream public surface (Command::kill_on_parent_death_scope and the ParentDeathCleanup enum it returns) that this binding adopts — see the Added entry above.

Fixed

• Correct the pipeline documentation to describe per-stage kill-on-drop sub-groups, chain-wide teardown fan-out, and composite timeout attribution, matching the processkit 2.3.x core.
• Fix rendered mdBook links that pointed at the nonexistent README.html, correct the uvloop section anchor, and align contributor/release instructions with the current mdBook-to-GitHub-Pages workflow.
• Make the documented just recipes run on Windows by selecting PowerShell 7 instead of relying on an unavailable sh executable.

1.3.0 - 2026-07-19

Added

• Add sample_stats(group, every), a pure-Python async generator for live ProcessGroup monitoring: a fused, periodic series of ProcessGroupStats snapshots built on top of ProcessGroup.stats()

1.2.4 - 2026-07-12

Added

• Add Open Graph and Twitter Card metadata to the docs site
• Add real Rust crate and .NET documentation links
• Add table border, header fill, and row striping to match the reference site

Changed

• Publish the documentation site to GitHub Pages on push to main
• Pin the GitHub Pages deploy actions to a commit SHA
• Reserve navigation placeholders for the Rust crate and .NET variant
• Restyle the docs site navigation and tables to match the ProcessKit look
• Move the Rust crate and .NET nav entries right after Home, then link them directly to their docs sites
• Match the reference site's pinned nav-group title styling, typography, CSS, and table borders/header fill/row striping/dark-theme colors (navy, not coal) more closely
• Move the implementation switcher above Home, then the Rust/Python/.NET version switcher above Overview, in the docs nav
• Rebuild the docs site with mdBook to match the ProcessKit family
• Render API-reference signatures as text and fix the generator's griffe types
• Give a clear diagnostic for parse_signal with out-of-range ints and floats, and align the property test with the corrected diagnostic
• Convert Supervisor to the frozen + Mutex<Option<...>> pattern
• Unify the named-preset parsers (and their property tests) on case-insensitive matching
• Bump the processkit dependency to 2.2.4

Fixed

• Fix wide-table scrolling specificity and code word-breaking
• Fix doc comment list-bullet misparse and reformat long test line

Removed

• Drop the external crates.io link from the Rust crate placeholder
• Remove stray trailing blank line from mkdocs.yml

1.2.3 - 2026-07-11

Changed

• Bump the processkit dependency to 2.2.3

Fixed

• Fix broken repo-relative README links for PyPI rendering

1.2.2 - 2026-07-10

Changed

• ProcessResult and SupervisionOutcome are no longer picklable — pickling either now raises TypeError (they were advertised as picklable in 1.2.0). Their equality is the underlying processkit crate's own comparison, which also spans a command's configured timeout and accepted success_codes — two fields the crate exposes through no accessor. A pickle could not read them back to reconstruct them, so a result from a command that set .timeout(...) or .success_codes(...) unpickled unequal to its original (identical visible fields and hash(), but !=), silently breaking the round-trip invariant a picklable value type promises. Rather than hand back a subtly-wrong value, both refuse loudly, matching BytesResult/RunProfile. Outcome and Finished remain picklable and round-trip exactly (an Outcome is fully determined by its Python-visible code/signal/ timed_out; a Finished adds only its stderr). To move a captured result across a process boundary — e.g. back from a concurrent.futures.ProcessPoolExecutor worker — pickle result.outcome (an Outcome), or persist result.stdout/.stderr/.code yourself.

Fixed

• CancellationToken's docstring (Rust doc comment and the .pyi stub) no longer claims that a child_token() shares the same cancellation state as its parent and siblings. The actual, already-tested behavior is parent-to-child only: a parent cancels its children, but cancelling a child never affects the parent or its other children.
• processkit.__version__ now matches its own docstring: the first access computes it via importlib.metadata.version() and caches the result (including the source-tree "unknown" outcome) for every later access, instead of re-scanning package metadata on every read. The first access is single-flight even under concurrent readers on a free-threaded build.
• CliClient default_env_fn resolvers are now fail-closed: a resolver that raises or returns a non-str aborts the triggering command()/verb with that exception, before the runner is reached, so no process is spawned. Previously the failure was only reported via the unraisable hook and the resolved value fell back to an empty string — running the command with a silently-missing credential. Applies uniformly to command(), the sync verbs, and the async verbs; a resolver whose key is already set by an explicit per-command env() or a static default_env still never runs (and so cannot abort the call).

1.2.1 - 2026-07-09

Added

• Add Command.prefer_local, exposing crate 2.2's bare-name resolution override
• Add a runnable Command.prefer_local usage example to docs/commands.md
• Add ProcessStdin.send_control for interactive control-byte delivery

Changed

• Broaden the Command.priority docstring privilege caveat to cover above_normal and a niced-parent normal
• Bump the processkit dependency requirement and lockfile to 2.2.0
• Apply rustfmt to the send_control signature
• Bump the processkit dependency to 2.2.1

Fixed

• Fix Windows-incompatible relative path-form assertion in the prefer_local example

1.2.0 - 2026-07-08

Added

• Command.stdout_tee / Command.stderr_tee now accept a Python writer object (anything with a callable write() — io.StringIO, sys.stderr, a text-mode file, a logger wrapper) in addition to a file path, mirroring the child's output straight into your own console/buffer/logger while still capturing it. Each decoded line (plus a "\n") is passed to write() as a str via an async-write bridge: every write is dispatched to the runtime's blocking pool (re-acquiring the GIL there) and awaited on the capture pump, so a slow — even sleeping — write() applies backpressure without blocking the event loop or deadlocking the runtime. The object is discriminated from a path by exposing write (neither str nor pathlib.Path does) and is never closed for you; append=True is meaningful only for a file path and raises ValueError if combined with a writer. A write() exception disables the tee for the rest of the run (a tracing warning under enable_logging(), the same isolation as the file tee) and is additionally reported via sys.unraisablehook — the run and its captured result are unaffected. The previous "a file path only, an arbitrary Python writer is deliberately not supported" restriction is lifted. See docs/streaming.md#tee-output-to-a-file.
• Command.on_stdout_line(callback) / Command.on_stderr_line(callback): a Callable[[str], None] invoked with every decoded 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. Also fires on the async verbs and on a streamed run (start()/astart() + stdout_lines()/output_events()); at most one handler per stream (a repeat call replaces the previous one); a raising callback is reported via sys.unraisablehook rather than propagated or breaking the run. Inert under stdout("inherit")/stdout("null") (resp. stderr(...)) and, for on_stdout_line only, under output_bytes() (which captures stdout raw, bypassing the line pump — stderr still goes through it, so on_stderr_line still fires there). See docs/streaming.md#live-per-line-callbacks.
• A benchmarks/ suite (pytest-benchmark, new bench dependency-group) measuring spawn+capture overhead against subprocess/asyncio.subprocess, ProcessGroup start/exit, line-streaming throughput, and output_all concurrency scaling — dev tooling only, no public API change. Runs nightly via the bench job in nightly-hardening.yml, never in the PR gate; see benchmarks/README.md.
• wait_for_path(path, *, timeout, interval=0.05) — a new async readiness helper alongside wait_until / wait_for_port / wait_for_line, polling until a filesystem path appears (a unix socket, a pid file, or any other marker a daemon creates once ready). Same timeout/interval discipline as its siblings (NaN/negative timeout and non-positive interval raise ValueError; timeout=0 still checks the path at least once) and raises WaitTimeout (also a TimeoutError) on expiry, now carrying a path field (WaitTimeout.__init__ gained a path: StrPath | None = None parameter).
• python -m processkit run -- <cmd> [args...]: a CLI wrapper that runs a command inside a kill-on-exit ProcessGroup with inherited stdio, for shell scripts and CI steps with no Python to write. Supports --timeout, --timeout-grace, --max-memory, --max-processes, and --cpu-quota; the child's own exit code is passed through unchanged, and a timeout / missing program / rejected resource limit is reported as a one-line stderr message with a documented, GNU-timeout-style exit code instead of a traceback. See docs/cli.md.
• Finished gains timed_out and signal properties that delegate to the nested outcome, so it now mirrors Outcome fully — matching code and exited_zero, which were already exposed directly — instead of requiring finished.outcome.timed_out / finished.outcome.signal.
• Command.stdin_file(path) — feed the child's stdin from a file, streamed in chunks by the crate rather than read whole into a Python bytes object, for large inputs (a psql dump, a tar archive, a multi-gigabyte log). Like most other builder methods (stdout_tee/stderr_tee are the deliberate exception), it does not touch the filesystem at build time — the path is opened lazily at spawn, so a missing/unreadable file surfaces as the generic ProcessError from the run/output verb, not FileNotFoundError. Reusable across retries/re-runs, like stdin_bytes/stdin_text; the usual "last stdin method wins" rule applies alongside stdin_bytes()/stdin_text()/ keep_stdin_open().
• ProcessResult and BytesResult gain diagnostic: str | None (stderr if it carries text, otherwise stdout, otherwise None — the same preference order as NonZeroExit/Timeout/Signalled.diagnostic on the exceptions) and outcome: Outcome (the same value RunProfile.outcome and the checking-verb exceptions expose). A result held as data (output()/output_bytes() without ensure_success()) no longer requires re-deriving these by hand. (An output_contains_any convenience was considered alongside these and rejected: the underlying processkit crate has no such method, so it wouldn't be parity with the crate or the exceptions like diagnostic/ outcome are — and it's a one-liner callers can already write themselves via combined, e.g. any(s in result.combined for s in needles).)
• Value semantics for the result types: ProcessResult, BytesResult, Outcome, Finished, RunProfile, and SupervisionOutcome now define __eq__ (comparing every field the underlying processkit crate's own PartialEq compares — not object's previous identity comparison) and a consistent __hash__ (none of their fields are stored floats, so hashing is sound), so two results can now be compared with == and used in a set or as a dict key without a manual field-by-field comparison. ProcessResult/Outcome/Finished/SupervisionOutcome are also picklable — e.g. to return a ProcessResult from a concurrent.futures.ProcessPoolExecutor worker. The underlying crate has no public constructor for any of these types, so unpickling reconstructs one via processkit.testing.ScriptedRunner (an in-memory, no-subprocess replay) — faithful for every field the Python binding exposes, but a command that customized success_codes()/timeout() is not guaranteed to compare == its original after a round trip (those two fields have no Python accessor to reconstruct exactly). BytesResult (raw stdout may not be valid UTF-8, and the only reconstruction channel available is text-only) and RunProfile (reports live OS resource-sampling telemetry with no synthesis path outside an actual monitored run) explicitly do not support pickling and raise a clear TypeError rather than failing silently or fabricating the missing data.

Changed

• CliClient(default_env_fn=...) now validates that every value in the mapping is callable at construction time, raising TypeError (naming the offending key) immediately instead of silently accepting a non-callable value and only discovering the mistake later — once per built command, as an unraisable-hook warning plus an always-empty resolved env var. Valid callables behave exactly as before.

Fixed

• Args (from processkit import Args) no longer rejects the single most common real call site — a variable annotated list[str] (or list[pathlib.Path] / list[os.PathLike[str]]) passed straight through to an argv-like parameter, e.g. args: list[str] = [...]; cmd.args(args). list is invariant, so the original list[StrPath] | tuple[StrPath, ...] spelling only ever accepted a list[StrPath]-annotated variable or a literal, not a list[str]/list[Path]/list[os.PathLike[str]]-annotated one, even though the values are runtime-identical — a static-typing-only false positive with no runtime effect. Args is now a union of the concrete homogeneous list shapes (list[str], list[Path], list[os.PathLike[str]]) instead of the single invariant list[StrPath]; a mixed str/ os.PathLike[str] argv is still accepted, now spelled as a tuple rather than a list literal (e.g. cmd.args((path, "literal"))). A bare str still does not type-check as Args (unchanged; see the Args docstring).

1.1.1 - 2026-07-06

Added

• Command.line_terminator(mode) / Command.stdout_line_terminator(mode) / Command.stderr_line_terminator(mode) — choose where the line pump splits a stream into lines: "newline" (default, splits on \n only, unchanged behavior) or "carriage_return" (also splits on a bare \r, delivering each frame of a curl/pip/apt-style redrawn-in-place progress bar live instead of piling it all up into one line at EOF). line_terminator sets both streams at once; the stdout_/stderr_ variants target one stream, leaving the other's framing untouched. Binds processkit 2.1.0's Command::line_terminator/stdout_line_terminator/stderr_line_terminator (LineTerminator), exposed as the new LineTerminatorName string-preset alias.
• testing.Reply.with_stderr(text) — attach stderr to a scripted reply, including a successful (Reply.ok(...)) one, without resorting to Reply.fail(0, ...) as a workaround.
• processkit.testing.DryRunRunner — a render-only test double that never spawns a process: every verb renders the command to its display-quoted line (via the crate's own Command.command_line() quoting) and returns a synthetic success, the seam behind a tool's own --dry-run/--echo mode. Inspect the rendered lines with commands() / only_command(), or stream them live as each call happens with on_invocation(callback). Works at every runner injection point (output_all and friends, Supervisor, CliClient, runner=), like the other doubles. (Binds processkit 2.1.0's testing::DryRunRunner.)
• Supervisor(..., give_up_when=classifier) — classify a permanent failure so supervision gives up instead of restarting a crash forever, reporting the new SupervisionOutcome.stopped == "gave_up". Bound as a Python callable (like stop_when, not a retry_if-style string preset — the crate's classifier is a per-attempt closure, and a useful verdict is result-specific, not a fixed vocabulary). The callback 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). Consulted only for a crash the policy would otherwise restart, ahead of max_restarts and the failure-storm guard. A crash verdict stops with stopped == "gave_up"; a launch-failure verdict has no result to report and surfaces the classified error directly from run()/arun(). Off by default — a permanent failure restarts as before. The classifier runs on the runtime thread under the GIL; a raising or non-bool callback reads as "not permanent" (keep restarting) and is surfaced via the unraisable hook, never silently swallowed.
• Command.umask(mask) — set the child's POSIX file-mode creation mask; on a non-POSIX platform the run raises Unsupported, matching the existing uid/gid/groups/setsid verbs.
• Command.priority(level) — set the child's CPU-scheduling priority, one of the named presets "idle", "below_normal", "normal", "above_normal", "high" (new Priority type alias). Unix nice/setpriority, Windows priority class — unlike the privilege/POSIX-only verbs above, supported on both platform families, so it never raises Unsupported. Raising to "high" on Unix without CAP_SYS_NICE/root raises PermissionDenied instead of silently applying a lower priority.
• Command.timeout_opt(seconds) — like timeout(), but takes float | None, convenient when a timeout arrives from config as Optional[float]: a value behaves exactly like timeout(seconds), None clears a prior timeout() exactly like no_timeout().
• Command.retry_never() — explicitly opt one command out of retrying, even when it runs through a CliClient configured with a default_retry_if.
• NonZeroExit / Timeout / Signalled now carry a stdout_bytes: bytes | None field — the exact raw stdout bytes when the error came from a checking verb over output_bytes() (e.g. BytesResult.ensure_success()), None on the text path (run() / output()) where stdout is already the complete decoded text. When present, these are the exact pre-decode bytes stdout is a lossy UTF-8 view of (they differ only for non-UTF-8 output). Binds processkit 2.1.0's Error::stdout_bytes().

Changed

• Command.output_limit(max_bytes=...)'s byte ceiling now also bounds the raw stdout of output_bytes() / aoutput_bytes(), matching processkit 2.1.0 — previously a byte cap bounded only the line-pumped stderr and raw stdout was always unbounded. Under on_overflow="error" an over-cap output_bytes() run now raises OutputTooLarge (with max_lines=None — raw bytes have no line count) where it once returned all bytes; under a drop mode its retained bytes are bounded to a head/tail with BytesResult.truncated set. A max_lines cap still never bounds raw stdout. This applies to every inherited output_bytes consumer that runs a Command built with such a policy (CliClient, Pipeline, RunningProcess, ProcessGroup, and the runner= doubles). The Supervisor capture policy is unaffected — it captures line-based output only and has no output_bytes verb.

1.1.0 - 2026-07-06

Breaking

• RunningProcess's consuming verbs now come in a sync/async pair, like everywhere else in this library, instead of being coroutine-only. Migration: await proc.wait() → await proc.aoutcome() (renamed — await is a reserved word, so the async twin of the new sync outcome() couldn't be called await()); await proc.finish() → await proc.afinish(); await proc.output() → await proc.aoutput(); await proc.output_bytes() → await proc.aoutput_bytes(); await proc.profile(...) → await proc.aprofile(...); await proc.shutdown(...) → await proc.ashutdown(...). Each bare name is now a new synchronous method (proc.outcome(), proc.finish(), proc.output(), proc.output_bytes(), proc.profile(...), proc.shutdown(...)), making a handle from the synchronous Command.start() / Runner.start() genuinely usable end-to-end with no event loop at all — not just for the monitor-and-kill() pattern. No aliasing was possible (the old bare names now mean something different — synchronous — so keeping them pointing at the old async behavior would be actively misleading, not merely redundant). RunningProcess.shutdown()/ashutdown() also now match ProcessGroup.shutdown()/ashutdown()'s naming exactly, closing a trap where the same verb name meant "call it" on one class but "await it" on the other.
• ProcessRunner no longer includes start/astart — it is now the capture/check verb surface only (output/run/exit_code/probe and their a-prefixed twins). A new StreamingRunner(ProcessRunner) protocol adds start/astart back for code that also needs a live RunningProcess handle. Migration: annotate an injection point that only calls the capture/check verbs as ProcessRunner (now narrower, easier for a custom double to satisfy); annotate one that also calls start/astart as StreamingRunner. Every built-in runner (Runner, ScriptedRunner, RecordingRunner, RecordReplayRunner) satisfies StreamingRunner (and therefore ProcessRunner too), so existing injected-runner call sites are unaffected — only code that annotated against ProcessRunner expecting start/astart to be part of it needs to switch to StreamingRunner. The internal _runner.py module (never part of the public import path) is renamed _protocols.py to reflect holding two protocols now, not one.
• wait_for() is renamed wait_until() — the old name collided with asyncio.wait_for, which bounds one awaitable, not a polled predicate (different semantics entirely). Migration: await wait_for(...) → await wait_until(...), same arguments. No alias was kept — a wait_for alias sitting next to asyncio.wait_for in the same import line would perpetuate exactly the confusion this rename fixes. All three readiness helpers (wait_until, wait_for_port, wait_for_line) now raise WaitTimeout (ProcessError, TimeoutError) instead of a bare TimeoutError on their own deadline — still catchable as except TimeoutError, but now carrying timeout_seconds (and, for wait_for_port, host/port) as structured fields instead of only a message string.

Added

• A pytest plugin, autoloaded via a pytest11 entry point in every pytest session where processkit is installed (nothing to add to conftest.py; the plugin module is pure Python and import-safe). It exposes the processkit.testing doubles as ready-made fixtures — scripted_runner (a fresh ScriptedRunner), recording_runner (a RecordingRunner spy replying Reply.ok(""), the neutral default), and record_replay_runner (a RecordReplayRunner bound to a per-test cassette) — so injecting a test double is a single fixture parameter. The cassette fixture is replay-by-default with a vcr-style switch to record (--processkit-record CLI flag, then the PROCESSKIT_RECORD env var, then the processkit_record ini option, in that precedence); its file lives under the test's tmp_path unless the processkit_cassette_dir ini option points at a kept directory, and its name is derived deterministically from the test's node id. A @pytest.mark.no_real_spawn marker (registered so it passes --strict-markers) makes any real spawn through Command/Pipeline/Runner/ProcessGroup inside the marked test fail loudly, while injected doubles keep working. Documented in docs/testing.md and the cookbook.
• Args and ReadableBuffer type aliases (from processkit import Args, ReadableBuffer). Args (list[StrPath] | tuple[StrPath, ...]) replaces Sequence[str]/Sequence[StrPath] on every argv-like parameter (Command's args, ScriptedRunner.on()/on_sequence()'s prefix, CliClient.command()/its verbs) — deliberately not Sequence[StrPath], since str is itself structurally a Sequence[str] (each character is a str), so that spelling let a bare string slip through everywhere an argv list was expected (cmd.args("--flag") type-checked, then exploded into one argument per character at runtime). This is a static-typing-only tightening — runtime behavior (and any caller not using mypy) is unaffected; a mypy-strict caller passing something other than a list/ tuple (an arbitrary custom Sequence) at one of these call sites may need to wrap it in list(...). ReadableBuffer (bytes | bytearray | memoryview) replaces the too-narrow bytes on Command.stdin_bytes() / ProcessStdin.write() — both already accepted bytearray/memoryview at runtime (PyO3's buffer-protocol extraction), so this only catches up the stub to reality, no runtime change.
• CliClient's command() and every verb (run/output/output_bytes/ exit_code/probe, a-prefixed twins) now accept a str or any os.PathLike[str] for each argv element, unified with Command's own arg/args typing — previously CliClient was str-only, so a pathlib.Path argument needed a manual str() there but not on Command.
• Documented explicitly: Timeout, ProcessNotFound, and PermissionDenied are transitively OSError subclasses too (since their builtin second base — TimeoutError/FileNotFoundError/PermissionError — has itself been an OSError subclass since Python 3.3), so except OSError catches all three alongside except ProcessError. No behavior change — this was already true; it just wasn't written down anywhere.
• Fixed: PermissionDenied.program is now typed str | None (was str) and reliably reads None — not a missing-attribute AttributeError — on the broader OS-refusal path with no program to name (is_permission_denied() also classifies a program-less Io failure, e.g. a group signal the OS refused, alongside the ordinary spawn-time denial that does name one). Mirrors the class-level default already used for Timeout.timeout_seconds.
• CancellationToken — a portable cancel switch: Command.cancel_on(token) (replaces any prior token — last write wins), Pipeline.cancel_on(token) (gap-fill — a stage with its own explicit token keeps it), and CliClient's default_cancel_on= (also gap-fill) tear the run/chain down when token fires, surfacing the new Cancelled exception. token.cancel() is idempotent; token.child_token() derives a token cancelled automatically with its parent but cancellable independently, for scoping a broader shutdown token down to one operation.
• Cancelled exception — a run deliberately cancelled via a CancellationToken. Previously such a cancellation surfaced only as a plain ProcessError (no dedicated subclass existed since cancel_on had no binding yet); now a distinct, terminal exception — never retried by Command.retry() or restarted by Supervisor, matching the crate's own contract (a cancelled token stays cancelled forever, so a replay could only fail the same way).
• ScriptedRunner.when(predicate, reply) — reply with reply when predicate(command) accepts it, for a match that isn't a plain argv prefix (on()) — e.g. inspecting cwd/arguments/flags via Command's own inspection accessors. predicate is infallible from the crate's perspective, like Supervisor.stop_when: a raising or non-bool predicate reads as "does not match", surfaced via the unraisable hook.
• Reply.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.
• RecordingRunner.new(inner) — wrap any of Runner, ScriptedRunner, RecordReplayRunner, or another RecordingRunner, recording every call made through it. The general form behind the existing replying(reply) (a recorder whose inner runner is always a fresh ScriptedRunner replying with one canned Reply) — new() lets a test combine recording with a double it already built (e.g. a RecordReplayRunner cassette) or with the real Runner.
• ProcessGroup is now itself a runner: group.output(cmd) / .run(cmd) / .exit_code(cmd) / .probe(cmd) / .output_bytes(cmd) (+ a-prefixed twins) run cmd as a shared member of the group (not a standalone private tree) — the same verb surface Runner/ScriptedRunner/… expose, for code written against that seam that should route every spawn through one shared group. (Not registered as a runner= injection target — a ProcessGroup carries real OS resources and is injected directly by callers who already hold one, not through that kwarg seam.)
• output_all() / aoutput_all() / output_all_bytes() / aoutput_all_bytes() now reject concurrency=0 with ValueError instead of silently clamping it to 1 (a confusing "asked for none, got some anyway").
• Command.no_timeout() — run without a timeout, and (unlike simply leaving it unset) opt out of a client-wide CliClient default_timeout gap-fill. Clears a prior .timeout(); the last of the two wins.
• Command.stdout_tee(path, *, append=False) / stderr_tee(path, *, append=False) — tee every decoded line of the stream to a file as it is produced (the line plus a \n, CRLF normalized) while the run also keeps capturing the full output: the one-line way to "stream a log to a file and still get the captured ProcessResult", without a manual loop over stdout_lines(). The sink is a file path (str / os.PathLike[str]); teeing to an arbitrary Python object as a live async writer is deliberately not supported yet (a separate, deferred feature — dispatching each line to a thread, re-acquiring the GIL, honoring backpressure across the FFI boundary is its own scope). The file is opened at build time — the crate takes a concrete sink, not a lazy factory — so an unopenable path (missing parent directory, a directory, a permission denial) raises the matching OSError subclass right at the builder call, not at run; it is created/truncated by default, or appended to with append=True. Inherited crate semantics: a slow sink applies backpressure (it does not block the runtime); a tee write error disables the tee for the rest of the run without breaking the run or its captured result (warned under enable_logging()); and the tee is inert unless the line pump runs — a no-op under stdout("inherit") / stdout("null") and under output_bytes() (raw capture), working with the line verbs (output() / aoutput() / run(), start() + stdout_lines() / output_events()). A reused command's shared sink handle appends across sequential re-runs (retries, Supervisor incarnations) and interleaves across concurrent pipeline stages.
• Command.command_line() — render the command as a single shell-quoted line for display (logs, error messages, a dry-run echo); includes argv, unlike the redacted repr(). Never used to actually execute anything. Plus Command.program / Command.arguments read-only properties (named arguments, not args — that name is already the builder method that appends args).
• Command.unchecked_in_pipe() — exempt a command, as a Pipeline stage, from pipefail attribution (its unclean exit, including a SIGPIPE, is skipped when the chain decides what to report); a no-op outside a Pipeline.
• ProcessResult.ensure_success() / BytesResult.ensure_success() — raise the same exception a checking verb would if the result's exit isn't in success_codes, for turning an already-captured output()/output_bytes() result into an error after the fact. Returns self unchanged on success, so it composes: cmd.output().ensure_success().stdout.
• .diagnostic: str | None on NonZeroExit, Timeout, and Signalled — 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/report something useful without knowing which of the three stream-bearing exceptions it caught.
• Command.timeout_signal() / ProcessGroup.signal() now also accept a raw platform signal number (an int), not just a portable name — the crate's Signal::Other escape hatch (Unix only; a raw number is Unsupported on Windows like every non-Kill signal, same as the named variants).
• CliClient.command(args) — a Command for program <args> with the client's defaults (timeout/env/retry/cancel) pre-applied; chain more builders for a customized one-off call, then pass the result to run() / output() / … (which now accept either a plain arg list or such a Command — the IntoCommand path). An explicit setting on the returned Command always wins over the client's default; only the gaps get filled.
• CliClient's default_env_fn={key: resolver, ...} — a per-key zero-arg resolver called fresh each time a command is built (not each retry attempt) to fill an environment variable, for a credential that should be read freshly rather than baked in once at client-construction time (a static default_env value). An explicit per-call env/default_env at the same key still wins — this only fills the gap.
• Supervisor's capture_max_bytes=/capture_max_lines=/ capture_on_overflow= — bound (or widen) the output captured from each supervised incarnation; the default is already a sensible bounded tail (Command.output_limit's own kwargs, applied here as constructor kwargs instead of a builder method, per the config-struct convention). Setting any of the three requires at least one of the two cap sizes, mirroring output_limit's own validation.
• Command.retry(retry_if, *, max_retries=, initial_backoff=, multiplier=, max_backoff=, jitter=) and CliClient's default_retry_if= (+ default_max_retries=/default_initial_backoff=/default_multiplier=/ default_max_backoff=/default_jitter=) — retry a run with exponential backoff, a cap, and jitter, while retry_if accepts the resulting error. Honored only by the success-checking verbs (run/exit_code/probe, and CliClient's equivalents); ignored by Supervisor (its own RestartPolicy governs keep-alive restarts — a different concern), output_all, and Pipeline. Bound as kwargs over the crate's RetryPolicy, not a mirrored pyclass (the established config-struct convention — see AGENTS.md). retry_if is a named preset over the crate's own error-classification accessors, not an arbitrary Python callable crossing the FFI boundary: "transient" (a bare-retry-clears spawn/IO condition — interrupted, would-block, a busy resource) or "transient_or_timeout" (also retries a .timeout() expiry). CliClient's tuning knobs require default_retry_if= to be set (raises ValueError otherwise) — the same explicit opt-in Command.retry()'s required retry_if already enforces.
• wait_for_line(lines, predicate, *, timeout) is generalized over the iterator's item type (previously hardcoded to AsyncIterator[str]) — it now works over any async iterator (e.g. RunningProcess.output_events()'s OutputEvent items), not just stdout lines, given a callable predicate. predicate also accepts a plain str as a substring-match shorthand (wait_for_line(lines, "listening on", timeout=10)) when the iterator yields str. Purely additive: an existing callable-predicate, str-iterator call site is unaffected.
• Invocation.env_is(name, value) / has_env(name) — the platform-correct (case-insensitive on Windows, last write wins) effective-override check. The existing env dict is plain Python dict semantics, not platform env-key rules: a same-case duplicate key collapses to its last value, but a differently-cased Windows duplicate ("Path"/"PATH") survives as two separate entries — use env_is()/has_env() for the correct answer either way.
• runner= keyword on output_all / aoutput_all / output_all_bytes / aoutput_all_bytes, Supervisor(...), and CliClient(...) — drives the batch/supervision/client through an injected runner (Runner, ScriptedRunner, RecordingRunner, or RecordReplayRunner) instead of the real one, so a test double stands in with no real process spawned. Defaults to the real Runner when omitted (no behavior change). CliClient was previously locked to the real runner; it is now just as testable as raw Command code.
• ScriptedRunner.on_sequence(prefix, replies) — reply with each of replies in turn on successive matching calls (fail a few times, then succeed), then repeat the last reply once exhausted. The declarative form for retry/ supervision test scenarios.
• Prebuilt wheels for Intel macOS (x86_64), cross-compiled from the arm64 (Apple Silicon) runner. Previously Intel Mac users installed from the sdist (needing a Rust toolchain); both macOS architectures are now covered.
• Prebuilt wheels for Windows on ARM (arm64), built natively on GitHub's free-for-public-repos windows-11-arm runner. Both families ship — the abi3 GIL wheel (CPython 3.10+) and the free-threaded cp314t wheel — so ARM64 Windows users (a growing laptop segment) get a binary pip install instead of a from-source build needing a Rust toolchain. No cibuildwheel override was needed: it already provides a native ARM64 CPython 3.10 (for the abi3 wheel) and a native ARM64 cp314t, so the existing build/skip selectors cover win_arm64 unchanged.
• An API reference section on the documentation site — a complete, per-symbol index of the public surface (every class, function, protocol, type alias, and exception, plus the processkit.testing submodule), reachable from the site navigation. It is rendered by mkdocstrings straight from the type stub (_processkit.pyi) and docstrings via griffe's static analysis (no compiled extension needed, so it builds in the extension-free Docs CI), and a drift guard (scripts/gen_api_reference.py --check plus tests/test_api_reference.py) fails if the page ever omits — or invents — a public symbol, so the reference cannot silently diverge from the real API.