Skip to content

Runner API Reference

Auto-generated API documentation for the runner module.

runner

Structured subprocess execution — zero dependencies, stdlib only, Python 3.10+.

Part of zerodep: https://github.com/Oaklight/zerodep Copyright (c) 2026 Peng Ding. MIT License.

Run external commands with controlled execution: timeouts with graceful kill escalation, streaming output, environment isolation, and cross-platform support. Designed as a building block (Layer 1) for higher-level execution frameworks.

Quick start::

from runner import run

result = run("echo hello world")
print(result.stdout)          # "hello world

" print(result.returncode) # 0 print(result.duration) # 0.003 (seconds)

Streaming output::

from runner import stream

with stream(["make", "build"]) as proc:
    for line in proc.iter_lines():
        print(f"[build] {line}", end="")

Async execution::

import asyncio
from runner import run_async

result = asyncio.run(run_async("ls -la"))

Requires Python 3.10+.

RunnerError

Bases: Exception

Base exception for all runner operations.

Source code in runner/runner.py 
81
82
class RunnerError(Exception):
    """Base exception for all runner operations."""

CommandNotFoundError

Bases: RunnerError

Raised when the command binary cannot be located on PATH.

Attributes:

Name Type Description
name

The command name that was not found.
Source code in runner/runner.py 
85
86
87
88
89
90
91
92
93
94
class CommandNotFoundError(RunnerError):
    """Raised when the command binary cannot be located on PATH.

    Attributes:
        name: The command name that was not found.
    """

    def __init__(self, name: str) -> None:
        self.name = name
        super().__init__(f"Command not found: {name}")

CommandFailedError

Bases: RunnerError

Raised when a command exits with a disallowed return code.

Attributes:

Name Type Description
result

The full RunResult including stdout, stderr, returncode, duration.
Source code in runner/runner.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class CommandFailedError(RunnerError):
    """Raised when a command exits with a disallowed return code.

    Attributes:
        result: The full RunResult including stdout, stderr, returncode,
            duration.
    """

    def __init__(self, result: RunResult) -> None:
        self.result = result
        cmd_str = " ".join(result.command)
        msg = f"Command failed (rc={result.returncode}): {cmd_str}"
        if result.stderr.strip():
            msg += f"\n{result.stderr.rstrip()}"
        super().__init__(msg)

CommandTimeoutError

Bases: RunnerError

Raised when a command exceeds its timeout.

Attributes:

Name Type Description
command

The command that timed out.
timeout

The timeout value in seconds.
partial_stdout

Any stdout captured before the timeout.
partial_stderr

Any stderr captured before the timeout.
Source code in runner/runner.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class CommandTimeoutError(RunnerError):
    """Raised when a command exceeds its timeout.

    Attributes:
        command: The command that timed out.
        timeout: The timeout value in seconds.
        partial_stdout: Any stdout captured before the timeout.
        partial_stderr: Any stderr captured before the timeout.
    """

    def __init__(
        self,
        command: tuple[str, ...],
        timeout: float,
        partial_stdout: str = "",
        partial_stderr: str = "",
    ) -> None:
        self.command = command
        self.timeout = timeout
        self.partial_stdout = partial_stdout
        self.partial_stderr = partial_stderr
        cmd_str = " ".join(command)
        super().__init__(f"Command timed out after {timeout}s: {cmd_str}")

CommandBlockedError

Bases: RunnerError

Raised when a command is rejected by allowlist/blocklist policy.

Attributes:

Name Type Description
command

The rejected command name.
reason

Human-readable explanation.
Source code in runner/runner.py 
139
140
141
142
143
144
145
146
147
148
149
150
class CommandBlockedError(RunnerError):
    """Raised when a command is rejected by allowlist/blocklist policy.

    Attributes:
        command: The rejected command name.
        reason: Human-readable explanation.
    """

    def __init__(self, command: str, reason: str) -> None:
        self.command = command
        self.reason = reason
        super().__init__(f"Command blocked: {command} ({reason})")

RunResult dataclass

Result of a completed command execution.

Attributes:

Name Type Description
command tuple[str, ...]

The command and arguments as a tuple.
returncode int

Process exit code.
stdout str

Captured standard output (decoded text).
stderr str

Captured standard error (decoded text).
duration float

Wall-clock execution time in seconds.
pid int

Process ID of the executed command.
Source code in runner/runner.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
@dataclasses.dataclass(frozen=True, slots=True)
class RunResult:
    """Result of a completed command execution.

    Attributes:
        command: The command and arguments as a tuple.
        returncode: Process exit code.
        stdout: Captured standard output (decoded text).
        stderr: Captured standard error (decoded text).
        duration: Wall-clock execution time in seconds.
        pid: Process ID of the executed command.
    """

    command: tuple[str, ...]
    returncode: int
    stdout: str
    stderr: str
    duration: float
    pid: int

StreamHandle

Live handle to a running process for streaming output.

Returned by the :func:stream context manager. Provides iterators over stdout and/or stderr lines and process control methods.

Lifecycle
  1. Create — the :func:stream context manager starts the subprocess, optionally writes input to stdin, then yields this handle. At this point the handle owns the process.
  2. Yield lines — the caller iterates via :meth:iter_lines or :meth:iter_any. Lines are yielded as they arrive.
  3. Cleanup — when the with block exits (normally or via exception), :meth:_cleanup is called automatically.
Process ownership

The handle takes exclusive ownership of the underlying :class:subprocess.Popen object. Callers must not interact with the Popen directly. The handle is responsible for ensuring the process is terminated when the context exits.

Unconsumed output

If the caller does not fully consume stdout/stderr (e.g. breaks out of the iterator early), the remaining pipe data is discarded during cleanup. The process is still terminated cleanly via :meth:_cleanup.

Cleanup semantics

:meth:_cleanup checks whether the process is still running (via poll()). If so, it calls :func:_terminate_with_escalation (SIGTERM, then SIGKILL after kill_delay seconds). The returncode attribute is then set from the process exit code. Cleanup is invoked by the finally clause in :func:stream, so it always runs even if the caller raises an exception.

Attributes:

Name Type Description
pid int

Process ID.
Source code in runner/runner.py 
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
class StreamHandle:
    """Live handle to a running process for streaming output.

    Returned by the :func:`stream` context manager.  Provides iterators
    over stdout and/or stderr lines and process control methods.

    Lifecycle:
        1. **Create** — the :func:`stream` context manager starts the
           subprocess, optionally writes *input* to stdin, then yields
           this handle.  At this point the handle **owns** the process.
        2. **Yield lines** — the caller iterates via :meth:`iter_lines`
           or :meth:`iter_any`.  Lines are yielded as they arrive.
        3. **Cleanup** — when the ``with`` block exits (normally or via
           exception), :meth:`_cleanup` is called automatically.

    Process ownership:
        The handle takes exclusive ownership of the underlying
        :class:`subprocess.Popen` object.  Callers must not interact
        with the ``Popen`` directly.  The handle is responsible for
        ensuring the process is terminated when the context exits.

    Unconsumed output:
        If the caller does not fully consume stdout/stderr (e.g. breaks
        out of the iterator early), the remaining pipe data is
        discarded during cleanup.  The process is still terminated
        cleanly via :meth:`_cleanup`.

    Cleanup semantics:
        :meth:`_cleanup` checks whether the process is still running
        (via ``poll()``).  If so, it calls
        :func:`_terminate_with_escalation` (SIGTERM, then SIGKILL after
        *kill_delay* seconds).  The ``returncode`` attribute is then
        set from the process exit code.  Cleanup is invoked by the
        ``finally`` clause in :func:`stream`, so it always runs even
        if the caller raises an exception.

    Attributes:
        pid: Process ID.
    """

    def __init__(
        self,
        proc: subprocess.Popen,  # type: ignore[type-arg]
        encoding: str,
        timeout: float | None,
        kill_delay: float,
    ) -> None:
        self._proc = proc
        self._encoding = encoding
        self._timeout = timeout
        self._kill_delay = kill_delay
        self.pid: int = proc.pid
        self._returncode: int | None = None

    @property
    def returncode(self) -> int | None:
        """Exit code, available after iteration completes or process exits."""
        if self._returncode is not None:
            return self._returncode
        rc = self._proc.poll()
        if rc is not None:
            self._returncode = rc
        return self._returncode

    def iter_lines(self, *, source: str = "stdout") -> Iterator[str]:
        """Iterate over lines from stdout or stderr.

        Args:
            source: ``"stdout"`` or ``"stderr"``.

        Yields:
            Lines of text (including trailing newline).
        """
        pipe = self._proc.stdout if source == "stdout" else self._proc.stderr
        if pipe is None:
            return
        try:
            for line in pipe:
                yield line
        except ValueError:
            pass
        self._returncode = self._proc.wait()

    def iter_any(self) -> Iterator[tuple[str, str]]:
        """Iterate over interleaved lines from both stdout and stderr.

        Yields:
            Tuples of ``(source, line)`` where source is ``"stdout"``
            or ``"stderr"``.
        """
        q: queue.Queue[tuple[str, str] | None] = queue.Queue()

        def _reader(pipe: IO[str], label: str) -> None:
            try:
                for line in pipe:
                    q.put((label, line))
            except ValueError:
                # Tier 2: best-effort observable — expected on pipe close
                pass
            finally:
                q.put(None)

        threads = []
        for pipe, label in [
            (self._proc.stdout, "stdout"),
            (self._proc.stderr, "stderr"),
        ]:
            if pipe is not None:
                t = threading.Thread(target=_reader, args=(pipe, label), daemon=True)
                t.start()
                threads.append(t)

        finished = 0
        total = len(threads)
        while finished < total:
            item = q.get()
            if item is None:
                finished += 1
            else:
                yield item

        for t in threads:
            t.join(timeout=2)
        self._returncode = self._proc.wait()

    def kill(self) -> None:
        """Forcibly kill the process."""
        self._proc.kill()
        self._proc.wait()

    def _cleanup(self) -> None:
        """Ensure the process is terminated and returncode is captured.

        Called automatically by the :func:`stream` context manager on
        exit.  If the process is still running (``poll()`` returns
        ``None``), escalates through SIGTERM -> SIGKILL.  Always sets
        ``self._returncode`` from the process exit code.
        """
        if self._proc.poll() is None:
            _terminate_with_escalation(self._proc, self._kill_delay)
        self._returncode = self._proc.returncode

returncode property

Exit code, available after iteration completes or process exits.

iter_lines(*, source='stdout')

Iterate over lines from stdout or stderr.

Parameters:

Name Type Description Default
source str

"stdout" or "stderr".

 'stdout'

Yields:

Type Description
str

Lines of text (including trailing newline).
Source code in runner/runner.py 
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
def iter_lines(self, *, source: str = "stdout") -> Iterator[str]:
    """Iterate over lines from stdout or stderr.

    Args:
        source: ``"stdout"`` or ``"stderr"``.

    Yields:
        Lines of text (including trailing newline).
    """
    pipe = self._proc.stdout if source == "stdout" else self._proc.stderr
    if pipe is None:
        return
    try:
        for line in pipe:
            yield line
    except ValueError:
        pass
    self._returncode = self._proc.wait()

iter_any()

Iterate over interleaved lines from both stdout and stderr.

Yields:

Type Description
str

Tuples of (source, line) where source is "stdout"
str

or "stderr".
Source code in runner/runner.py 
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
def iter_any(self) -> Iterator[tuple[str, str]]:
    """Iterate over interleaved lines from both stdout and stderr.

    Yields:
        Tuples of ``(source, line)`` where source is ``"stdout"``
        or ``"stderr"``.
    """
    q: queue.Queue[tuple[str, str] | None] = queue.Queue()

    def _reader(pipe: IO[str], label: str) -> None:
        try:
            for line in pipe:
                q.put((label, line))
        except ValueError:
            # Tier 2: best-effort observable — expected on pipe close
            pass
        finally:
            q.put(None)

    threads = []
    for pipe, label in [
        (self._proc.stdout, "stdout"),
        (self._proc.stderr, "stderr"),
    ]:
        if pipe is not None:
            t = threading.Thread(target=_reader, args=(pipe, label), daemon=True)
            t.start()
            threads.append(t)

    finished = 0
    total = len(threads)
    while finished < total:
        item = q.get()
        if item is None:
            finished += 1
        else:
            yield item

    for t in threads:
        t.join(timeout=2)
    self._returncode = self._proc.wait()

kill()

Forcibly kill the process.

Source code in runner/runner.py 
1023
1024
1025
1026
def kill(self) -> None:
    """Forcibly kill the process."""
    self._proc.kill()
    self._proc.wait()

AsyncStreamHandle

Async live handle to a running process for streaming output.

Returned by the :func:stream_async async context manager.

Lifecycle
  1. Create — :func:stream_async starts the subprocess via asyncio.create_subprocess_exec, optionally writes input to stdin, then yields this handle. The handle owns the process from this point.
  2. Yield lines — the caller iterates via :meth:aiter_lines or :meth:aiter_any. Lines are yielded as they arrive from the async stream readers.
  3. Cleanup — when the async with block exits (normally or via exception), :meth:_cleanup is awaited automatically.
Process ownership

The handle takes exclusive ownership of the underlying asyncio.subprocess.Process. Callers must not interact with the process object directly.

Unconsumed output

If the caller does not fully consume stdout/stderr (e.g. breaks out of the async iterator early), remaining pipe data is discarded during cleanup. The process is still terminated cleanly via :meth:_cleanup.

Cleanup semantics

:meth:_cleanup checks proc.returncode. If None (the process is still running), it awaits :func:_async_terminate_with_escalation (SIGTERM, then SIGKILL after kill_delay seconds). The returncode attribute is then set. Cleanup is invoked by the finally clause in :func:stream_async.

Attributes:

Name Type Description
pid int

Process ID.
Source code in runner/runner.py 
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
class AsyncStreamHandle:
    """Async live handle to a running process for streaming output.

    Returned by the :func:`stream_async` async context manager.

    Lifecycle:
        1. **Create** — :func:`stream_async` starts the subprocess via
           ``asyncio.create_subprocess_exec``, optionally writes *input*
           to stdin, then yields this handle.  The handle **owns** the
           process from this point.
        2. **Yield lines** — the caller iterates via
           :meth:`aiter_lines` or :meth:`aiter_any`.  Lines are yielded
           as they arrive from the async stream readers.
        3. **Cleanup** — when the ``async with`` block exits (normally
           or via exception), :meth:`_cleanup` is awaited automatically.

    Process ownership:
        The handle takes exclusive ownership of the underlying
        ``asyncio.subprocess.Process``.  Callers must not interact with
        the process object directly.

    Unconsumed output:
        If the caller does not fully consume stdout/stderr (e.g. breaks
        out of the async iterator early), remaining pipe data is
        discarded during cleanup.  The process is still terminated
        cleanly via :meth:`_cleanup`.

    Cleanup semantics:
        :meth:`_cleanup` checks ``proc.returncode``.  If ``None`` (the
        process is still running), it awaits
        :func:`_async_terminate_with_escalation` (SIGTERM, then SIGKILL
        after *kill_delay* seconds).  The ``returncode`` attribute is
        then set.  Cleanup is invoked by the ``finally`` clause in
        :func:`stream_async`.

    Attributes:
        pid: Process ID.
    """

    def __init__(
        self,
        proc: asyncio.subprocess.Process,
        encoding: str,
        timeout: float | None,
        kill_delay: float,
    ) -> None:
        self._proc = proc
        self._encoding = encoding
        self._timeout = timeout
        self._kill_delay = kill_delay
        self.pid: int = proc.pid  # type: ignore[assignment]
        self._returncode: int | None = None

    @property
    def returncode(self) -> int | None:
        """Exit code, available after iteration completes or process exits."""
        if self._returncode is not None:
            return self._returncode
        rc = self._proc.returncode
        if rc is not None:
            self._returncode = rc
        return self._returncode

    async def aiter_lines(self, *, source: str = "stdout") -> AsyncIterator[str]:
        """Iterate over lines from stdout or stderr.

        Args:
            source: ``"stdout"`` or ``"stderr"``.

        Yields:
            Lines of text (including trailing newline).
        """
        stream_reader = self._proc.stdout if source == "stdout" else self._proc.stderr
        if stream_reader is None:
            return
        while True:
            line_bytes = await stream_reader.readline()
            if not line_bytes:
                break
            yield line_bytes.decode(self._encoding)
        await self._proc.wait()
        self._returncode = self._proc.returncode

    async def aiter_any(self) -> AsyncIterator[tuple[str, str]]:
        """Iterate over interleaved lines from both stdout and stderr.

        Yields:
            Tuples of ``(source, line)`` where source is ``"stdout"``
            or ``"stderr"``.
        """
        q: asyncio.Queue[tuple[str, str] | None] = asyncio.Queue()

        async def _reader(
            stream_reader: asyncio.StreamReader | None, label: str
        ) -> None:
            if stream_reader is None:
                await q.put(None)
                return
            while True:
                line_bytes = await stream_reader.readline()
                if not line_bytes:
                    break
                await q.put((label, line_bytes.decode(self._encoding)))
            await q.put(None)

        tasks = [
            asyncio.create_task(_reader(self._proc.stdout, "stdout")),
            asyncio.create_task(_reader(self._proc.stderr, "stderr")),
        ]

        finished = 0
        while finished < len(tasks):
            item = await q.get()
            if item is None:
                finished += 1
            else:
                yield item

        await asyncio.gather(*tasks)
        await self._proc.wait()
        self._returncode = self._proc.returncode

    async def kill(self) -> None:
        """Forcibly kill the process."""
        self._proc.kill()
        await self._proc.wait()

    async def _cleanup(self) -> None:
        """Ensure the process is terminated and returncode is captured.

        Called automatically by the :func:`stream_async` context manager
        on exit.  If the process is still running (``returncode is
        None``), escalates through SIGTERM -> SIGKILL.  Always sets
        ``self._returncode`` from the process exit code.
        """
        if self._proc.returncode is None:
            await _async_terminate_with_escalation(self._proc, self._kill_delay)
        self._returncode = self._proc.returncode

returncode property

Exit code, available after iteration completes or process exits.

aiter_lines(*, source='stdout') async

Iterate over lines from stdout or stderr.

Parameters:

Name Type Description Default
source str

"stdout" or "stderr".

 'stdout'

Yields:

Type Description
AsyncIterator[str]

Lines of text (including trailing newline).
Source code in runner/runner.py 
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
async def aiter_lines(self, *, source: str = "stdout") -> AsyncIterator[str]:
    """Iterate over lines from stdout or stderr.

    Args:
        source: ``"stdout"`` or ``"stderr"``.

    Yields:
        Lines of text (including trailing newline).
    """
    stream_reader = self._proc.stdout if source == "stdout" else self._proc.stderr
    if stream_reader is None:
        return
    while True:
        line_bytes = await stream_reader.readline()
        if not line_bytes:
            break
        yield line_bytes.decode(self._encoding)
    await self._proc.wait()
    self._returncode = self._proc.returncode

aiter_any() async

Iterate over interleaved lines from both stdout and stderr.

Yields:

Type Description
AsyncIterator[tuple[str, str]]

Tuples of (source, line) where source is "stdout"
AsyncIterator[tuple[str, str]]

or "stderr".
Source code in runner/runner.py 
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
async def aiter_any(self) -> AsyncIterator[tuple[str, str]]:
    """Iterate over interleaved lines from both stdout and stderr.

    Yields:
        Tuples of ``(source, line)`` where source is ``"stdout"``
        or ``"stderr"``.
    """
    q: asyncio.Queue[tuple[str, str] | None] = asyncio.Queue()

    async def _reader(
        stream_reader: asyncio.StreamReader | None, label: str
    ) -> None:
        if stream_reader is None:
            await q.put(None)
            return
        while True:
            line_bytes = await stream_reader.readline()
            if not line_bytes:
                break
            await q.put((label, line_bytes.decode(self._encoding)))
        await q.put(None)

    tasks = [
        asyncio.create_task(_reader(self._proc.stdout, "stdout")),
        asyncio.create_task(_reader(self._proc.stderr, "stderr")),
    ]

    finished = 0
    while finished < len(tasks):
        item = await q.get()
        if item is None:
            finished += 1
        else:
            yield item

    await asyncio.gather(*tasks)
    await self._proc.wait()
    self._returncode = self._proc.returncode

kill() async

Forcibly kill the process.

Source code in runner/runner.py 
1243
1244
1245
1246
async def kill(self) -> None:
    """Forcibly kill the process."""
    self._proc.kill()
    await self._proc.wait()

run(cmd, *, input=None, cwd=None, env=None, env_extra=None, env_remove=None, timeout=DEFAULT_TIMEOUT, kill_delay=DEFAULT_KILL_DELAY, check=True, encoding=DEFAULT_ENCODING, on_stdout=None, on_stderr=None, allowed_commands=None, blocked_commands=None)

Run a command synchronously and return the result.

Parameters:

Name Type Description Default
cmd str | Sequence[str]

Command as a string (auto-split) or sequence of arguments.

 required
input str | None

Text to send on stdin.

 None
cwd str | Path | None

Working directory for the subprocess.

 None
env dict[str, str] | None

Complete replacement environment (no inheritance).

 None
env_extra dict[str, str] | None

Extra variables to merge into the inherited environment.

 None
env_remove Sequence[str] | None

Variables to strip from the inherited environment.

 None
timeout float | None

Maximum seconds to wait. None means no timeout.

 DEFAULT_TIMEOUT
kill_delay float

Seconds to wait between SIGTERM and SIGKILL.

 DEFAULT_KILL_DELAY
check bool

If True, raise CommandFailedError on non-zero exit.

 True
encoding str

Text encoding for stdout/stderr.

 DEFAULT_ENCODING
on_stdout Callable[[str], None] | None

Per-line callback for stdout (output is still captured).

 None
on_stderr Callable[[str], None] | None

Per-line callback for stderr (output is still captured).

 None
allowed_commands Sequence[str] | None

If set, only these command names are permitted.

 None
blocked_commands Sequence[str] | None

If set, these command names are rejected.

 None

Returns:

Type Description
RunResult

A RunResult with captured output, exit code, and timing.

Raises:

Type Description
CommandNotFoundError

If the command binary is not found.
CommandFailedError

If check=True and the exit code is non-zero.
CommandTimeoutError

If execution exceeds timeout.
CommandBlockedError

If the command violates the policy.
ValueError

If the command is empty.
Source code in runner/runner.py 
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
def run(
    cmd: str | Sequence[str],
    *,
    input: str | None = None,  # noqa: A002
    cwd: str | Path | None = None,
    env: dict[str, str] | None = None,
    env_extra: dict[str, str] | None = None,
    env_remove: Sequence[str] | None = None,
    timeout: float | None = DEFAULT_TIMEOUT,
    kill_delay: float = DEFAULT_KILL_DELAY,
    check: bool = True,
    encoding: str = DEFAULT_ENCODING,
    on_stdout: Callable[[str], None] | None = None,
    on_stderr: Callable[[str], None] | None = None,
    allowed_commands: Sequence[str] | None = None,
    blocked_commands: Sequence[str] | None = None,
) -> RunResult:
    """Run a command synchronously and return the result.

    Args:
        cmd: Command as a string (auto-split) or sequence of arguments.
        input: Text to send on stdin.
        cwd: Working directory for the subprocess.
        env: Complete replacement environment (no inheritance).
        env_extra: Extra variables to merge into the inherited environment.
        env_remove: Variables to strip from the inherited environment.
        timeout: Maximum seconds to wait. None means no timeout.
        kill_delay: Seconds to wait between SIGTERM and SIGKILL.
        check: If True, raise CommandFailedError on non-zero exit.
        encoding: Text encoding for stdout/stderr.
        on_stdout: Per-line callback for stdout (output is still captured).
        on_stderr: Per-line callback for stderr (output is still captured).
        allowed_commands: If set, only these command names are permitted.
        blocked_commands: If set, these command names are rejected.

    Returns:
        A RunResult with captured output, exit code, and timing.

    Raises:
        CommandNotFoundError: If the command binary is not found.
        CommandFailedError: If check=True and the exit code is non-zero.
        CommandTimeoutError: If execution exceeds *timeout*.
        CommandBlockedError: If the command violates the policy.
        ValueError: If the command is empty.
    """
    # Phase 1-3: command normalization, policy, environment
    cmd_tuple = _parse_cmd(cmd)
    _check_command_policy(cmd_tuple[0], allowed_commands, blocked_commands)
    computed_env = _build_env(env, env_extra, env_remove)
    popen_kwargs = _popen_platform_kwargs()

    if which(cmd_tuple[0]) is None and not Path(cmd_tuple[0]).is_absolute():
        raise CommandNotFoundError(cmd_tuple[0])

    t0 = time.monotonic()

    # Phase 4: process startup
    try:
        proc = subprocess.Popen(
            cmd_tuple,
            stdin=subprocess.PIPE if input is not None else subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            cwd=cwd,
            env=computed_env,
            encoding=encoding,
            **popen_kwargs,
        )
    except FileNotFoundError as exc:
        raise CommandNotFoundError(cmd_tuple[0]) from exc

    pid = proc.pid

    # Phase 5+6: stdout/stderr collection + timeout
    has_callbacks = on_stdout is not None or on_stderr is not None
    try:
        if has_callbacks:
            stdout_text, stderr_text = _sync_collect_with_callbacks(
                proc, on_stdout, on_stderr, input, timeout, kill_delay, cmd_tuple
            )
        else:
            stdout_text, stderr_text = _sync_collect_simple(
                proc, input, timeout, kill_delay, cmd_tuple, encoding
            )
    except CommandTimeoutError:
        raise
    except Exception:
        if proc.poll() is None:
            _terminate_with_escalation(proc, kill_delay)
        raise

    duration = time.monotonic() - t0

    # Phase 7+8: result construction and exit-code check
    result = RunResult(
        command=cmd_tuple,
        returncode=proc.returncode,
        stdout=stdout_text,
        stderr=stderr_text,
        duration=duration,
        pid=pid,
    )

    if check and proc.returncode != 0:
        raise CommandFailedError(result)

    return result

run_async(cmd, *, input=None, cwd=None, env=None, env_extra=None, env_remove=None, timeout=DEFAULT_TIMEOUT, kill_delay=DEFAULT_KILL_DELAY, check=True, encoding=DEFAULT_ENCODING, on_stdout=None, on_stderr=None, allowed_commands=None, blocked_commands=None) async

Run a command asynchronously and return the result.

Async counterpart of :func:run. Uses asyncio.create_subprocess_exec internally.

Parameters:

Name Type Description Default
cmd str | Sequence[str]

Command as a string (auto-split) or sequence of arguments.

 required
input str | None

Text to send on stdin.

 None
cwd str | Path | None

Working directory for the subprocess.

 None
env dict[str, str] | None

Complete replacement environment (no inheritance).

 None
env_extra dict[str, str] | None

Extra variables to merge into the inherited environment.

 None
env_remove Sequence[str] | None

Variables to strip from the inherited environment.

 None
timeout float | None

Maximum seconds to wait. None means no timeout.

 DEFAULT_TIMEOUT
kill_delay float

Seconds to wait between SIGTERM and SIGKILL.

 DEFAULT_KILL_DELAY
check bool

If True, raise CommandFailedError on non-zero exit.

 True
encoding str

Text encoding for stdout/stderr.

 DEFAULT_ENCODING
on_stdout Callable[[str], None] | None

Per-line callback for stdout (output is still captured).

 None
on_stderr Callable[[str], None] | None

Per-line callback for stderr (output is still captured).

 None
allowed_commands Sequence[str] | None

If set, only these command names are permitted.

 None
blocked_commands Sequence[str] | None

If set, these command names are rejected.

 None

Returns:

Type Description
RunResult

A RunResult with captured output, exit code, and timing.

Raises:

Type Description
CommandNotFoundError

If the command binary is not found.
CommandFailedError

If check=True and the exit code is non-zero.
CommandTimeoutError

If execution exceeds timeout.
CommandBlockedError

If the command violates the policy.
ValueError

If the command is empty.
Source code in runner/runner.py 
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
async def run_async(
    cmd: str | Sequence[str],
    *,
    input: str | None = None,  # noqa: A002
    cwd: str | Path | None = None,
    env: dict[str, str] | None = None,
    env_extra: dict[str, str] | None = None,
    env_remove: Sequence[str] | None = None,
    timeout: float | None = DEFAULT_TIMEOUT,
    kill_delay: float = DEFAULT_KILL_DELAY,
    check: bool = True,
    encoding: str = DEFAULT_ENCODING,
    on_stdout: Callable[[str], None] | None = None,
    on_stderr: Callable[[str], None] | None = None,
    allowed_commands: Sequence[str] | None = None,
    blocked_commands: Sequence[str] | None = None,
) -> RunResult:
    """Run a command asynchronously and return the result.

    Async counterpart of :func:`run`. Uses
    ``asyncio.create_subprocess_exec`` internally.

    Args:
        cmd: Command as a string (auto-split) or sequence of arguments.
        input: Text to send on stdin.
        cwd: Working directory for the subprocess.
        env: Complete replacement environment (no inheritance).
        env_extra: Extra variables to merge into the inherited environment.
        env_remove: Variables to strip from the inherited environment.
        timeout: Maximum seconds to wait. None means no timeout.
        kill_delay: Seconds to wait between SIGTERM and SIGKILL.
        check: If True, raise CommandFailedError on non-zero exit.
        encoding: Text encoding for stdout/stderr.
        on_stdout: Per-line callback for stdout (output is still captured).
        on_stderr: Per-line callback for stderr (output is still captured).
        allowed_commands: If set, only these command names are permitted.
        blocked_commands: If set, these command names are rejected.

    Returns:
        A RunResult with captured output, exit code, and timing.

    Raises:
        CommandNotFoundError: If the command binary is not found.
        CommandFailedError: If check=True and the exit code is non-zero.
        CommandTimeoutError: If execution exceeds *timeout*.
        CommandBlockedError: If the command violates the policy.
        ValueError: If the command is empty.
    """
    # Phase 1-3: command normalization, policy, environment
    cmd_tuple = _parse_cmd(cmd)
    _check_command_policy(cmd_tuple[0], allowed_commands, blocked_commands)
    computed_env = _build_env(env, env_extra, env_remove)

    if which(cmd_tuple[0]) is None and not Path(cmd_tuple[0]).is_absolute():
        raise CommandNotFoundError(cmd_tuple[0])

    t0 = time.monotonic()

    # Phase 4: process startup
    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd_tuple,
            stdin=(
                asyncio.subprocess.PIPE
                if input is not None
                else asyncio.subprocess.DEVNULL
            ),
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=cwd,
            env=computed_env,
        )
    except FileNotFoundError as exc:
        raise CommandNotFoundError(cmd_tuple[0]) from exc

    pid = proc.pid
    input_bytes = input.encode(encoding) if input is not None else None

    # Phase 5+6: stdout/stderr collection + timeout
    has_callbacks = on_stdout is not None or on_stderr is not None
    try:
        if has_callbacks:
            stdout_text, stderr_text = await _async_collect_with_callbacks(
                proc,
                on_stdout,
                on_stderr,
                input_bytes,
                timeout,
                kill_delay,
                cmd_tuple,
                encoding,
            )
        else:
            stdout_text, stderr_text = await _async_collect_simple(
                proc,
                input_bytes,
                timeout,
                kill_delay,
                cmd_tuple,
                encoding,
            )
    except CommandTimeoutError:
        raise
    except Exception:
        if proc.returncode is None:
            await _async_terminate_with_escalation(proc, kill_delay)
        raise

    duration = time.monotonic() - t0

    # Phase 7+8: result construction and exit-code check
    result = RunResult(
        command=cmd_tuple,
        returncode=proc.returncode if proc.returncode is not None else -1,
        stdout=stdout_text,
        stderr=stderr_text,
        duration=duration,
        pid=pid,
    )

    if check and result.returncode != 0:
        raise CommandFailedError(result)

    return result

stream(cmd, *, input=None, cwd=None, env=None, env_extra=None, env_remove=None, timeout=None, kill_delay=DEFAULT_KILL_DELAY, encoding=DEFAULT_ENCODING, allowed_commands=None, blocked_commands=None)

Context manager for streaming subprocess output.

Yields a :class:StreamHandle that provides line iterators over stdout and stderr. The process is automatically cleaned up on context exit.

Parameters:

Name Type Description Default
cmd str | Sequence[str]

Command as a string (auto-split) or sequence of arguments.

 required
input str | None

Text to send on stdin.

 None
cwd str | Path | None

Working directory for the subprocess.

 None
env dict[str, str] | None

Complete replacement environment (no inheritance).

 None
env_extra dict[str, str] | None

Extra variables to merge into the inherited environment.

 None
env_remove Sequence[str] | None

Variables to strip from the inherited environment.

 None
timeout float | None

Maximum seconds for the process. None means no timeout.

 None
kill_delay float

Seconds to wait between SIGTERM and SIGKILL.

 DEFAULT_KILL_DELAY
encoding str

Text encoding for stdout/stderr.

 DEFAULT_ENCODING
allowed_commands Sequence[str] | None

If set, only these command names are permitted.

 None
blocked_commands Sequence[str] | None

If set, these command names are rejected.

 None

Yields:

Type Description
StreamHandle

A StreamHandle for reading process output.

Raises:

Type Description
CommandNotFoundError

If the command binary is not found.
CommandBlockedError

If the command violates the policy.
ValueError

If the command is empty.
Source code in runner/runner.py 
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
@contextmanager
def stream(
    cmd: str | Sequence[str],
    *,
    input: str | None = None,  # noqa: A002
    cwd: str | Path | None = None,
    env: dict[str, str] | None = None,
    env_extra: dict[str, str] | None = None,
    env_remove: Sequence[str] | None = None,
    timeout: float | None = None,
    kill_delay: float = DEFAULT_KILL_DELAY,
    encoding: str = DEFAULT_ENCODING,
    allowed_commands: Sequence[str] | None = None,
    blocked_commands: Sequence[str] | None = None,
) -> Iterator[StreamHandle]:
    """Context manager for streaming subprocess output.

    Yields a :class:`StreamHandle` that provides line iterators over
    stdout and stderr.  The process is automatically cleaned up on
    context exit.

    Args:
        cmd: Command as a string (auto-split) or sequence of arguments.
        input: Text to send on stdin.
        cwd: Working directory for the subprocess.
        env: Complete replacement environment (no inheritance).
        env_extra: Extra variables to merge into the inherited environment.
        env_remove: Variables to strip from the inherited environment.
        timeout: Maximum seconds for the process. None means no timeout.
        kill_delay: Seconds to wait between SIGTERM and SIGKILL.
        encoding: Text encoding for stdout/stderr.
        allowed_commands: If set, only these command names are permitted.
        blocked_commands: If set, these command names are rejected.

    Yields:
        A StreamHandle for reading process output.

    Raises:
        CommandNotFoundError: If the command binary is not found.
        CommandBlockedError: If the command violates the policy.
        ValueError: If the command is empty.
    """
    cmd_tuple = _parse_cmd(cmd)
    _check_command_policy(cmd_tuple[0], allowed_commands, blocked_commands)

    computed_env = _build_env(env, env_extra, env_remove)
    popen_kwargs = _popen_platform_kwargs()

    if which(cmd_tuple[0]) is None and not Path(cmd_tuple[0]).is_absolute():
        raise CommandNotFoundError(cmd_tuple[0])

    try:
        proc = subprocess.Popen(
            cmd_tuple,
            stdin=subprocess.PIPE if input is not None else subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            cwd=cwd,
            env=computed_env,
            encoding=encoding,
            **popen_kwargs,
        )
    except FileNotFoundError as exc:
        raise CommandNotFoundError(cmd_tuple[0]) from exc

    if input is not None and proc.stdin is not None:
        proc.stdin.write(input)
        proc.stdin.close()

    handle = StreamHandle(proc, encoding, timeout, kill_delay)
    try:
        yield handle
    finally:
        # Tier 1: must-succeed — process cleanup
        handle._cleanup()

stream_async(cmd, *, input=None, cwd=None, env=None, env_extra=None, env_remove=None, timeout=None, kill_delay=DEFAULT_KILL_DELAY, encoding=DEFAULT_ENCODING, allowed_commands=None, blocked_commands=None) async

Async context manager for streaming subprocess output.

Yields an :class:AsyncStreamHandle that provides async line iterators over stdout and stderr. The process is automatically cleaned up on context exit.

Parameters:

Name Type Description Default
cmd str | Sequence[str]

Command as a string (auto-split) or sequence of arguments.

 required
input str | None

Text to send on stdin.

 None
cwd str | Path | None

Working directory for the subprocess.

 None
env dict[str, str] | None

Complete replacement environment (no inheritance).

 None
env_extra dict[str, str] | None

Extra variables to merge into the inherited environment.

 None
env_remove Sequence[str] | None

Variables to strip from the inherited environment.

 None
timeout float | None

Maximum seconds for the process. None means no timeout.

 None
kill_delay float

Seconds to wait between SIGTERM and SIGKILL.

 DEFAULT_KILL_DELAY
encoding str

Text encoding for stdout/stderr.

 DEFAULT_ENCODING
allowed_commands Sequence[str] | None

If set, only these command names are permitted.

 None
blocked_commands Sequence[str] | None

If set, these command names are rejected.

 None

Yields:

Type Description
AsyncIterator[AsyncStreamHandle]

An AsyncStreamHandle for reading process output.

Raises:

Type Description
CommandNotFoundError

If the command binary is not found.
CommandBlockedError

If the command violates the policy.
ValueError

If the command is empty.
Source code in runner/runner.py 
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
@asynccontextmanager
async def stream_async(
    cmd: str | Sequence[str],
    *,
    input: str | None = None,  # noqa: A002
    cwd: str | Path | None = None,
    env: dict[str, str] | None = None,
    env_extra: dict[str, str] | None = None,
    env_remove: Sequence[str] | None = None,
    timeout: float | None = None,
    kill_delay: float = DEFAULT_KILL_DELAY,
    encoding: str = DEFAULT_ENCODING,
    allowed_commands: Sequence[str] | None = None,
    blocked_commands: Sequence[str] | None = None,
) -> AsyncIterator[AsyncStreamHandle]:
    """Async context manager for streaming subprocess output.

    Yields an :class:`AsyncStreamHandle` that provides async line
    iterators over stdout and stderr.  The process is automatically
    cleaned up on context exit.

    Args:
        cmd: Command as a string (auto-split) or sequence of arguments.
        input: Text to send on stdin.
        cwd: Working directory for the subprocess.
        env: Complete replacement environment (no inheritance).
        env_extra: Extra variables to merge into the inherited environment.
        env_remove: Variables to strip from the inherited environment.
        timeout: Maximum seconds for the process. None means no timeout.
        kill_delay: Seconds to wait between SIGTERM and SIGKILL.
        encoding: Text encoding for stdout/stderr.
        allowed_commands: If set, only these command names are permitted.
        blocked_commands: If set, these command names are rejected.

    Yields:
        An AsyncStreamHandle for reading process output.

    Raises:
        CommandNotFoundError: If the command binary is not found.
        CommandBlockedError: If the command violates the policy.
        ValueError: If the command is empty.
    """
    cmd_tuple = _parse_cmd(cmd)
    _check_command_policy(cmd_tuple[0], allowed_commands, blocked_commands)

    computed_env = _build_env(env, env_extra, env_remove)

    if which(cmd_tuple[0]) is None and not Path(cmd_tuple[0]).is_absolute():
        raise CommandNotFoundError(cmd_tuple[0])

    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd_tuple,
            stdin=(
                asyncio.subprocess.PIPE
                if input is not None
                else asyncio.subprocess.DEVNULL
            ),
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=cwd,
            env=computed_env,
        )
    except FileNotFoundError as exc:
        raise CommandNotFoundError(cmd_tuple[0]) from exc

    if input is not None and proc.stdin is not None:
        proc.stdin.write(input.encode(encoding))
        await proc.stdin.drain()
        proc.stdin.close()
        await proc.stdin.wait_closed()

    handle = AsyncStreamHandle(proc, encoding, timeout, kill_delay)
    try:
        yield handle
    finally:
        # Tier 1: must-succeed — async process cleanup
        await handle._cleanup()

shell_split(s)

Split a shell command string into a list of arguments.

Uses :func:shlex.split with POSIX mode on Unix and non-POSIX on Windows.

Parameters:

Name Type Description Default
s str

Shell command string, e.g. 'git commit -m "hello world"'.

 required

Returns:

Type Description
list[str]

List of arguments.

Raises:

Type Description
ValueError

On unterminated quotes or other parse errors.
Source code in runner/runner.py 
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
def shell_split(s: str) -> list[str]:
    """Split a shell command string into a list of arguments.

    Uses :func:`shlex.split` with POSIX mode on Unix and non-POSIX on
    Windows.

    Args:
        s: Shell command string, e.g. ``'git commit -m "hello world"'``.

    Returns:
        List of arguments.

    Raises:
        ValueError: On unterminated quotes or other parse errors.
    """
    return shlex.split(s, posix=not _IS_WINDOWS)

shell_quote(*args)

Quote arguments for safe shell interpolation.

On Unix, uses :func:shlex.quote. On Windows, uses cmd.exe-safe quoting with double-quote escaping.

Parameters:

Name Type Description Default
*args str

Individual arguments to quote.

 ()

Returns:

Type Description
str

Space-joined quoted string.
Source code in runner/runner.py 
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
def shell_quote(*args: str) -> str:
    """Quote arguments for safe shell interpolation.

    On Unix, uses :func:`shlex.quote`.  On Windows, uses ``cmd.exe``-safe
    quoting with double-quote escaping.

    Args:
        *args: Individual arguments to quote.

    Returns:
        Space-joined quoted string.
    """
    if _IS_WINDOWS:
        return " ".join(_win_quote(a) for a in args)
    return " ".join(shlex.quote(a) for a in args)

which(name)

Locate a command on the system PATH.

Cross-platform wrapper around :func:shutil.which.

Binary lookup order (Pattern 2 convention): 1. Exact path — if name is absolute, return it directly if it exists. 2. PATH search — delegates to :func:shutil.which, which walks os.environ["PATH"] entries in order, respecting PATHEXT on Windows.

Parameters:

Name Type Description Default
name str

Command name (e.g. "git").

 required

Returns:

Type Description
str | None

Absolute path to the binary, or None if not found.
Source code in runner/runner.py 
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
def which(name: str) -> str | None:
    """Locate a command on the system PATH.

    Cross-platform wrapper around :func:`shutil.which`.

    Binary lookup order (Pattern 2 convention):
      1. Exact path — if *name* is absolute, return it directly if it exists.
      2. PATH search — delegates to :func:`shutil.which`, which walks
         ``os.environ["PATH"]`` entries in order, respecting PATHEXT on
         Windows.

    Args:
        name: Command name (e.g. ``"git"``).

    Returns:
        Absolute path to the binary, or ``None`` if not found.
    """
    return shutil.which(name)