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

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

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

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_path,
    wait_for_port,
    wait_for_unix_socket,
    wait_for_http,
    wait_for_line,
)

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

Wait for a unix socket or pid file to appear

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

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

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

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

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

Build a shell-free pipeline

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

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

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

Run many commands at once

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

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.

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; run_json / arun_json run like run (requiring a zero exit) and hand back the already-parsed object, so you skip the run(...) + json.loads(...) + error-mapping boilerplate:

from processkit import CliClient, InvalidJson

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

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

For testable code, pass runner= (a ScriptedRunner and friends from processkit.testing) to drive every verb through a double instead of the real runner — see Testing your code.

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.

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 is reported via the unraisable hook and treated as "don't stop".

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

Sandbox an untrusted tree with resource limits

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

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

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

Program, arguments, working directory

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

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

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 stdout/stderr line for that long — for a tool that hangs silently while a healthy long job keeps printing. It composes with timeout() (whichever threshold is reached first wins) and validates like it (finite, > 0).

from processkit import Command, IdleTimeout

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

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

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.

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

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

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

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

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

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.

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 a property of the group, set once at construction and 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"))

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

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

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

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

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

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

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.

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
• Tee output to a file
• Live per-line callbacks
• Interleaved stdout and stderr
• Interactive stdin
• 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

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.

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

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).
• No-op unless the line pump runs. The tee fires from the line-capture pump, so it is inert under stdout("inherit") / stdout("null") (no pump) and under output_bytes() (raw capture, no line pump). Reach for it with the line verbs — output() / aoutput(), run(), or start() + stdout_lines() / output_events().

Live per-line callbacks

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

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.

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.

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

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

Not the same as inherit_stdin(). keep_stdin_open() + take_stdin() hands you a crate-managed pipe you write to from Python — the crate mediates every byte. inherit_stdin() 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. Six free async helpers replace the arbitrary asyncio.sleep, each bounded by its own deadline:

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

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

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

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

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

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

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

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

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

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

Semantics, deliberately uniform:

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

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

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

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.

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 yields no partial stdout — the chain is run-to-completion or nothing. A per-stage Command.timeout(...) first reaps that stage's whole subtree; the resulting checked Timeout then tears down the remaining stage sub-groups and is attributed to the timed-out stage under pipefail. See Timeouts & cancellation; 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 stdout/stderr line for that long. This is the "hung tool" case a wall-clock timeout handles poorly: a legitimately long job (a build, a test suite, a data export) keeps printing progress, so you can bound its silence tightly without guessing a generous ceiling for its total runtime. The two compose — set both, and whichever threshold is reached first wins.

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

from processkit import Command, IdleTimeout

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

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

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

From the CLI, python -m processkit run --idle-timeout SECONDS applies the same mechanism and exits 123 (distinct from --timeout's 124) on a silent child — see the CLI reference.

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); the same captured-vs-raised rule applies to whichever verb you finish it with.
• 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
• 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 surfaced through the unraisable hook (next bullet), so the supervisor keeps going rather than stopping — i.e. a sync verb in the predicate is a no-op stop gate, not a crash. If you must run a check, precompute it before the supervised run, or use the result handed to the predicate.
• A predicate that raises does not stop supervision. The exception is surfaced through Python's unraisable hook and treated as "don't stop" — a buggy predicate degrades to keep going, it does not crash the supervisor.

Reading the outcome

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

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.

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

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. Each fixture yields one of the doubles below, so it satisfies the same ProcessRunner seam and spawns no real process:

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 — review one before committing it (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 env values are redacted. program, args, cwd, stdout, and stderr are stored verbatim and can carry secrets — a --password=… flag, a token echoed to output — so review a fixture before committing it, and keep secret-bearing cassettes out of shared trees. (save() writes the file owner-only — 0600 on Unix — and refuses to follow a symlink, so a fresh cassette isn't world-readable; the review is still on you before committing it.)

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

Deeper: how a ProcessResult is shaped before it's captured — the Cookbook.

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

• 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

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

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

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

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

Resource limits: hard cap or best effort?

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

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

doctor: preflight-diagnose the environment

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

python -m processkit doctor

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

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

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

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

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

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

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

What you don't get here

This is a v1, deliberately minimal wrapper — reach for the Python API directly for anything beyond it: piping several commands together (Pipelines), 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. There is also no --output-limit flag: stdio here is always inherited straight through to your terminal, so there is no captured-output buffer for Command.output_limit(...) to bound in the first place — that method only matters when a caller captures output via the Python API instead.

Next: Process groups · 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.

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. ProcessGroup.mechanism tells you which mechanism is active at runtime: "job_object", "cgroup_v2", or "process_group".

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.

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

API reference

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

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

Building & running commands

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

Command

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

A command builder. Builder methods return a new Command.

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

arg

def arg(arg: StrPath) -> Command

args

def args(args: Args) -> Command

cwd

def cwd(path: StrPath) -> Command

prefer_local

def prefer_local(dir: StrPath) -> Command

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

env

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

envs

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

env_remove

def env_remove(key: str) -> Command

env_clear

def env_clear() -> Command

inherit_env

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

stdin_bytes

def stdin_bytes(data: ReadableBuffer) -> Command

stdin_text

def stdin_text(text: str) -> Command

stdin_file

def stdin_file(path: StrPath) -> Command

keep_stdin_open

def keep_stdin_open() -> Command

inherit_stdin

def inherit_stdin() -> Command

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

timeout

def timeout(seconds: float) -> Command

idle_timeout

def idle_timeout(seconds: float) -> Command

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

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

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

timeout_grace

def timeout_grace(seconds: float) -> Command

timeout_signal

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

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

no_timeout

def no_timeout() -> Command

timeout_opt

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

cancel_on

def cancel_on(token: CancellationToken) -> Command

success_codes

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

retry

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

retry_never

def retry_never() -> Command

stdout

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

stderr

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

encoding

def encoding(label: str) -> Command

stdout_encoding

def stdout_encoding(label: str) -> Command

stderr_encoding

def stderr_encoding(label: str) -> Command

line_terminator

def line_terminator(mode: LineTerminatorName) -> Command

Choose where the line pump splits both streams into lines. "newline" (the default) splits on \n only; "carriage_return" also splits on a bare \r (one not immediately followed by \n), delivered live — for curl/pip/apt-style \r-redrawn progress output that would otherwise pile up into a single line until EOF. A \r\n pair still counts as one terminator. Shared by stdout_lines()/output_events(), the per-line handlers (on_stdout_line/on_stderr_line), stdout_tee/stderr_tee, and output_string alike; set both streams here or independently with stdout_line_terminator/stderr_line_terminator. Unknown preset raises ValueError.

stdout_line_terminator

def stdout_line_terminator(mode: LineTerminatorName) -> Command

Choose where the line pump splits stdout into lines (see line_terminator); stderr framing is left untouched.

stderr_line_terminator

def stderr_line_terminator(mode: LineTerminatorName) -> Command

Choose where the line pump splits stderr into lines (see line_terminator); stdout framing is left untouched. Handy when progress output lands on stderr while stdout stays newline-structured.

stdout_tee

def stdout_tee(sink: StrPath | SupportsWrite, *, append: bool = ...) -> Command

Tee every decoded stdout line (line + \n) to sink as it is produced, while the run also keeps capturing the full output (the sink does not steal from ProcessResult.stdout).

sink is either a file path (str / os.PathLike[str]) or a Python writer — any object with a callable write() (an io.StringIO, sys.stderr, a text-mode file, a logger wrapper), picked apart by whether it exposes write (neither str nor pathlib.Path does).

• File path: teed as raw UTF-8 bytes, opened at build time (not at run) — created if absent and truncated, or append mode when append=True; an unopenable path raises the matching OSError subclass right here.
• Writer: each decoded line (then "\n") is passed to write() as a str (a text sink — a binary writer whose write(str) raises TypeError is the wrong object here). Every write is dispatched to a blocking thread and awaited on the pump, so a slow write() applies backpressure without blocking the event loop; the object is not closed for you. append is meaningless for a writer — passing append=True with one raises ValueError.

A write error disables the tee for the rest of the run (a tracing warning under enable_logging()) while the run and its captured result continue unaffected; a writer's write() exception is additionally reported via sys.unraisablehook. Inert unless stdout is piped through the line pump — a no-op under stdout("inherit") / stdout("null") and under output_bytes() (raw capture).

stderr_tee

def stderr_tee(sink: StrPath | SupportsWrite, *, append: bool = ...) -> Command

Tee every decoded stderr line to sink. Same contract as stdout_tee — a file path (opened at build time, truncate by default or append) or a Python writer object with a callable write() (fed each decoded line as a str via the same blocking-pool async-write bridge, never closed for you), coexisting with capture, inert unless stderr is piped through the line pump.

stdout_file

def stdout_file(path: StrPath, *, append: bool = ...) -> Command

Redirect the child's stdout straight to a file, opened at spawn time — the child writes to the file's own descriptor, with no parent-side pump, tee, or capture buffer. The direct-redirect cousin of stdout_tee (which instead also captures and mirrors each line): a > / >> shell redirect, minus the shell.

The binding folds the crate's three spellings (stdout_file / stdout_file_append / stdout_file_truncate) into one append= kwarg, mirroring the sibling stdout_tee(sink, *, append=False) rather than a 1:1 copy of the core's convenience aliases. append=False (the default) creates or truncates the file on every spawn; append=True creates or appends — the mode for a shared log across Supervisor incarnations / retry() attempts (each re-run appends to the one file with no separator).

Opened at spawn, not now (unlike stdout_tee). Only the path is stored; the file is opened when the command launches, so a not-yet- existing path is not a build-time error and each re-run / retry reopens it. An unopenable path (a missing parent directory, a permission denial) surfaces from the run verb at launch, not from this call.

No capture — use a non-capturing verb. With stdout on the file there is no pipe to read, so the capture/streaming verbs (output() / run() / output_bytes() / their a-twins, and start() + stdout_lines() / output_events()) raise ProcessError ("stdout is not piped … so the capture verbs have nothing to read") instead of returning empty output — drive it with exit_code() / probe(). A later stdout("pipe"/"inherit"/"null") clears the redirect and restores the ordinary stdio mode.

stderr_file

def stderr_file(path: StrPath, *, append: bool = ...) -> Command

Redirect the child's stderr straight to a file, opened at spawn time. Same contract as stdout_file — a child-owned descriptor with no parent-side pump/tee/buffer, the path opened lazily at launch (a missing path is not a build-time error), append=False truncating on every spawn while append=True appends (the shared-Supervisor-log mode), and a later stderr("pipe"/"inherit"/"null") clearing the redirect.

Unlike stdout_file, this does not disable the capture verbs: only a non-piped stdout gates them, so output() / run() keep working and return the child's stdout while stderr is diverted to the file and result.stderr comes back empty.

on_stdout_line

def on_stdout_line(callback: Callable[[str], None]) -> Command

Call callback with every decoded stdout line as it is produced — the way to give the synchronous surface (.output()/.run()) live progress observation during an otherwise-blocking call, without losing the full capture: callback observes the same decoded lines that land in ProcessResult.stdout, it does not replace them. Also fires on the async verbs and on a streamed run (start()/ astart() + stdout_lines()/output_events()) — one callback, every path.

callback is infallible: an exception raised inside it is reported via sys.unraisablehook rather than propagated — it never derails the run or alters the captured result. At most one handler per stream — a repeat call replaces the previous one (builder semantics, like timeout()); compose inside one callable to fan out.

Inert under stdout("inherit")/stdout("null") (no pump runs) and under output_bytes() (stdout is captured raw there, bypassing the line pump).

on_stderr_line

def on_stderr_line(callback: Callable[[str], None]) -> Command

Call callback with every decoded stderr line as it is produced. Same contract as on_stdout_line — full capture unaffected, fires on sync/async/streamed paths alike, infallible (a raising callback goes to sys.unraisablehook, never propagates), at most one handler per stream.

Inert under stderr("inherit")/stderr("null"). Unlike on_stdout_line, not silenced by output_bytes(): that verb only bypasses the stdout line pump for its raw-bytes capture — stderr still decodes through the line pump exactly as under output(), so this callback still fires.

kill_on_parent_death

def kill_on_parent_death() -> Command

Add best-effort hardening for abrupt owner death, beyond ordinary kill-on-drop teardown. Windows already reaps the whole Job Object tree; Linux arms PR_SET_PDEATHSIG for the direct child only (grandchildren survive); macOS/BSD have no equivalent and treat this as a no-op. Use kill_on_parent_death_scope() to report the platform's actual reach.

kill_on_parent_death_scope

def kill_on_parent_death_scope() -> str

The scope of parent-death cleanup this platform actually achieves when the owner dies abruptly (a SIGKILL or crash, where graceful Drop teardown never runs), as a stable string:

• "whole_tree" — Windows: the kernel closes the Job Object handle on owner death and kill-on-close reaps the direct child and every descendant.
• "direct_child_only" — Linux: PR_SET_PDEATHSIG reaches only the direct child; grandchildren survive the owner's abrupt death.
• "unsupported" — macOS/BSD: no pdeathsig equivalent, so an abrupt owner death triggers no cleanup at all.

An honest capability report, not a request — read it to state the real reach of kill_on_parent_death() (best-effort on Unix) instead of overpromising a whole-tree guarantee the OS cannot keep. It covers only the abrupt-death path: ordinary graceful teardown still kills the whole tree on every platform regardless.

A staticmethod — the scope is fixed per target at build time and does not depend on instance state or on whether kill_on_parent_death() was called, so call it on the class (Command.kill_on_parent_death_scope()) or on any instance for the same answer.

create_no_window

def create_no_window() -> Command

windows_graceful_ctrl_break

def windows_graceful_ctrl_break() -> Command

Windows: opt in to a graceful teardown — at a graceful timeout (timeout_grace) or a ProcessGroup shutdown, send the direct child a console CTRL_BREAK before the grace window, so a console child (a CLI, Node, Python, or Go service that installs a CTRL_BREAK handler) can flush and exit cleanly ahead of the hard TerminateJobObject fallback. Without it Windows has no soft-signal tier and a graceful timeout collapses straight to an atomic Job Object kill; any survivor past the grace is still hard-killed.

Boundaries. Console-only — a child spawned create_no_window() (or otherwise detached) shares no console, never receives the event, and rides the grace to the hard kill; a GUI/service parent with no console of its own can't deliver it either. It is CTRL_BREAK, not CTRL_C, and timeout_signal does not apply. Only the direct child is addressed — its own descendants get it via the shared console/group, but an adopted process does not. A harmless no-op outside Windows (Unix's graceful tier already sends a real signal) — unlike the POSIX-only uid/gid/groups/setsid/umask, which raise Unsupported off-platform rather than silently no-op'ing.

uid

def uid(uid: int) -> Command

gid

def gid(gid: int) -> Command

groups

def groups(gids: Sequence[int]) -> Command

setsid

def setsid() -> Command

umask

def umask(mask: int) -> Command

priority

def priority(level: Priority) -> Command

output_limit

def output_limit(
    *,
    max_bytes: int | None = ...,
    max_lines: int | None = ...,
    on_overflow: Literal['drop_oldest', 'drop_newest', 'error'] = ...,
) -> Command

output

def output() -> ProcessResult

output_bytes

def output_bytes() -> BytesResult

run

def run() -> str

exit_code

def exit_code() -> int

probe

def probe() -> bool

resolve_program

def resolve_program() -> str

Resolve this command's program to a concrete executable path without launching it — a spawn-free, side-effect-free preflight ("is this tool installed?"). Reuses the same PATH/PATHEXT/execute-bit lookup a real run performs — a bare name against this command's prefer_local() directories (priority order) then the effective PATH, a path-form program directly — honoring a relocated child PATH (env()/env_remove()/env_clear()/inherit_env()), so the result is exactly what a spawn of this same command would find. Returns the resolved absolute path; raises ProcessNotFound (also a FileNotFoundError, with a searched diagnostic) on a miss. No a-prefixed async twin — the probe is synchronous and needs no runtime.

aoutput

def aoutput() -> Awaitable[ProcessResult]

aoutput_bytes

def aoutput_bytes() -> Awaitable[BytesResult]

arun

def arun() -> Awaitable[str]

aexit_code

def aexit_code() -> Awaitable[int]

aprobe

def aprobe() -> Awaitable[bool]

start

def start() -> RunningProcess

astart

def astart() -> Awaitable[RunningProcess]

unchecked_in_pipe

def unchecked_in_pipe() -> Command

program

program: str

arguments

arguments: list[str]

command_line

def command_line() -> str

pipe

def pipe(other: Command) -> Pipeline

CliClient

CliClient(
    program: StrPath,
    *,
    default_timeout: float | None = ...,
    default_env: Mapping[str, str] | None = ...,
    default_env_remove: Sequence[str] | None = ...,
    default_env_fn: Mapping[str, Callable[[], str]] | None = ...,
    default_retry_if: RetryIf | None = ...,
    default_max_retries: int | None = ...,
    default_initial_backoff: float | None = ...,
    default_multiplier: float | None = ...,
    default_max_backoff: float | None = ...,
    default_jitter: bool | None = ...,
    default_cancel_on: CancellationToken | None = ...,
    runner: RunnerLike | None = ...,
)

A program bound to default timeout/env/retry, run with the real Runner by default or an injected runner= (a ScriptedRunner and friends, for testable code with no real spawns). The verbs take just the per-call arguments.

command

def command(args: Args) -> Command

A Command for program <args>, the client's defaults pre-applied — chain more builders, then pass it to a verb below instead of a plain arg list. An explicit setting on it always wins over the default.

run

def run(call: Args | Command) -> str

run_json

def run_json(call: Args | Command) -> Any

Run like run (require a zero exit) and parse the stdout as JSON, returning the decoded object (a dict/list/str/number/ bool/None) — the run(...) + json.loads(...) + error-attribution boilerplate the many CLIs that emit machine JSON otherwise force on every caller. A non-zero exit raises NonZeroExit (like run); stdout that does not parse raises InvalidJson (a ProcessError carrying the client's program and a bounded stdout fragment), never a bare json.JSONDecodeError. Returns Any: JSON admits any of those shapes, so narrow the result yourself.

output

def output(call: Args | Command) -> ProcessResult

output_bytes

def output_bytes(call: Args | Command) -> BytesResult

exit_code

def exit_code(call: Args | Command) -> int

probe

def probe(call: Args | Command) -> bool

resolve_program

def resolve_program() -> str

Resolve this client's program to a concrete executable path without spawning it — the client-level preflight ("is this tool installed?"), with no side effects. Applies the client's defaults (so a default_env/default_env_fn that relocates PATH is honored as at launch), then resolves via the same PATH/PATHEXT/execute-bit logic a real run uses. Returns the resolved absolute path; a default_env_fn that raises or returns a non-str aborts it fail-closed (like the run verbs), and a miss raises ProcessNotFound (also a FileNotFoundError, with a searched diagnostic). No a-prefixed async twin — the probe is synchronous.

arun

def arun(call: Args | Command) -> Awaitable[str]

arun_json

def arun_json(call: Args | Command) -> Awaitable[Any]

Async counterpart of run_json() — await it for the decoded JSON object. Same contract: a non-zero exit propagates NonZeroExit, stdout that does not parse propagates InvalidJson (never a bare json.JSONDecodeError), both out of the await.

aoutput

def aoutput(call: Args | Command) -> Awaitable[ProcessResult]

aoutput_bytes

def aoutput_bytes(call: Args | Command) -> Awaitable[BytesResult]

aexit_code

def aexit_code(call: Args | Command) -> Awaitable[int]

aprobe

def aprobe(call: Args | Command) -> Awaitable[bool]

Pipeline

class Pipeline

A shell-free pipeline a | b | c. Each stage runs in its own kill-on-drop sub-group; checked failure, chain timeout, or cancellation fans teardown across every sub-group, while outcome attribution follows pipefail semantics.

By design, no start/astart — see Command.pipe()'s stub/binding comment: a pipeline is a whole-chain verb, with no natural "handle to a live chain" to hand back. Stream an individual stage by start()ing that one Command directly instead.

pipe

def pipe(other: Command) -> Pipeline

timeout

def timeout(seconds: float) -> Pipeline

cancel_on

def cancel_on(token: CancellationToken) -> Pipeline

output

def output() -> ProcessResult

output_bytes

def output_bytes() -> BytesResult

run

def run() -> str

exit_code

def exit_code() -> int

probe

def probe() -> bool

aoutput

def aoutput() -> Awaitable[ProcessResult]

aoutput_bytes

def aoutput_bytes() -> Awaitable[BytesResult]

arun

def arun() -> Awaitable[str]

aexit_code

def aexit_code() -> Awaitable[int]

aprobe

def aprobe() -> Awaitable[bool]

RunningProcess

class RunningProcess

A handle to a started process: stream output, write stdin, wait for exit.

Usable as a (async) context manager — exiting the block tears the process down (a hard kill of the whole private tree for a standalone start()/astart() handle). stdout_lines() / output_events() / take_stdin() / kill() are synchronous setup calls; the iterator / handle they return is what you await.

Every consuming verb — outcome/finish/output/output_bytes/ profile/shutdown — comes in a sync/async pair, like everywhere else in this library: the bare name blocks the calling thread (via the same interruptible driver as Command.output()), the a-prefixed twin is a coroutine (outcome/aoutcome rather than the unusable wait/ await, since await is a reserved word). Either member of a pair consumes the handle — afterwards it is spent (pid and the other getters return None, and every consuming verb raises). Use whichever matches your calling code, regardless of whether the handle came from start() or astart().

pid

pid: int | None

elapsed_seconds

elapsed_seconds: float | None

cpu_time_seconds

cpu_time_seconds: float | None

peak_memory_bytes

peak_memory_bytes: int | None

stdout_line_count

stdout_line_count: int | None

stderr_line_count

stderr_line_count: int | None

owns_group

owns_group: bool | None

stdout_lines

def stdout_lines() -> StdoutLines

output_events

def output_events() -> OutputEvents

Interleaved stdout+stderr lines as an async iterator (call once).

Consumes both pipes, so pick this or stdout_lines(). Report the run afterwards with finish()/afinish() (outcome + stderr) or outcome()/aoutcome() — they report it whether you iterate to the end or break out early. output()/output_bytes()/profile() raise once this stream has taken the run over, which it does as soon as it observes the child exit: its stdout was streamed away and its stderr was delivered as events, so they have nothing left to capture or sample. Teardown is unaffected — leaving the handle's context-manager block (or dropping it) hard-kills the whole tree even after the stream has taken the run over, which is what stops a grandchild still holding the pipe from outliving the block.

take_stdin

def take_stdin() -> ProcessStdin

The writable stdin handle. Raises ProcessError if stdin was not kept open (build the Command with keep_stdin_open()) or was already taken — so a missing setup fails here, not with a later AttributeError.

kill

def kill() -> None

Begin tearing the tree down without waiting (like subprocess.Popen.kill(): fire-and-forget).

outcome

def outcome() -> Outcome

aoutcome

def aoutcome() -> Awaitable[Outcome]

finish

def finish() -> Finished

afinish

def afinish() -> Awaitable[Finished]

output

def output() -> ProcessResult

Wait for exit and capture the full ProcessResult; consumes the handle.

Raises ProcessError once an output_events() stream has taken this run over — which that stream does as soon as it observes the child exit, whether or not you iterated to the end: it consumed stdout, delivered stderr as events and completed the run, so there is nothing left to capture. Read such a run with finish()/afinish() or outcome()/aoutcome() instead. Stopping the iteration while the child is still running leaves the run with this handle, and this still does what it did before the processkit 3.0 migration: returns empty captures with a real outcome. Which of the two a given break gets depends on the child's timing, so after streaming events prefer a finisher.

aoutput

def aoutput() -> Awaitable[ProcessResult]

Async counterpart of output — same output_events() restriction.

output_bytes

def output_bytes() -> BytesResult

Wait for exit and capture raw stdout as a BytesResult; consumes the handle. Raises ProcessError once an output_events() stream has taken the run over, under the same condition and for the same reason as output.

aoutput_bytes

def aoutput_bytes() -> Awaitable[BytesResult]

Async counterpart of output_bytes — same output_events() restriction.

profile

def profile(every_seconds: float) -> RunProfile

Wait for exit while sampling resource usage every every_seconds, returning a RunProfile; consumes the handle.

Raises ProcessError once an output_events() stream has taken this run over — which that stream does as soon as it observes the child exit, whether or not you iterated to the end: the run is over, so there is no live run left to sample. Use finish()/afinish() or outcome()/ aoutcome() for its outcome. Stopping the iteration while the child is still running leaves the run with this handle, and this still profiles the rest of it.

aprofile

def aprofile(every_seconds: float) -> Awaitable[RunProfile]

Async counterpart of profile — same output_events() restriction.

shutdown

def shutdown(grace_seconds: float) -> Outcome

Graceful teardown (signal -> wait grace_seconds -> hard kill), returning the Outcome; consumes the handle. Only for a standalone start()/astart() handle — a handle from ProcessGroup.start() raises Unsupported; tear such a child down via the group (or kill()). Named to match ProcessGroup.shutdown()/ashutdown().

After an output_events() stream has taken the run over the child has already exited, so there is nothing to signal: this reports that run's real outcome, waiting for its output to finish draining just as finish() does, rather than escalating against surviving grandchildren. Leave the handle's context-manager block instead when a hard bound matters more than the outcome.

ashutdown

def ashutdown(grace_seconds: float) -> Awaitable[Outcome]

Async counterpart of shutdown.

Program resolution

Resolve a program to its concrete executable path without launching it — a spawn-free preflight ("is this tool installed?") that reuses the same PATH/PATHEXT/execute-bit lookup a real run performs, so it never disagrees with what a spawn would find. The module-level which searches the process PATH; Command.resolve_program() and CliClient.resolve_program() additionally honor a prefer_local directory and a relocated child PATH. A miss raises ProcessNotFound.

which

def which(program: StrPath) -> str

Results & outcomes

What a finished (or streamed) run reports back. A non-zero exit, a timeout, and a signal-kill are all data on these types — never raised by the capturing verbs.

ProcessResult

class ProcessResult

The captured result of a finished run. A non-zero exit, a timeout, and a signal-kill are all reported as data here — never raised by output().

Value semantics: ==/hash() compare every field (program/stdout/stderr/ outcome/success codes — not the incidental duration_seconds/truncated). Not picklable: equality also spans the configured timeout and accepted success_codes, which processkit exposes no accessor to read back, so a pickled result could not reconstruct them and would compare unequal to its original for any command that set .timeout(...)/.success_codes(...); pickling raises TypeError. Pickle result.outcome (an Outcome, which round-trips exactly — e.g. to return it from a concurrent.futures.ProcessPoolExecutor worker), or persist result.stdout/.stderr/.code yourself, to cross a process boundary.

stdout

stdout: str

stderr

stderr: str

code

code: int | None

is_success

is_success: bool

timed_out

timed_out: bool

signal

signal: int | None

program

program: str

duration_seconds

duration_seconds: float

truncated

truncated: bool

combined

combined: str

diagnostic

diagnostic: str | None

The best human-facing message: stderr if it carries text, otherwise stdout, otherwise None if both are blank — the same preference order as NonZeroExit/Timeout/Signalled.diagnostic.

outcome

outcome: Outcome

The full run outcome (code / signal / timed_out), the same value RunProfile.outcome and the checking-verb exceptions expose.

ensure_success

def ensure_success() -> ProcessResult

Raise the same exception a checking verb would if this result's exit isn't in success_codes; returns self unchanged otherwise, so it composes: cmd.output().ensure_success().stdout.

BytesResult

class BytesResult

The captured result of a run with raw-bytes stdout (Command.output_bytes()); stderr stays decoded text. A non-zero exit, a timeout, and a signal-kill are all data here, never raised.

Value semantics: ==/hash() compare every field, same as ProcessResult. Not picklable — raw stdout may not be valid UTF-8 and processkit has no way to reconstruct one from arbitrary bytes outside a real run; pickling raises TypeError. Pickle a ProcessResult (Command.output()) instead, or persist the fields you need yourself.

stdout

stdout: bytes

stderr

stderr: str

code

code: int | None

is_success

is_success: bool

timed_out

timed_out: bool

signal

signal: int | None

program

program: str

duration_seconds

duration_seconds: float

truncated

truncated: bool

Whether captured output was truncated by an output_limit(...) cap — the line-captured stderr under any cap, and (since processkit 2.1.0) the raw stdout too when an output_limit(max_bytes=...) byte ceiling bounds it to a head/tail. A max_lines cap never truncates raw stdout (bytes have no line count); only a max_bytes cap does.

diagnostic

diagnostic: str | None

See ProcessResult.diagnostic. Raw stdout is lossily decoded to text for this message when stderr is blank.

outcome

outcome: Outcome

See ProcessResult.outcome.

ensure_success

def ensure_success() -> BytesResult

See ProcessResult.ensure_success().

Outcome

class Outcome

How a process ended.

There is no is_success here on purpose: an Outcome carries no success_codes context, so it cannot give the command's own success verdict the way ProcessResult.is_success does. Use exited_zero for the literal "exit code 0" test, or compare code against your accepted set.

Value semantics: ==/hash() compare code/signal/timed_out (equivalently, which variant this is and its payload); picklable.

code

code: int | None

signal

signal: int | None

timed_out

timed_out: bool

exited_zero

exited_zero: bool

Finished

class Finished

A process's outcome plus captured stderr (stdout was streamed).

Mirrors Outcome's code, exited_zero, timed_out, and signal directly (in addition to the nested outcome), so callers don't need to reach through .outcome for fields they already use on Outcome. Like Outcome, it exposes exited_zero (literal "exit code 0"), not an is_success that would falsely imply success_codes were considered.

Value semantics: ==/hash() compare outcome/stderr; picklable.

outcome

outcome: Outcome

stderr

stderr: str

code

code: int | None

exited_zero

exited_zero: bool

timed_out

timed_out: bool

signal

signal: int | None

RunProfile

class RunProfile

A resource-usage profile sampled across a run (RunningProcess.profile), plus the run's outcome — profile() is a superset of outcome().

Value semantics: ==/hash() compare every field (outcome/ duration_seconds/cpu_time_seconds/peak_memory_bytes/samples; all exact underneath, though two are exposed here as float). Not picklable — it reports live OS resource-sampling telemetry that processkit has no way to reconstruct outside an actual monitored run; pickling raises TypeError.

code

code: int | None

signal

signal: int | None

timed_out

timed_out: bool

outcome

outcome: Outcome

duration_seconds

duration_seconds: float

cpu_time_seconds

cpu_time_seconds: float | None

peak_memory_bytes

peak_memory_bytes: int | None

samples

samples: int

avg_cpu_cores

avg_cpu_cores: float | None

Streaming & interactive I/O

The live handles a started RunningProcess hands out: async iterators over its output (line by line, or as interleaved stdout/stderr events) and a writable stdin.

StdoutLines

class StdoutLines

Async iterator over a process's stdout, line by line.

OutputEvents

class OutputEvents

Async iterator over stdout + stderr as interleaved OutputEvents.

Yields output lines only. The underlying core stream is the child's whole lifecycle (it also reports process start and exit), but those non-line events are filtered out here rather than surfaced as an OutputEvent with an empty text — which would be indistinguishable from a real blank output line. Process start is RunningProcess.pid; the exit is what the finisher you call afterwards returns.

Draining this iterator to its end also drives the run to completion, so the documented order — iterate fully, then await proc.afinish() (or aoutcome()) — terminates. The finisher then reports that same run.

OutputEvent

class OutputEvent

One captured line and the stream it came from.

Value semantics: ==/hash() compare is_stderr/text; picklable.

stream

stream: Literal['stdout', 'stderr']

is_stderr

is_stderr: bool

text

text: str

ProcessStdin

class ProcessStdin

A writable handle to a running process's stdin (all methods awaitable).

write

def write(data: ReadableBuffer) -> Awaitable[None]

write_line

def write_line(line: str) -> Awaitable[None]

send_control

def send_control(control: str) -> Awaitable[None]

Write one mapped control byte, e.g. "c" -> Ctrl-C (\x03).

This writes a byte to the child's stdin pipe, not a terminal signal; real SIGINT/SIGTSTP delivery requires a pseudoterminal.

flush

def flush() -> Awaitable[None]

close

def close() -> Awaitable[None]

Process groups

Kill-on-drop containment for a whole process tree — start children into it, signal or suspend the group, and reap the entire tree (grandchildren included) on exit. MemberInfo is the enriched per-member snapshot members_info() returns; sample_stats turns a one-shot stats() snapshot into a periodic async series for live monitoring.

ProcessGroup

ProcessGroup(
    *,
    max_memory: int | None = ...,
    max_processes: int | None = ...,
    cpu_quota: float | None = ...,
    shutdown_grace: float | None = ...,
    escalate_to_kill: bool | None = ...,
)

A kill-on-drop container for a process tree; use as a (async) context manager. Also a ProcessRunner in its own right (see _RunnerVerbs): its run verbs run command as a shared member of this group (not a standalone tree) — the same verb surface Runner/ScriptedRunner/… expose (not an extract_runner target, though — see runner.rs).

mechanism

mechanism: Literal['job_object', 'cgroup_v2', 'process_group', 'unknown']

members

def members() -> list[int]

members_info

def members_info() -> list[MemberInfo]

An enriched, point-in-time snapshot of the group's members — the same set as members(), but each pid carried in a MemberInfo alongside best-effort ppid/exe_name/start_time. Synchronous only (the crate offers no async twin). See MemberInfo for the per-field platform matrix and the start_time opacity/pid-reuse note.

signal

def signal(name: SignalName | int) -> None

Send a signal to every process in the tree: a name (term/kill/int/hup/quit/usr1/usr2) or a raw platform signal number (Unix only). On Windows a Job Object has no POSIX signals, so only "kill" is deliverable and any other name/number raises Unsupported. A raw number is validated as a real, deliverable signal (1..=SIGRTMAX on Unix); 0 (the existence probe), negatives, and out-of-range values raise ValueError instead of a silent no-op, and a bool raises TypeError.

suspend

def suspend() -> None

resume

def resume() -> None

kill_all

def kill_all() -> None

stats

def stats() -> ProcessGroupStats

shutdown

def shutdown() -> None

ashutdown

def ashutdown() -> Awaitable[None]

ProcessGroupStats

class ProcessGroupStats

A snapshot of a ProcessGroup's resource usage.

active_process_count

active_process_count: int

peak_memory_bytes

peak_memory_bytes: int | None

total_cpu_time_seconds

total_cpu_time_seconds: float | None

MemberInfo

class MemberInfo

An enriched, point-in-time snapshot of one member of a ProcessGroup's tree — its pid plus best-effort metadata.

The metadata-carrying companion to a bare pid from ProcessGroup.members(). Which members appear follows the same platform matrix as members() (the whole tree on Windows and the Linux cgroup backend; the tracked group leaders on the POSIX process-group fallback). Every field beyond pid is independently None wherever the platform can't report it — never a fabricated value — and a member that exits mid-snapshot is silently omitted, not invented.

The raw command line / environment is deliberately never carried, on any platform: an argv routinely holds secrets, and redaction is the consumer's policy to own.

Field availability (None where the platform can't report it): ppid, exe_name, and start_time are populated on Windows, Linux, and macOS, and are always None on the BSDs (no wired-up per-process reader).

pid

pid: int

The member's process id — always present. Point-in-time, like a pid from members(): pair it with start_time to tell a recycled number apart from the original process.

ppid

ppid: int | None

The member's parent process id, or None where unreadable (always None on the BSDs).

exe_name

exe_name: str | None

The member's short image base name — never a full path, and never a command line (the crate never exposes argv/env). None where unreadable (always None on the BSDs).

start_time

start_time: int | None

An opaque per-process identity token, or None where unreadable — not a wall-clock timestamp. Its unit and epoch are platform-specific (Windows creation FILETIME, 100ns intervals since 1601; Linux /proc/<pid>/stat field 22, clock ticks since boot; macOS microseconds since the Unix epoch; always None on the BSDs), so do not interpret it or compare it across platforms. Its sole use is pairing with pid: two snapshots whose pid and start_time both match name the same process instance, telling a recycled pid apart from the original.

sample_stats

async def sample_stats(
    group: ProcessGroup,
    every: float,
) -> AsyncIterator[ProcessGroupStats]

Sample group.stats() on an interval, forever, as an async series of ProcessGroupStats snapshots — a pure-Python analogue of the crate's ProcessGroup::sample_stats (its StatsSampler borrows the group by lifetime and has no FFI-safe equivalent here; this is plain Python built directly on the already-public group.stats(), living alongside the readiness helpers above for the same reason).

async for snapshot in sample_stats(group, every): ... — the first snapshot is taken immediately (no initial sleep), then one every every seconds, for as long as you keep consuming. There is no overall deadline; stop by breaking out of the loop or otherwise abandoning/closing the generator yourself.

Fused, and louder than the crate's stream. The crate's StatsSampler swallows the error on the first failed sample and just ends the series silently — a caller has to separately call stats() to learn why. This generator instead lets group.stats()'s own exception (a ProcessError — e.g. "ProcessGroup is already closed" once the group has torn down, or an Unsupported/OS-error-derived failure from the platform's resource query) propagate out of the async for untouched — the underlying cause is never hidden behind a quiet end-of-series. That still fuses the series: once this generator function raises, it is exhausted by Python's own async-generator protocol, so a further __anext__ (another loop iteration, a second async for over the same object) raises StopAsyncIteration rather than calling group.stats() again or replaying the same error. If the group is already closed/invalid before the first snapshot (e.g. iteration starts only after group.shutdown() already ran), that same exception surfaces on the very first async for step, not silently as an empty series.

every is validated up front: NaN and negative values raise ValueError (the shared convention with the readiness helpers' timeout/interval). Unlike the crate — which clamps a zero period to 1 ms because tokio panics on a zero-duration interval — every=0 is accepted here as-is: asyncio.sleep(0) has no such restriction, so it means "sample as fast as the event loop allows," with no artificial floor.

Supervision

Keep a command alive: restart it per a policy, with backoff and jitter, until a stop condition is met.

Supervisor

Supervisor(
    command: Command,
    *,
    restart: Literal['always', 'on_crash', 'never'] | None = ...,
    max_restarts: int | None = ...,
    backoff_initial: float | None = ...,
    backoff_factor: float | None = ...,
    max_backoff: float | None = ...,
    jitter: bool | None = ...,
    stop_when: Callable[[ProcessResult], bool] | None = ...,
    give_up_when: Callable[[ProcessResult | ProcessError], bool] | None = ...,
    storm_pause: float | None = ...,
    failure_threshold: float | None = ...,
    failure_decay: float | None = ...,
    capture_max_bytes: int | None = ...,
    capture_max_lines: int | None = ...,
    capture_on_overflow: Literal['drop_oldest', 'drop_newest', 'error'] | None = ...,
    health_check: Callable[[], bool] | None = ...,
    health_check_interval: float | None = ...,
    health_check_failures: int | None = ...,
    runner: RunnerLike | None = ...,
)

Keep a command alive: restart per policy with backoff until a stop condition.

run

def run() -> SupervisionOutcome

arun

def arun() -> Awaitable[SupervisionOutcome]

SupervisionOutcome

class SupervisionOutcome

The result of a Supervisor.run().

Value semantics: ==/hash() compare every field (final_result via ProcessResult's own comparison, plus restarts/stopped/storm_pauses/liveness_kills). Not picklable: its identity includes final_result (a ProcessResult), which cannot be faithfully reconstructed from a pickle (its timeout/ success_codes have no accessor to read back), so pickling raises TypeError. Read the fields you need, or pickle final_result.outcome (an Outcome, which round-trips exactly), to cross a process boundary.

final_result

final_result: ProcessResult

restarts

restarts: int

stopped

stopped: Literal['policy_satisfied', 'predicate', 'restarts_exhausted', 'gave_up', 'unhealthy', 'unknown']

storm_pauses

storm_pauses: int

liveness_kills

liveness_kills: int

Cancellation

A portable cancel switch, wired into a run via Command.cancel_on(), Pipeline.cancel_on(), or CliClient's default_cancel_on=.

CancellationToken

class CancellationToken

A cancel switch: fire it to tear down every run wired to it via Command.cancel_on() / CliClient's default_cancel_on= / Pipeline.cancel_on() — surfacing Cancelled. Cheap to clone/share: every clone refers to the same underlying state, so cancelling any clone cancels every run wired to it. A cancelled token stays cancelled forever.

child_token() derives a separate, scoped token: it is cancelled automatically when this one is, but cancelling it back does NOT propagate to this token or to its other children — cancellation only flows parent-to-child, never child-to-parent or between siblings.

cancel

def cancel() -> None

is_cancelled

def is_cancelled() -> bool

child_token

def child_token() -> CancellationToken

A new token that is cancelled automatically when this one is, but can also be cancelled independently — cancelling the child does not affect this token or its other children.

Batch execution

Run many commands with bounded concurrency, each result — or a ProcessError for a spawn/I/O failure — in its own slot. The output_all family is collect-all (every result in input order once the whole batch finishes); aoutput_as_completed and its _bytes twin instead stream each (index, result) pair as it finishes, for progress and early reaction on a large fan-out.

output_all

def output_all(
    commands: Sequence[Command],
    *,
    concurrency: int | None = ...,
    runner: RunnerLike | None = ...,
) -> list[ProcessResult | ProcessError]

Run a collect-all batch in input order. With concurrency=None, use the process-available CPU count (CPU affinity/cgroup-aware), falling back to 4 if it cannot be determined. A non-positive value raises ValueError.

output_all_bytes

def output_all_bytes(
    commands: Sequence[Command],
    *,
    concurrency: int | None = ...,
    runner: RunnerLike | None = ...,
) -> list[BytesResult | ProcessError]

Raw-bytes output_all with the same concurrency default and validation.

aoutput_all

def aoutput_all(
    commands: Sequence[Command],
    *,
    concurrency: int | None = ...,
    runner: RunnerLike | None = ...,
) -> Awaitable[list[ProcessResult | ProcessError]]

Async output_all with the same concurrency default and validation.

aoutput_all_bytes

def aoutput_all_bytes(
    commands: Sequence[Command],
    *,
    concurrency: int | None = ...,
    runner: RunnerLike | None = ...,
) -> Awaitable[list[BytesResult | ProcessError]]

Async raw-bytes batch with the same concurrency default and validation.

aoutput_as_completed

def aoutput_as_completed(
    commands: Sequence[Command],
    *,
    concurrency: int | None = None,
) -> AsyncIterator[tuple[int, ProcessResult | ProcessError]]

Run commands with bounded concurrency, yielding each (original index, ProcessResult | ProcessError) pair as that command finishes — the streaming, pure-Python counterpart to the compiled aoutput_all.

Where aoutput_all is collect-all (nothing is visible until the whole batch is done), this is an async iterator — async for index, result in aoutput_as_completed(commands, concurrency=8): ... — that hands each result back the moment its command completes, so a large fan-out reports progress and lets you react to early finishers instead of blocking on the slowest command in the batch.

Completion order, not input order. Pairs arrive in the order their commands finish, which is generally not the input order; the index (a command's position in commands) is what re-associates a result with the command that produced it. Every command is yielded exactly once, and the iterator is exhausted once all of them have been.

Errors are per-slot data, not a series-ending raise (aligned with output_all): a command that fails to spawn — or hits an I/O error, or is cancelled through its own CancellationToken — yields its ProcessError in its own pair, and never short-circuits the others. A non-zero exit, a timeout, and a signal-kill are, as everywhere in this library, data on a ProcessResult, not errors at all.

Hard concurrency cap. At most concurrency commands are ever live at once (an asyncio.Semaphore gates each Command.aoutput()), so fanning out hundreds of commands can't exhaust file descriptors or the process table — the same bound aoutput_all gives, held while streaming. concurrency defaults to the process-available CPU count (CPU affinity/cgroup-aware on Python 3.13+), falling back to os.cpu_count() and then 4; this matches the batch family. A non-positive value raises ValueError rather than being silently clamped.

No orphans on cancellation or early exit. Cancelling the task consuming this iterator — or simply breaking out of the async for early — tears down every command still in flight: each Command.aoutput() reaps its whole process subtree (grandchildren included) on cancellation, and this iterator drives that teardown for all live slots before it finishes unwinding. No started child is left orphaned, whether the batch ran to completion, was abandoned partway, or was cancelled outright.

Built directly on Command.aoutput(); unlike the compiled aoutput_all family it takes no runner= double — the streaming layer is deliberately kept minimal, so for a hermetic batch that doesn't need streaming reach for aoutput_all(..., runner=...) instead. For raw bytes output (no UTF-8 decode) use the twin aoutput_as_completed_bytes.

aoutput_as_completed_bytes

def aoutput_as_completed_bytes(
    commands: Sequence[Command],
    *,
    concurrency: int | None = None,
) -> AsyncIterator[tuple[int, BytesResult | ProcessError]]

The raw-bytes twin of aoutput_as_completed: the identical streaming, concurrency-cap, per-slot-error, and no-orphan-on-cancellation contract, but each finished command yields a BytesResult — raw-bytes stdout for non-UTF-8 or binary output, while stderr stays decoded text — in place of a text ProcessResult, mirroring how aoutput_all_bytes relates to aoutput_all. With concurrency=None, it uses the same process-available CPU-count default (and fallbacks) as every other batch entry point. See aoutput_as_completed for the full contract.

Readiness helpers

Asyncio helpers that wait for a condition — a matching output line, an open TCP port, an HTTP endpoint answering with an expected status, a filesystem path or Unix-domain socket, or any polled predicate — bounded by a deadline.

wait_until

async def wait_until(
    predicate: Callable[[], bool | Awaitable[bool]],
    *,
    timeout: float,
    interval: float = 0.05,
) -> None

Poll predicate until it returns true, or timeout seconds elapse.

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

predicate may be synchronous or return an awaitable. Polls every interval seconds; raises WaitTimeout (also a TimeoutError) if the deadline passes first. A synchronous predicate runs on the event loop, so keep it non-blocking — use an async predicate for anything that does I/O. If predicate's awaitable is already a asyncio.Future/asyncio.Task you own, note it is never cancelled by this helper on timeout — only abandoned, so cancel or await it yourself afterwards if that matters.

timeout<=0 contract (shared with wait_for_port / wait_for_line): at timeout=0, predicate is still evaluated (at least once) before any deadline check, so an already-true predicate succeeds instead of failing before it was ever checked. A negative timeout is rejected outright — raises ValueError, same as NaN — rather than being treated as "expired" or silently accepted.

wait_for_line

async def wait_for_line(
    lines: AsyncIterator[Any],
    predicate: str | Callable[[Any], bool],
    *,
    timeout: float,
) -> Any

Consume from an async iterator until predicate matches an item.

predicate is either a callable (predicate(item) -> bool) or, for a str-yielding iterator only, a plain str — a shorthand for "the item contains this substring" (predicate in item). Not just for StdoutLines: any async iterator works (e.g. OutputEvents, with a callable predicate over its OutputEvent items).

Returns the matching item. Raises WaitTimeout (also a TimeoutError, carrying timeout_seconds) if nothing matches within timeout seconds, or propagates whatever predicate or the iterator itself raised (a ProcessError if the stream ends first) untouched — never masked behind the timeout. Items read before the match are consumed; iteration may continue afterward only when a match was found — on a WaitTimeout, exactly how far the iterator advanced past the last inspected item is unspecified (cancellation of the internal scan races the iterator's own advancement), so don't rely on its position after a timeout.

timeout<=0 contract (shared with wait_until / wait_for_port): at timeout=0, the iterator is still scanned (at least one tick), so an item that already matches (already sitting in the iterator) succeeds instead of failing before it was ever inspected. A negative timeout is rejected outright — raises ValueError, same as NaN — rather than being treated as "expired" or silently accepted.

wait_for_port

async def wait_for_port(
    host: str,
    port: int,
    *,
    timeout: float,
    interval: float = 0.05,
) -> None

Wait until a TCP connection to (host, port) succeeds.

Polls every interval seconds until the port accepts a connection or timeout seconds elapse, in which case WaitTimeout (also a TimeoutError) is raised — carrying host/port — chained from the last connection attempt's exception (e.g. a DNS failure survives as the cause instead of being silently dropped).

timeout<=0 contract (shared with wait_until / wait_for_line): at timeout=0, a connection attempt is still made (at least one), so an already-ready port succeeds instead of failing before a connection was ever tried — this first attempt is not cut short by the already-expired deadline. It IS bounded, though: to a short, fixed event-loop tick (or a smaller caller-supplied interval), not left uncapped — an unresolvable/blackhole address would otherwise be free to block on the OS's own (much longer, or absent) connect/DNS timeout well past the caller's requested deadline. A negative timeout is rejected outright — raises ValueError, same as NaN — rather than being treated as "expired" or silently accepted.

wait_for_http

async def wait_for_http(
    host: str,
    port: int,
    path: str = '/',
    *,
    timeout: float,
    interval: float = 0.05,
    expected_status: Container[int] | Callable[[int], bool] | None = None,
) -> None

Wait until an HTTP GET of http://host:port/path answers with an acceptable status code.

A stronger readiness signal than wait_for_port: a server often accepts TCP connections while still warming up and answering 503, so a bare port probe reports ready too early. This one performs a minimal HTTP/1.1 GET (hand-rolled over asyncio.open_connection — no http.client / urllib / third-party dependency) every interval seconds and succeeds only once the response's status code is accepted.

expected_status decides what "accepted" means: either a container tested with in or a predicate Callable[[int], bool] for arbitrary logic (e.g. lambda c: c == 204). The default (None) accepts any 2xx code — equivalent to passing range(200, 300). The whole request/response is bounded by the deadline, so a server that accepts the connection but never answers can't outlive timeout.

On failure the deadline raises WaitTimeout (also a TimeoutError), carrying host / port / path and chained (as __cause__) from the last attempt's failure — a connection error (e.g. a refused connect or a DNS failure) or a ProcessError recording the last unexpected status code — so the evidence for why it never became ready survives.

timeout<=0 contract (shared with wait_until / wait_for_port / wait_for_line / wait_for_path): at timeout=0 one request attempt is still made (at least one), so an already-ready endpoint succeeds instead of failing before it was ever probed; that first attempt is bounded to a short, fixed event-loop tick (or a smaller caller-supplied interval), never left uncapped. A negative timeout is rejected outright — raises ValueError, same as NaN — as is a non-positive interval.

host and path are validated up front, before any connection is attempted (fail-fast, not "after one retry cycle"): an IPv6 literal host (e.g. "::1") is bracketed in the Host header per RFC 9112/3986 (Host: [::1]:8080, never the ambiguous Host: ::1:8080); a path containing whitespace or a control character (including CR/LF — which could otherwise inject extra request/header lines from an untrusted path) raises ValueError; and a host/path with a character that can't be encoded as latin-1 (required for the request line) raises ValueError instead of a raw UnicodeEncodeError.

wait_for_path

async def wait_for_path(
    path: StrPath,
    *,
    timeout: float,
    interval: float = 0.05,
) -> None

Wait until path exists on the filesystem.

Polls every interval seconds until path.exists() returns true or timeout seconds elapse, in which case WaitTimeout (also a TimeoutError) is raised, carrying path. A unix-socket, a pid file, or any other marker file a daemon creates once ready are all typical uses. For a Unix-domain socket that must actually accept connections, use wait_for_unix_socket; for a TCP port or an arbitrary predicate, see wait_for_port / wait_until instead (wait_until(lambda: path.exists(), ...) is exactly what this helper does, named for readability and given the same WaitTimeout discipline as its siblings).

timeout<=0 contract (shared with wait_until / wait_for_port / wait_for_line): at timeout=0, path is still checked (at least once) before any deadline check, so an already-existing path succeeds instead of failing before it was ever checked. A negative timeout is rejected outright — raises ValueError, same as NaN — rather than being treated as "expired" or silently accepted.

wait_for_unix_socket

async def wait_for_unix_socket(
    path: StrPath,
    *,
    timeout: float,
    interval: float = 0.05,
) -> None

Wait until a Unix-domain socket at path accepts a connection.

Unlike wait_for_path, this proves that the socket has started accepting connections, rather than only that its filesystem entry exists. Polls every interval seconds until a connection succeeds or timeout seconds elapse, in which case WaitTimeout (also a TimeoutError) is raised, carrying path and chained from the last connection failure.

Platforms lacking Unix-domain-socket support — no socket.AF_UNIX or no asyncio.open_unix_connection (asyncio binds the latter only when the former existed at import) — raise Unsupported instead of silently downgrading to a filesystem-existence check. At timeout=0 one bounded connection attempt still runs, so an already-ready socket succeeds; negative and NaN timeouts are rejected with ValueError.

WaitTimeout

WaitTimeout(
    message: str,
    *,
    timeout_seconds: float,
    host: str | None = None,
    port: int | None = None,
    path: StrPath | None = None,
)

A readiness helper (wait_until / wait_for_line / wait_for_port / wait_for_http / wait_for_path / wait_for_unix_socket) didn't succeed within its deadline.

Also a builtin TimeoutError, so except TimeoutError catches it too — the same convention a run's own .timeout() uses (see Timeout). Always carries timeout_seconds; wait_for_port and wait_for_http additionally set host / port (and wait_for_http also path), while wait_for_path and wait_for_unix_socket set path (all None for wait_until / wait_for_line, which have none of these). wait_for_port / wait_for_http, and wait_for_unix_socket also chain the last attempt's failure as __cause__ (a connection error, or — for wait_for_http — the last unexpected status code).

timeout_seconds

timeout_seconds = timeout_seconds

host

host = host

port

port = port

path

path = path

Observability

Opt-in bridging of the core's per-run tracing events to Python logging.

enable_logging

def enable_logging() -> bool

The runner seam

The dependency-injection seam: annotate your code against a protocol, inject the real Runner in production and a test double (see the Testing section) in tests. ProcessRunner is the capture/check verbs; StreamingRunner adds start/astart.

ProcessRunner

class ProcessRunner

The capture/check run verbs as a structural type: output/output_bytes/ run/exit_code/probe and their a-prefixed async twins — no streaming.

Every built-in runner satisfies this (and the wider StreamingRunner). Prefer this narrower protocol when your own code only calls these verbs — a hand-rolled double then only needs to implement five verbs (times two for the async twins), not the full runner surface. CliClient also satisfies ProcessRunner: each capture/check verb accepts either per-call Args (which it combines with its bound program) or a Command (whose explicit settings win over client defaults). It is not a StreamingRunner, because it has no start/astart verbs.

output

def output(command: Command, /) -> ProcessResult

output_bytes

def output_bytes(command: Command, /) -> BytesResult

run

def run(command: Command, /) -> str

exit_code

def exit_code(command: Command, /) -> int

probe

def probe(command: Command, /) -> bool

aoutput

def aoutput(command: Command, /) -> Awaitable[ProcessResult]

aoutput_bytes

def aoutput_bytes(command: Command, /) -> Awaitable[BytesResult]

arun

def arun(command: Command, /) -> Awaitable[str]

aexit_code

def aexit_code(command: Command, /) -> Awaitable[int]

aprobe

def aprobe(command: Command, /) -> Awaitable[bool]

StreamingRunner

class StreamingRunner

ProcessRunner plus start/astart — the full runner verb surface, for code that also needs a live RunningProcess handle to stream.

Runner, ScriptedRunner, RecordReplayRunner, RecordingRunner, and DryRunRunner all satisfy it. A hand-rolled double can implement the capture/check verbs easily, but start/astart must return a RunningProcess, which has no public constructor — and the built-in runners are @final, so a fully-conforming custom runner in practice means wrapping one (delegating start/astart to it; use ScriptedRunner for streaming doubles).

start

def start(command: Command, /) -> RunningProcess

astart

def astart(command: Command, /) -> Awaitable[RunningProcess]

Runner

class Runner

The real process runner — inject it for testable code.

Exceptions

Every error raised by the package descends from ProcessError, so a single except ProcessError catches them all. Timeout, ProcessNotFound, and PermissionDenied also subclass a builtin (TimeoutError / FileNotFoundError / PermissionError, each itself an OSError), so the stdlib except clauses catch them too.

ProcessError

class ProcessError

Base class for every error raised by this package.

NonZeroExit

class NonZeroExit

run() / exit_code() got a non-zero exit.

program

program: str

code

code: int

stdout

stdout: str

stderr

stderr: str

stdout_bytes

stdout_bytes: bytes | None

diagnostic

diagnostic: str | None

Timeout

class Timeout

A run exceeded its configured timeout.

Also a builtin TimeoutError, so except TimeoutError catches it too — and since TimeoutError is itself an OSError subclass (as of Python 3.3), except OSError catches it as well (the same is true of ProcessNotFound/FileNotFoundError and PermissionDenied/PermissionError below — all three dual-base exceptions are transitively OSError).