Merge pull request #465 from goodboy/trio_033_upgrade

Add `supervise_run_process()` to `trionics`
wkt/start_or_cancel_tests_474
Bd 2026-06-24 23:30:18 -04:00 committed by GitHub
commit d95cd94e20
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 901 additions and 11 deletions

View File

@ -0,0 +1,146 @@
---
model: claude-opus-4-7[1m]
service: claude
session: trio-0.33-subproc-supervisor-retroactive
timestamp: 2026-06-01T23:14:29Z
git_ref: 0e3e008b
scope: code
substantive: true
raw_file: 20260601T231429Z_0e3e008b_prompt_io.raw.md
---
## Prompt
**RETROACTIVE LOG** — original session prompts not
preserved; reconstructed from the staged work product.
The work designs a `trio.Nursery.start()`-style wrapper
around `trio.run_process()` for SC-friendly subprocess
supervision. From the resulting code shape, the
prompting intent was:
1. Surface rc!=0 `CalledProcessError` DETERMINISTICALLY,
without the nursery-eg-wrapping that complicates
`collapse_eg()` usage and races the relay reader on
trio's `check=True`-driven cancel cascade.
2. ALWAYS isolate the parent controlling-tty so a
spawned child can't emit terminal control-seqs onto
the launching tty (clobbering scrollback). Default
`stdin=DEVNULL`; default `stdout=DEVNULL` unless
explicitly relayed/overridden; distinguish "caller
passed nothing" from "caller passed `None` for
inherit".
3. Optional live per-line relay of child std-streams to
the `tractor` log — STREAMED (not
buffered-until-exit) so long-lived daemon output is
visible during the run. Pick a custom log level that
shows at usual `info`/`devx` console levels but is
separately filterable.
4. Concurrent pipe-drain reader MANDATORY when piping
without `capture_*` — without it the child blocks on
`write()` once the OS pipe buffer fills (~64KiB),
causing deadlocks on output bursts.
5. Non-blocking `tn.start()` semantics: hand the live
`trio.Process` to the parent immediately;
supervise/relay run to completion in the supervisor
coro.
6. Hermetic `trio`-only unit tests (no actor-runtime)
covering each of: per-line relay, tty isolation,
no-deadlock on >64KiB unnewlined output, CPE
rebuild w/ stderr relay, CPE rebuild on the silent
drain+capture path.
## Response summary
Adds `tractor/trionics/_subproc.py` (296 LOC) +
`tests/trionics/test_subproc.py` (230 LOC) + a
re-export in `tractor/trionics/__init__.py`.
**`supervise_run_process()`** (public, re-exported)
- `check=False` is forced to `trio.run_process`; the
rc-check runs in the supervisor coro AFTER `own_tn`
unwinds (both the child AND the relay readers have
hit EOF + fully drained). A BARE
`subprocess.CalledProcessError` is rebuilt + raised
from there, with `.stderr` bytes passed in the
constructor AND attached as an `add_note()`'d
`|_.stderr:` block for legible teardown logs.
- `stdin=DEVNULL` always. `stdout` default chosen via a
`_UNSET` sentinel: `relay_stdout=True` → PIPE,
explicit `stdout=...` → as given, else `DEVNULL`.
`stderr` defaults to PIPE whenever we relay OR need
the CPE note (when `check=True`), else `DEVNULL`.
- `relay_level='io'` (custom level 21; sorts just
above stdlib `INFO`=20 so it shows at usual
`info`/`devx` levels and stays separately
filterable). `runtime`=15 would silently filter at
default levels, so it's rejected as a default.
- `task_status.started(trio_proc)` delivers the live
process immediately. The internal `own_tn`
supervises `trio.run_process` + any relay readers to
completion.
- `**run_process_kwargs` forward verbatim;
`stdin/stdout/stderr/check` are MANAGED keys
(override on conflict).
- Crash-handling deliberately NOT baked in — compose
`maybe_open_crash_handler()` on top at the call-site.
**`_relay_stream_lines()`** (internal helper)
- Three modes (combinable): `emit`-only (live per-line
relay), `accum`-only (silent drain+capture for a CPE
note), or both (live relay AND capture).
- Per-line split handles cross-chunk residuals via a
rolling `residual` bytes buffer; flushes any trailing
un-newline-term'd line at EOF.
- `async with stream:` ensures aclose at EOF/cancel
(mirrors trio's internal `_subprocess` drain idiom).
**`_add_stderr_note()`** (internal helper)
- `add_note()`s a `textwrap.indent(...)`'d
`|_.stderr:` block onto a `CalledProcessError` for
teardown logs.
**Tests** (5 hermetic, trio-only) — `_capture_relay`
fixture monkeypatches `_subproc.log.<level>` to a list:
- `test_stdout_relayed_per_line`: per-line stdout
relay carries each `line=N` to the records.
- `test_parent_tty_isolated`: `readlink /proc/self/fd/0`
and `fd/1` from the child show `pipe:` (fd1) +
`/dev/null` (fd0); NO `/dev/pts/*`.
- `test_no_deadlock_on_big_unnewlined_output`: 200KiB
of `x` with no newlines completes inside
`fail_after(2)` — exercises the concurrent drain.
- `test_stderr_relay_and_cpe_rebuild`: rc=3 with
`relay_stderr=True` raises bare CPE
(via `collapse_eg()`) with `b'boom' in cpe.stderr`,
the note attached, AND per-line live relay.
- `test_nonrelay_cpe_note`: rc=7 with no relay still
produces CPE with `.stderr` + note via the silent
drain+capture path.
## Files changed
- `tractor/trionics/_subproc.py` — NEW. Public
`supervise_run_process()` + helpers
`_relay_stream_lines()` / `_add_stderr_note()` + the
`_UNSET` sentinel.
- `tests/trionics/test_subproc.py` — NEW. 5 hermetic
trio-only tests + `_capture_relay` monkeypatch
fixture.
- `tractor/trionics/__init__.py` — re-export
`supervise_run_process`.
## Human edits
**RETROACTIVE**: this log is being written from the
staged diff, not from a live session. The code as
staged is the canonical artifact; any human edits the
user made during the originating design session are
already integrated and cannot be separated post-hoc.
The `.raw.md` sibling is a diff-pointer placeholder,
NOT a pre-edit transcript.
Future prompt-io entries for in-flight work should be
written DURING the design session per the skill
contract so the pre-edit `.raw.md` captures the
unedited model output for genuine provenance.

View File

@ -0,0 +1,106 @@
---
model: claude-opus-4-7[1m]
service: claude
timestamp: 2026-06-01T23:14:29Z
git_ref: 0e3e008b
diff_cmd: git diff HEAD~1..HEAD
---
# RETROACTIVE — original model output not preserved
This `.raw.md` would normally contain the verbatim
pre-human-edit response from the design session that
produced the staged `_subproc.py` module + tests. That
session's transcript is not available, so this file
serves as a diff-pointer placeholder + transparency
note.
## Authoritative artifact
The committed code IS the artifact of record. Once the
companion commit lands, the unified diff is:
> `git diff HEAD~1..HEAD -- tractor/trionics/_subproc.py`
> `git diff HEAD~1..HEAD -- tests/trionics/test_subproc.py`
> `git diff HEAD~1..HEAD -- tractor/trionics/__init__.py`
Before committing, substitute `--cached` for the
pre-commit form.
## What is NOT here
Because this is retroactive:
- No verbatim chain-of-thought / discussion prose from
the design session.
- No rejected alternatives the model considered before
arriving at the final shape (e.g. whether the
rc-check should live inside `own_tn` vs after it; the
`_UNSET` sentinel vs a `None`-means-DEVNULL
convention; `io` vs `info` as the default relay
level).
- No pre-edit code blocks as the model first emitted
them, separable from any user cleanup applied before
the diff was staged.
## Inferred design choices visible in the final code
(Documented here because they're the kind of decision
detail an unedited raw transcript would have captured.)
1. **Post-drain rc-check in the supervisor coro body,
AFTER `own_tn.__aexit__`.** Placing the
`CalledProcessError` raise here (not inside
`own_tn`) means the EG-unwrap happens at the OUTER
`tn.start()` boundary — callers do `collapse_eg()`
if they want bare. Doing the raise INSIDE `own_tn`
would cancel the still-draining relay reader
mid-flight and lose stderr lines.
2. **`_UNSET` sentinel for `stdout`.** A plain default
of `None` couldn't distinguish "use the safe
`DEVNULL` default" from "caller explicitly passed
`None` (inherit, presumably knowingly)". The
sentinel keeps the SAFE default while letting power
users opt into inherit.
3. **`relay_level='io'` (custom level 21).** Chosen to
sort just above stdlib `INFO`=20 so a default
`--ll info` shows the relay, but it remains a
distinct level so users can filter
`tractor.trionics:io` separately. Picking
`runtime`=15 would have made the relay invisible at
default verbosity (a footgun for daemon supervisors
whose whole point is "I want to see this output").
4. **Reader is MANDATORY, not opt-in cosmetic.** With
`stdout=PIPE` / `stderr=PIPE` we OWN the drain
responsibility — there's no `trio.capture_*` running
under the hood here. The ~64KiB OS pipe buffer
means a child writing more than that without us
reading hangs at `write()` — a deadlock that won't
show up in small-output tests, which is why the
200KiB-no-newline test is in the suite.
5. **`task_status.started(trio_proc)` BEFORE the
`own_tn` exits.** Without this, `tn.start()` would
block until the child exits — losing the "start a
long-lived daemon and continue with parent work"
use case. With it, the parent gets the live process
handle immediately and the supervise+relay tasks
run in the supervisor coro until the child exits.
6. **`__notes__` via `add_note()` for the CPE
`.stderr`.** The `.stderr` attribute is what
`subprocess` callers expect; the `add_note()` is
what trio's exception-rendering shows. Both wired so
programmatic AND human consumers see the stderr at
teardown.
## Honesty statement
This file's content is RECONSTRUCTED from the staged
code, not extracted from a verbatim model transcript.
The prompt-io skill's intent is for the `.raw.md` to
be a pre-edit fossil; that's not possible here. Future
work should write the prompt-io entry DURING the
design session.

View File

@ -5,6 +5,7 @@
from __future__ import annotations
import platform
import os
import re
import signal
import time
from typing import (
@ -294,6 +295,26 @@ def expect(
PROMPT = r"\(Pdb\+\)"
# Strip terminal color / ANSI-VT100 escape sequences so
# substring matching against REPL + traceback output stays
# robust to color leakage — Python 3.13's colored tracebacks,
# `pdbp`'s pygments highlighting, etc. — even when
# `PYTHON_COLORS=0` (set in the `spawn` fixture) isn't honored
# by every renderer in the spawned subproc.
# Regex per https://stackoverflow.com/a/14693789
_ansi_re: re.Pattern = re.compile(
r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])'
)
def ansi_strip(text: str) -> str:
'''
Remove ANSI/VT100 escape sequences from `text`.
'''
return _ansi_re.sub('', text)
def in_prompt_msg(
child: SpawnBase,
parts: list[str],
@ -313,7 +334,7 @@ def in_prompt_msg(
'''
__tracebackhide__: bool = False
before: str = str(child.before.decode())
before: str = ansi_strip(str(child.before.decode()))
for part in parts:
if part not in before:
if pause_on_false:
@ -333,9 +354,9 @@ def in_prompt_msg(
return True
# TODO: todo support terminal color-chars stripping so we can match
# against call stack frame output from the the 'll' command the like!
# -[ ] SO answer for stipping ANSI codes: https://stackoverflow.com/a/14693789
# NB: color-char stripping (so we can match against call-stack
# frame output from the `ll` command and the like) is handled by
# `ansi_strip()` applied inside `in_prompt_msg()` + below.
def assert_before(
child: SpawnBase,
patts: list[str],
@ -356,7 +377,7 @@ def assert_before(
err_on_false=True,
**kwargs
)
before: str = str(child.before.decode())
before: str = ansi_strip(str(child.before.decode()))
return before

View File

@ -1186,7 +1186,12 @@ def test_shield_pause(
"('cancelled_before_pause'", # actor name
_repl_fail_msg,
"trio.Cancelled",
"raise Cancelled._create()",
# trio >=0.30 raises via a multi-line
# `raise Cancelled._create(source=.., reason=..,
# source_task=..)` (cancel-reason metadata), so
# match the open-paren form only, NOT the legacy
# bare `()`.
"raise Cancelled._create(",
# we should be handling a taskc inside
# the first `.port_mortem()` sin-shield!
@ -1204,7 +1209,12 @@ def test_shield_pause(
"('root'", # actor name
_repl_fail_msg,
"trio.Cancelled",
"raise Cancelled._create()",
# trio >=0.30 raises via a multi-line
# `raise Cancelled._create(source=.., reason=..,
# source_task=..)` (cancel-reason metadata), so
# match the open-paren form only, NOT the legacy
# bare `()`.
"raise Cancelled._create(",
# handling a taskc inside the first unshielded
# `.port_mortem()`.

View File

@ -263,9 +263,18 @@ def test_simple_context(
except error_parent:
pass
except BaseExceptionGroup as beg:
# XXX: on windows it seems we may have to expect the group error
# XXX: on windows — and under `trio>=0.33`'s strict
# exception-groups, where a lone `error_parent` now
# arrives wrapped in a 1-exc `ExceptionGroup` rather than
# collapsed to the bare exc — accept either the
# `error_parent` nested in the group OR a cancel-only
# (multi-cancelled) group.
from tractor.trionics import is_multi_cancelled
assert is_multi_cancelled(beg)
assert (
is_multi_cancelled(beg)
or
beg.subgroup(error_parent) is not None
)
else:
trio.run(main)

View File

@ -150,7 +150,12 @@ async def test_most_beautiful_word(
The main ``tractor`` routine.
'''
with trio.fail_after(1):
# actor spawn + IPC round-trip is comfortably sub-second on a
# warm box, but slow/noisy CI runners (esp. macOS) blow a flat
# 1s deadline. Scale for CI/CPU-throttle headroom — `== 1s`
# locally where `cpu_scaling_factor()` is `1.0`.
from .conftest import cpu_scaling_factor
with trio.fail_after(1 * cpu_scaling_factor()):
async with tractor.open_nursery(
debug_mode=debug_mode,
) as an:

View File

@ -0,0 +1,235 @@
'''
Unit tests for `tractor.trionics.supervise_run_process` (in
`tractor.trionics._subproc`) and its per-line std-stream relay.
Hermetic `trio`-only coverage (no actor-runtime needed):
- per-line stdout relay -> `log.io`
- parent controlling-tty isolation (child fd1 is a pipe, fd0
`/dev/null` never the parent `/dev/pts/*`)
- mandatory concurrent pipe-drain (no deadlock on >64KiB
no-newline output)
- live stderr relay + `CalledProcessError` rebuild (rc!=0 note)
- legacy capture-stderr CPE note path
'''
from functools import partial
import platform
import subprocess
import pytest
import trio
from tractor.trionics import (
_subproc,
collapse_eg,
supervise_run_process,
)
def _capture_relay(monkeypatch, level: str = 'io') -> list[str]:
'''
Redirect `_subproc.log.<level>` (the relay's emit method —
`io` by default, see `supervise_run_process(relay_level=...)`)
into a list so tests can assert on the relayed lines.
'''
records: list[str] = []
monkeypatch.setattr(
_subproc.log,
level,
lambda msg, *a, **k: records.append(msg),
)
return records
def test_stdout_relayed_per_line(monkeypatch):
records = _capture_relay(monkeypatch)
cmd = [
'sh', '-c',
'for i in 1 2 3; do echo line=$i; done',
]
async def main():
async with trio.open_nursery() as tn:
await tn.start(
partial(
supervise_run_process,
cmd,
label='t-out',
relay_stdout=True,
)
)
trio.run(main)
out_lines = [r for r in records if '[t-out:out]' in r]
assert any('line=1' in r for r in out_lines)
assert any('line=2' in r for r in out_lines)
assert any('line=3' in r for r in out_lines)
@pytest.mark.skipif(
platform.system() != 'Linux',
reason='reads `/proc/self/fd` — Linux-only',
)
def test_parent_tty_isolated(monkeypatch):
records = _capture_relay(monkeypatch)
cmd = [
'sh', '-c',
'readlink /proc/self/fd/0; readlink /proc/self/fd/1',
]
async def main():
async with trio.open_nursery() as tn:
await tn.start(
partial(
supervise_run_process,
cmd,
label='t-tty',
relay_stdout=True,
)
)
trio.run(main)
relayed = '\n'.join(records)
# fd1 (stdout) must be OUR pipe, never a controlling tty.
assert 'pipe:' in relayed
assert '/dev/pts/' not in relayed
# fd0 (stdin) is pinned to DEVNULL.
assert '/dev/null' in relayed
def test_no_deadlock_on_big_unnewlined_output(monkeypatch):
'''
>64KiB of output with NO newline: only completes because the
relay reader concurrently drains the pipe (else the child
blocks on `write()` when the OS pipe buffer fills).
'''
records = _capture_relay(monkeypatch)
cmd = [
'sh', '-c',
'head -c 200000 /dev/zero | tr "\\0" x',
]
async def main():
# generous vs the ~ms real runtime, but bounded so a
# genuine pipe-fill deadlock fails fast.
with trio.fail_after(2):
async with trio.open_nursery() as tn:
await tn.start(
partial(
supervise_run_process,
cmd,
label='t-big',
relay_stdout=True,
)
)
trio.run(main)
big = ''.join(
r.split('] ', 1)[-1]
for r in records
if '[t-big:out]' in r
)
assert len(big) == 200_000
def test_stderr_relay_and_cpe_rebuild(monkeypatch):
'''
`relay_stderr=True` PIPEs stderr ourselves (mutually
exclusive with trio's `capture_stderr`), so on rc!=0 the
wrapper rebuilds a `CalledProcessError` from the live
accumulator and `.add_note()`s its `.stderr` AND the
stderr is relayed per-line live.
'''
records = _capture_relay(monkeypatch)
cmd = [
'sh', '-c',
'echo boom 1>&2; exit 3',
]
async def main():
# `collapse_eg()` unwraps the parent-nursery's single-exc
# eg so the bare CPE bubbles straight out (mirrors real
# caller usage).
async with (
collapse_eg(),
trio.open_nursery() as tn,
):
await tn.start(
partial(
supervise_run_process,
cmd,
label='t-err',
relay_stderr=True,
check=True,
)
)
with pytest.raises(subprocess.CalledProcessError) as ei:
trio.run(main)
cpe = ei.value
assert cpe.returncode == 3
# rebuilt `.stderr` (trio did NOT capture since we PIPE'd it).
assert b'boom' in (cpe.stderr or b'')
# note attached for legible teardown reporting.
assert any(
'boom' in n
for n in getattr(cpe, '__notes__', [])
)
# AND it was relayed live per-line.
assert any(
'[t-err:err]' in r and 'boom' in r
for r in records
)
def test_nonrelay_cpe_note(monkeypatch):
'''
No live relay: stderr is silently drained + captured (NOT
emitted), and on rc!=0 the wrapper rebuilds the
`CalledProcessError` from that accumulator with a `.stderr`
note same deterministic post-drain path as the relay case.
'''
cmd = [
'sh', '-c',
'echo nope 1>&2; exit 7',
]
async def main():
async with (
collapse_eg(),
trio.open_nursery() as tn,
):
await tn.start(
partial(
supervise_run_process,
cmd,
label='t-legacy',
check=True,
# relay_* default False -> silent
# drain+capture for the CPE note.
)
)
with pytest.raises(subprocess.CalledProcessError) as ei:
trio.run(main)
cpe = ei.value
assert cpe.returncode == 7
assert b'nope' in (cpe.stderr or b'')
assert any(
'nope' in n
for n in getattr(cpe, '__notes__', [])
)

View File

@ -398,7 +398,7 @@ async def handle_stream_from_peer(
uid,
None,
)
if event:
if event is not None:
con_status_steps += (
' -> Waking subactor spawn waiters: '
f'{event.statistics().tasks_waiting}\n'

View File

@ -38,3 +38,6 @@ from ._taskc import (
maybe_raise_from_masking_exc as maybe_raise_from_masking_exc,
start_or_cancel as start_or_cancel,
)
from ._subproc import (
supervise_run_process as supervise_run_process,
)

View File

@ -0,0 +1,355 @@
# tractor: distributed structured concurrency.
# Copyright 2018-eternity Tyler Goodlet.
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
SC-friendly `trio.run_process()` supervision: a `tn.start()`
style wrapper which surfaces rc!=0 errors deterministically and
(optionally) live-relays the child's std-streams to the `tractor`
log.
'''
from __future__ import annotations
from functools import partial
import subprocess
import textwrap
from typing import (
Callable,
)
import trio
from ..log import get_logger
log = get_logger()
# sentinel so `supervise_run_process(stdout=...)` can tell
# "caller passed nothing" (-> tty-safe `DEVNULL` default) from
# an explicit `stdout=None` (inherit) override.
_UNSET = object()
# cap the stderr bytes retained for a rc!=0 `CalledProcessError`
# note: a long-lived (`check=True`) child could otherwise buffer
# its ENTIRE stderr history in parent mem — keep only the TAIL
# (the bytes nearest the failure, which are what you want to see).
_STDERR_NOTE_TAIL_BYTES: int = 16 * 1024
def _decode_line(raw: bytes) -> str:
'''
Decode one relayed line: replace undecodable bytes and trim a
trailing carriage-return (so CRLF-term'd lines read clean).
'''
return raw.decode(errors='replace').rstrip('\r')
def _add_stderr_note(
cpe: subprocess.CalledProcessError,
stderr_bytes: bytes,
) -> None:
'''
Attach an indented `|_.stderr:` note to a
`CalledProcessError` for legible rc!=0 reporting at
teardown.
'''
stderr_str: str = stderr_bytes.decode(errors='replace')
cpe.add_note(
f'|_.stderr:\n'
f'{textwrap.indent(stderr_str, prefix=" "*3)}'
)
async def _relay_stream_lines(
stream: trio.abc.ReceiveStream,
*,
emit: Callable[[str], None]|None = None,
tag: str = '',
accum: bytearray|None = None,
accum_cap: int|None = None,
) -> None:
'''
Concurrently drain a child subproc's `stdout`/`stderr`
PIPE; relay each COMPLETE line to `emit` (a bound
`log.<level>` method) prefixed with `tag` (e.g.
`f'{label}:out'`) and/or append raw bytes to `accum`.
This reader is MANDATORY whenever a bare
`stdout=`/`stderr=PIPE` is used WITHOUT `trio`'s
`capture_*` (which would spawn trio's own internal drain
task): nothing else drains the OS pipe, so once its kernel
buffer (~64KiB) fills the child blocks on `write()` ->
deadlock.
Modes (combine freely):
- `emit`-only: live per-line relay (e.g. `relay_stdout`).
- `accum`-only: silent drain + capture (e.g. stderr kept
for a `CalledProcessError` note WITHOUT relaying it).
- both: relay AND capture (e.g. `relay_stderr` with `check=True`).
'''
# NOTE, mirrors `trio._subprocess`'s internal
# `async with stream: async for ...` drain idiom — except
# here we EMIT per-line (and/or accumulate) instead of
# only accumulating.
residual: bytes = b''
async with stream: # aclose at EOF/cancel
async for chunk in stream: # ends at child-exit EOF
if accum is not None:
accum += chunk
# bound retained bytes to the TAIL (see caller +
# `_STDERR_NOTE_TAIL_BYTES`) so a chatty long-lived
# child can't grow parent mem without limit.
if (
accum_cap is not None
and
len(accum) > accum_cap
):
del accum[:len(accum) - accum_cap]
if emit is None:
continue # drain(+accum)-only
buf: bytes = residual + chunk
*lines, residual = buf.split(b'\n')
for raw in lines:
emit(f'[{tag}] {_decode_line(raw)}')
# flush any trailing partial (un-newline-term'd) line @ EOF
if (
emit is not None
and
residual
):
emit(f'[{tag}] {_decode_line(residual)}')
async def supervise_run_process(
cmd: list[str]|str,
*,
check: bool = True,
label: str|None = None,
# per-line `log.*` relay of the child's std-streams
# (tty-safe, capture-safe, STREAMED — not
# buffered-until-exit, so it suits long-lived daemons).
relay_stdout: bool = False,
relay_stderr: bool = False,
# default `io` (our custom level, value 21): the relay
# exists to make windowless-spawn output VISIBLE, and
# `IO`(21) sorts just ABOVE `INFO`(20) so it shows at the
# usual `info`/`devx` console levels (a `runtime`(15) relay
# would be silently filtered) while staying distinctly
# labelled + separately filterable.
relay_level: str = 'io',
# non-relay `stdout` override; defaults (via `_UNSET`) to
# `DEVNULL` so we NEVER inherit (+ thus can't clobber) the
# parent controlling-tty.
stdout: int|None = _UNSET,
task_status: trio.TaskStatus[
trio.Process
] = trio.TASK_STATUS_IGNORED,
# any other `trio.run_process()` kwarg (env, shell, cwd,
# start_new_session, executable, ...) forwarded verbatim;
# our MANAGED keys (stdin/stdout/stderr/check) are set
# below and WIN on conflict.
**run_process_kwargs,
) -> None:
'''
A `trio.Nursery.start()`-style `trio.run_process()`
wrapper which,
- surfaces a rc!=0 `subprocess.CalledProcessError`
DETERMINISTICALLY: we pass `check=False` to `trio` and
do our OWN post-drain rc-check, (re)building + raising the
CPE (with a `.stderr` note) from this coro's body AFTER the
child exits so it's never wrapped by `trio.run_process`'s
INTERNAL nursery nor race-cancelled mid-drain. It IS still
raised as a task into the *parent* `tn`, so like any
`tn.start()`-task raise it surfaces `ExceptionGroup`-
wrapped under `trio>=0.33`; unwrap via `except*` or
`trionics.collapse_eg()`.
- ALWAYS isolates the parent controlling-tty
(`stdin=DEVNULL`, and `stdout=DEVNULL` unless
relayed/overridden) so a spawned program can't emit
terminal control-seqs onto the launching tty (which
would clobber its scrollback).
- optionally live-relays `stdout`/`stderr` per-line to
`log.<relay_level>` via concurrent reader tasks (see
`_relay_stream_lines`).
Delivers the live `trio.Process` via
`task_status.started()` then SUPERVISES it (the
`run_process` bg task + any relay readers) to completion
in this coro i.e. the parent `tn.start()` returns
immediately/non-blocking.
NOTE: any crash-handling / `repl_fixture` layer is
intentionally NOT baked in here compose it ON TOP at the
call-site, e.g.
async with maybe_open_crash_handler():
await tn.start(
partial(supervise_run_process, cmd, ...),
)
'''
# resolve the relay emit-method ONLY when actually relaying so
# a bad/typo'd `relay_level` can't crash a NON-relay call (and
# we don't bind an unused method on the common silent path).
emit: Callable[[str], None]|None = (
getattr(log, relay_level)
if (relay_stdout or relay_stderr)
else None
)
tag: str = (
label
or
(cmd if isinstance(cmd, str) else ' '.join(cmd))
)
# forward any extra `trio.run_process` kwargs verbatim;
# MANAGED keys below override on conflict.
rp_kwargs: dict = dict(run_process_kwargs)
# XXX reject `trio`'s `capture_stdout`/`capture_stderr`: they
# ALIAS our MANAGED `stdout`/`stderr` keys, so forwarding them
# makes `trio.run_process` raise an opaque
# `ValueError("can't specify both ...")`. Fail LOUD + early
# with a pointer to the supported knobs instead.
for _cap_kw in ('capture_stdout', 'capture_stderr'):
if _cap_kw in rp_kwargs:
raise ValueError(
f'{_cap_kw!r} is unsupported here; use '
f'`relay_stdout=`/`relay_stderr=` for a live relay, '
f'or the `stdout=` override / default `check=True` '
f'stderr-capture instead.'
)
# XXX ALWAYS isolate the controlling-tty's stdin.
rp_kwargs['stdin'] = subprocess.DEVNULL
# stdout: relay -> our own PIPE (drained by the reader
# below); else an explicit override; else tty-safe
# `DEVNULL`.
if relay_stdout:
rp_kwargs['stdout'] = subprocess.PIPE
elif stdout is not _UNSET:
# XXX a bare `stdout=PIPE` override has NO drain reader
# (only `relay_stdout` spins one up), so it would deadlock
# once the ~64KiB OS pipe buffer fills — reject it.
if stdout is subprocess.PIPE:
raise ValueError(
'Use `relay_stdout=True` to PIPE *and* drain '
'stdout; a bare `stdout=subprocess.PIPE` override '
'has no reader and will deadlock once the ~64KiB '
'pipe buffer fills.'
)
rp_kwargs['stdout'] = stdout
else:
rp_kwargs['stdout'] = subprocess.DEVNULL
# stderr: PIPE (+ our reader) when we either RELAY it OR
# need it captured for a rc!=0 CPE note; else tty-safe
# `DEVNULL`. We accumulate ONLY when `check` (the note is
# the only consumer).
#
# XXX we ALWAYS pass `check=False` to `trio` and do our
# OWN deterministic post-drain rc-check (below) so `trio`
# never raises a nursery-eg-wrapped CPE — no `collapse_eg`
# workaround, no reader race-cancel.
want_stderr_pipe: bool = relay_stderr or check
stderr_accum: bytearray|None = bytearray() if check else None
rp_kwargs['check'] = False
rp_kwargs['stderr'] = (
subprocess.PIPE if want_stderr_pipe
else subprocess.DEVNULL
)
async with trio.open_nursery() as own_tn:
trio_proc: trio.Process = await own_tn.start(
partial(
trio.run_process,
cmd,
**rp_kwargs,
)
)
# spin up the concurrent pipe-drain relay reader(s) —
# see `_relay_stream_lines` for why these are mandatory
# (not cosmetic) when piping without `capture_*`.
if relay_stdout:
own_tn.start_soon(
partial(
_relay_stream_lines,
trio_proc.stdout,
emit=emit,
tag=f'{tag}:out',
)
)
if want_stderr_pipe:
own_tn.start_soon(
partial(
_relay_stream_lines,
trio_proc.stderr,
# relay live only if asked; else silent
# drain+capture for the CPE note.
emit=emit if relay_stderr else None,
tag=f'{tag}:err',
accum=stderr_accum,
accum_cap=_STDERR_NOTE_TAIL_BYTES,
)
)
# hand the live proc up to the parent WITHOUT blocking
# on the bg supervise/relay tasks (keeps non-blocking
# `tn.start()` semantics).
task_status.started(trio_proc)
# ===== deterministic post-drain rc-check (BOTH paths) =====
# `own_tn` only unwinds once `run_process` AND the relay
# reader(s) have hit EOF + FULLY drained — so `stderr_accum`
# is COMPLETE here (no race vs an early CPE-cancel). Rebuild
# + raise a BARE `CalledProcessError` (the parent `tn` will
# eg-wrap it like any task-raise; callers `collapse_eg()` if
# they want it bare).
if (
check
and
trio_proc.returncode
):
stderr_bytes: bytes = (
bytes(stderr_accum)
if stderr_accum is not None
else b''
)
# NOTE: stdout is NOT captured (DEVNULL unless overridden)
# so the CPE carries only (tail-bounded) stderr; a cmd that
# logs failures to *stdout* should `relay_stdout=True` to
# surface them in the note.
cpe = subprocess.CalledProcessError(
returncode=trio_proc.returncode,
cmd=trio_proc.args,
stderr=stderr_bytes,
)
_add_stderr_note(cpe, stderr_bytes)
raise cpe