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

program

program: str

timeout_seconds

timeout_seconds: float | None

stdout

stdout: str

stderr

stderr: str

stdout_bytes

stdout_bytes: bytes | None

diagnostic

diagnostic: str | None

IdleTimeout

class IdleTimeout

A run produced no stdout/stderr line for its Command.idle_timeout(...) window (the child went silent) and was killed.

A deliberate sibling of Timeout, not a subclass: an idle (inactivity) timeout is a distinct condition from a wall-clock timeout() expiry — "the child went silent" vs "the run took too long overall" — so except IdleTimeout does not swallow a wall-clock Timeout and vice-versa, while except ProcessError still catches both. Raised from the streaming iterators (stdout_lines() / output_events()) on the handle from start()/astart(); the one-shot capture verbs do not enforce idle_timeout (see its docstring).

idle_timeout_seconds

idle_timeout_seconds: float

Signalled

class Signalled

A run was killed by a signal.

program

program: str

signal

signal: int | None

stdout

stdout: str

stderr

stderr: str

stdout_bytes

stdout_bytes: bytes | None

diagnostic

diagnostic: str | None

ProcessNotFound

class ProcessNotFound

The program could not be found / spawned.

Also a builtin FileNotFoundError (what subprocess raises), so except FileNotFoundError catches it too.

program

program: str

searched

searched: str | None

PermissionDenied

class PermissionDenied

The program could not be spawned because of insufficient permissions (e.g. a non-executable file), or a permission-denied OS error surfaced from elsewhere in the run (e.g. a group signal the OS refused).

Also a builtin PermissionError, so except PermissionError catches it too.

program

program: str | None

ResourceLimit

class ResourceLimit

A resource limit (memory / processes / CPU) was invalid or could not be enforced by the active containment mechanism. The reason is the exception message (str(exc)); it carries no extra structured field.

Unsupported

class Unsupported

The operation is not supported on this platform.

operation

operation: str

OutputTooLarge

class OutputTooLarge

Captured output hit an output_limit(..., on_overflow="error") ceiling.

total_bytes (and the max_bytes ceiling it crossed) count raw bytes read from the child's output pipe — line terminators and invalid-UTF-8 bytes included — not the bytes of the decoded text on ProcessResult.stdout, so total_bytes can exceed len(stdout.encode()) for the same run. total_lines is the line count of the line-captured output.

program

program: str

max_lines

max_lines: int | None

max_bytes

max_bytes: int | None

total_lines

total_lines: int

total_bytes

total_bytes: int

Cancelled

class Cancelled

The run was deliberately cancelled via a CancellationToken wired with Command.cancel_on() / CliClient's default_cancel_on= / Pipeline.cancel_on(). Terminal — never retried by Command.retry() or restarted by Supervisor (the token stays cancelled forever, so another attempt could only fail the same way).

program

program: str

InvalidJson

class InvalidJson

CliClient.run_json() / arun_json() ran the command successfully (a zero exit, like run) but its stdout did not parse as JSON.

A ProcessError subclass raised in place of a bare json.JSONDecodeError, so the failure is attributed (which program, and what the parser reported in str(exc)) and a single except ProcessError still catches it. A deliberate sibling of NonZeroExit, not a subclass: the run itself succeeded — only its output shape is wrong — so except InvalidJson isolates a bad-payload failure without also catching a genuine non-zero exit.

program

program: str

stdout

stdout: str

Type aliases

Exported so your own wrappers can annotate against the same types the API accepts.

Args

Args = list[str] | list[Path] | list[os.PathLike[str]] | tuple[StrPath, ...]

LineTerminatorName

LineTerminatorName = Literal['newline', 'carriage_return']

Priority

Priority = Literal['idle', 'below_normal', 'normal', 'above_normal', 'high']

ReadableBuffer

ReadableBuffer = bytes | bytearray | memoryview

RetryIf

RetryIf = Literal['transient', 'transient_or_timeout']

SignalName

SignalName = Literal['term', 'kill', 'int', 'hup', 'quit', 'usr1', 'usr2']

StrPath

StrPath = str | os.PathLike[str]

Testing

Runner test doubles, in the processkit.testing submodule. Inject one in tests — all satisfy the ProcessRunner protocol — so the code under test spawns no real processes.

ScriptedRunner

class ScriptedRunner

A scripted test double for Runner.

on

def on(prefix: Args, reply: Reply) -> None

on_sequence

def on_sequence(prefix: Args, replies: Sequence[Reply]) -> None

when

def when(predicate: Callable[[Command], bool], reply: Reply) -> None

fallback

def fallback(reply: Reply) -> None

RecordReplayRunner

class RecordReplayRunner

Records real runs to a cassette file (record) and replays them without spawning (replay); shares the Runner run-verb surface.

record

def record(path: StrPath) -> RecordReplayRunner

replay

def replay(path: StrPath) -> RecordReplayRunner

save

def save() -> None

RecordingRunner

class RecordingRunner

A recording test double: replies to every command with a canned Reply and records each call, so a test can assert on what its code ran. Shares the Runner run-verb surface; inspect calls with calls() / only_call().

replying

def replying(reply: Reply) -> RecordingRunner

new

def new(inner: RunnerLike) -> RecordingRunner

calls

def calls() -> list[Invocation]

only_call

def only_call() -> Invocation

DryRunRunner

class DryRunRunner

A dry-run test double: never spawns a process. Every verb renders the command to its display-quoted line (like Command.command_line()) and returns a synthetic successful result — the seam behind a tool's own --dry-run/--echo mode. Shares the Runner run-verb surface; inspect the rendered lines with commands() / only_command(), or stream them live with on_invocation().

on_invocation

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

commands

def commands() -> list[str]

only_command

def only_command() -> str

Reply

class Reply

A canned reply for a ScriptedRunner rule.

ok

def ok(stdout: str) -> Reply

fail

def fail(code: int, stderr: str) -> Reply

timeout

def timeout() -> Reply

signalled

def signalled(signal: int | None = ...) -> Reply

pending

def pending() -> Reply

lines

def lines(lines: Sequence[str]) -> Reply

with_stdout

def with_stdout(stdout: str) -> Reply

with_stderr

def with_stderr(stderr: str) -> Reply

with_line_delay

def with_line_delay(seconds: float) -> Reply

Invocation

class Invocation

One call captured by a RecordingRunner: the program, args, cwd, env overrides, and whether stdin was supplied. Values are inspectable for assertions; the repr stays redacted (program, arg count, cwd, env names, has_stdin — never argv or env values).

program

program: str

args

args: list[str]

cwd

cwd: str | None

env

env: dict[str, str | None]

env_is

def env_is(name: str, value: str) -> bool

has_env

def has_env(name: str) -> bool

has_stdin

has_stdin: bool

has_flag

def has_flag(flag: str) -> bool