From 6255c17db93643ca52355e362e4a2d793e131002 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 29 May 2026 18:18:41 -0400 Subject: [PATCH 01/10] Use `is not None` check for peer-connect `event` Matches the explicit `dict.pop(uid, None)` contract one line above; same semantics as the prior truthy check. (this commit msg was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code (cherry picked from commit 0e3e008b0c2fdad0dc2c67d03d7877fa17e0d0ff) (cherry picked from commit 13ed668512808fbb457d62c5897c11c4c796c117) --- tractor/ipc/_server.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tractor/ipc/_server.py b/tractor/ipc/_server.py index 9701ec6d..31f4c6b0 100644 --- a/tractor/ipc/_server.py +++ b/tractor/ipc/_server.py @@ -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' From 008824d650df98a1dbcf0c61ca9d3549e3ef0211 Mon Sep 17 00:00:00 2001 From: goodboy Date: Mon, 1 Jun 2026 19:29:46 -0400 Subject: [PATCH 02/10] Add `supervise_run_process` to `trionics._subproc` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A `trio.Nursery.start()`-style wrapper around `trio.run_process()` that surfaces rc!=0 errors deterministically, ALWAYS isolates the parent controlling-tty, and optionally live-relays the child's std-streams to `log.` per-line. Suits both short-lived test-runners + long-lived daemons. `supervise_run_process()`, - Deterministic rc!=0: pass `check=False` to `trio` and do our OWN post-drain rc-check from the supervisor coro body AFTER `own_tn.__aexit__` — NOT inside the internal nursery, since that would race-cancel the still-draining relay reader and lose stderr lines. (Re)build + raise a BARE `subprocess.CalledProcessError`: `.stderr=` for programmatic callers + an `add_note()`'d `|_.stderr:` block for human teardown logs. No nursery-eg-wrapped CPE to `collapse_eg` around. - Parent controlling-tty isolation: `stdin=DEVNULL` always, `stdout=DEVNULL` unless relayed/overridden (via `stdout=` kwarg w/ `_UNSET` sentinel so explicit `None` = inherit still works). Prevents a spawned program from clobbering the launching tty's scrollback w/ control-seqs. - Live per-line relay: `relay_stdout=True`/ `relay_stderr=True` → relayed to `log.` (default `'io'`, our custom level 21). Picked to sort just above stdlib `INFO`=20 so it shows at usual `info`/`devx` levels yet stays separately filterable; `runtime`=15 was REJECTED as a default since it'd be silently filtered at usual verbosity — footgun for daemon supervisors whose whole point is visibility. STREAMED, not buffered-until-exit. - Non-blocking `tn.start()` semantics: live `trio.Process` handed up via `task_status.started()` immediately (else `tn.start()` would block till child exit, losing the long-lived-daemon use case). Supervise/relay bg tasks run to completion in this coro. - `**run_process_kwargs` forwarded verbatim (env, shell, cwd, start_new_session, executable, ...); MANAGED keys (`stdin`/`stdout`/`stderr`/`check`) win on conflict. - Crash-handling layer intentionally NOT baked in — compose `maybe_open_crash_handler()` ON TOP at the call-site. `_relay_stream_lines()` helper, - Concurrent pipe-drain reader. MANDATORY whenever piping w/o `capture_*` since nothing else drains the OS pipe — child blocks on `write()` once kernel buf (~64KiB) fills → deadlock. - Modes (combine freely): `emit`-only live relay, `accum`-only silent drain+capture (for the CPE note), or both. Per-line splitting handles cross-chunk residuals + flushes any trailing un-newline-term'd line at EOF. `_add_stderr_note()` helper, - Attaches an indented `|_.stderr:` note to a CPE via `add_note()` for legible rc!=0 reporting at teardown. Tests (`tests/trionics/test_subproc.py`), - Hermetic `trio`-only (no actor-runtime). - `test_stdout_relayed_per_line`: per-line stdout relay. - `test_parent_tty_isolated`: child fd1 is OUR pipe (no `/dev/pts/*`), fd0 pinned to `/dev/null`. - `test_no_deadlock_on_big_unnewlined_output`: 200KiB no-newline output completes under `fail_after(2)` — exercises the concurrent drain (without it, the child blocks at ~64KiB). - `test_stderr_relay_and_cpe_rebuild`: rc!=0 w/ `relay_stderr=True` → bare `CalledProcessError` w/ the `.stderr` note + per-line live relay. - `test_nonrelay_cpe_note`: rc!=0 w/o relay → same deterministic post-drain CPE w/ `.stderr` note (silent drain+capture path). Re-export `supervise_run_process` from `tractor.trionics`. Prompt-IO: ai/prompt-io/claude/20260601T231429Z_0e3e008b_prompt_io.md (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code (cherry picked from commit f595acc76c696317255fca9861d1940603f8d93b) (cherry picked from commit 6df9ee11bc37518f7247806b7c72dfe6e43a315c) --- .../20260601T231429Z_0e3e008b_prompt_io.md | 146 +++++++++ ...20260601T231429Z_0e3e008b_prompt_io.raw.md | 106 +++++++ tests/trionics/test_subproc.py | 230 ++++++++++++++ tractor/trionics/__init__.py | 3 + tractor/trionics/_subproc.py | 296 ++++++++++++++++++ 5 files changed, 781 insertions(+) create mode 100644 ai/prompt-io/claude/20260601T231429Z_0e3e008b_prompt_io.md create mode 100644 ai/prompt-io/claude/20260601T231429Z_0e3e008b_prompt_io.raw.md create mode 100644 tests/trionics/test_subproc.py create mode 100644 tractor/trionics/_subproc.py diff --git a/ai/prompt-io/claude/20260601T231429Z_0e3e008b_prompt_io.md b/ai/prompt-io/claude/20260601T231429Z_0e3e008b_prompt_io.md new file mode 100644 index 00000000..2f58db49 --- /dev/null +++ b/ai/prompt-io/claude/20260601T231429Z_0e3e008b_prompt_io.md @@ -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.` 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. diff --git a/ai/prompt-io/claude/20260601T231429Z_0e3e008b_prompt_io.raw.md b/ai/prompt-io/claude/20260601T231429Z_0e3e008b_prompt_io.raw.md new file mode 100644 index 00000000..362c4f4b --- /dev/null +++ b/ai/prompt-io/claude/20260601T231429Z_0e3e008b_prompt_io.raw.md @@ -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. diff --git a/tests/trionics/test_subproc.py b/tests/trionics/test_subproc.py new file mode 100644 index 00000000..136bca9f --- /dev/null +++ b/tests/trionics/test_subproc.py @@ -0,0 +1,230 @@ +''' +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 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.` (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) + + +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__', []) + ) diff --git a/tractor/trionics/__init__.py b/tractor/trionics/__init__.py index 7271b0f3..6cf57b7b 100644 --- a/tractor/trionics/__init__.py +++ b/tractor/trionics/__init__.py @@ -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, +) diff --git a/tractor/trionics/_subproc.py b/tractor/trionics/_subproc.py new file mode 100644 index 00000000..4cd62200 --- /dev/null +++ b/tractor/trionics/_subproc.py @@ -0,0 +1,296 @@ +# 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 . + +''' +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() + + +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, +) -> None: + ''' + Concurrently drain a child subproc's `stdout`/`stderr` + PIPE; relay each COMPLETE line to `emit` (a bound + `log.` 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 + if emit is None: + continue # drain(+accum)-only + buf: bytes = residual + chunk + *lines, residual = buf.split(b'\n') + for raw in lines: + line: str = raw.decode( + errors='replace', + ).rstrip('\r') + emit(f'[{tag}] {line}') + + # flush any trailing partial (un-newline-term'd) line @ EOF + if ( + emit is not None + and + residual + ): + line: str = residual.decode( + errors='replace', + ).rstrip('\r') + emit(f'[{tag}] {line}') + + +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 = _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 a + BARE CPE (with a `.stderr` note) from this coro's body + AFTER the child exits — so there's no nursery-eg-wrapped + CPE to catch/`collapse_eg`, and the relay reader is never + race-cancelled mid-drain. + + - 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.` 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, ...), + ) + + ''' + emit: Callable[[str], None] = getattr(log, relay_level) + 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 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: + 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, + ) + ) + + # 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'' + ) + cpe = subprocess.CalledProcessError( + returncode=trio_proc.returncode, + cmd=trio_proc.args, + stderr=stderr_bytes, + ) + _add_stderr_note(cpe, stderr_bytes) + raise cpe From 14329ec1d6d439abdd68ff545b5a761833a8b6a7 Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 2 Jun 2026 00:41:06 -0400 Subject: [PATCH 03/10] Strip ANSI + accept `_create(...)` in devx tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two version-compat fixes for the `devx` debugger test-suite, both about matching upstream output that got more verbose w/ recent lib releases. ANSI stripping (`tests/devx/conftest.py`), - Add `ansi_strip(text)` helper + `_ansi_re` pattern (regex per https://stackoverflow.com/a/14693789). - Apply inside `in_prompt_msg()` + `assert_before()` so substring matches against REPL/traceback output stay robust to color leakage. - Motivated by py3.13's colored tracebacks + `pdbp`/pygments highlighting leaking ANSI even when `PYTHON_COLORS=0` is set in the `spawn` fixture (not every renderer in the spawned subproc honors it). - Replaces the longstanding inline TODO that linked the SO answer w/o impl'ing. trio 0.30+ `Cancelled._create(` match (`test_debugger`), - In `test_shield_pause` swap the two `"raise Cancelled._create()"` assertion patterns → `"raise Cancelled._create("` (open-paren form, no closing). - trio >=0.30 raises a multi-line `raise Cancelled._create(source=.., reason=.., source_task=..)` w/ cancel-reason metadata, so the legacy bare-`()` form no longer matches. Inline comment documents the trio-version pivot. (this commit msg was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code (cherry picked from commit 3854cf5ecb9cc90f93b5cd829a42e162061f3ffb) (cherry picked from commit c07cf2546bf90dfec30c99498b5653a7374868a3) --- tests/devx/conftest.py | 31 ++++++++++++++++++++++++++----- tests/devx/test_debugger.py | 14 ++++++++++++-- 2 files changed, 38 insertions(+), 7 deletions(-) diff --git a/tests/devx/conftest.py b/tests/devx/conftest.py index 7b0d96bb..e64d38dc 100644 --- a/tests/devx/conftest.py +++ b/tests/devx/conftest.py @@ -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 diff --git a/tests/devx/test_debugger.py b/tests/devx/test_debugger.py index 94515aa4..1c3338c5 100644 --- a/tests/devx/test_debugger.py +++ b/tests/devx/test_debugger.py @@ -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()`. From 8fcb9eb7d61f08429726704184974831a937e6e9 Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 23 Jun 2026 19:03:56 -0400 Subject: [PATCH 04/10] Skip `test_parent_tty_isolated` off-Linux (`/proc`-only) --- tests/trionics/test_subproc.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/trionics/test_subproc.py b/tests/trionics/test_subproc.py index 136bca9f..a860ec6e 100644 --- a/tests/trionics/test_subproc.py +++ b/tests/trionics/test_subproc.py @@ -14,6 +14,7 @@ Hermetic `trio`-only coverage (no actor-runtime needed): ''' from functools import partial +import platform import subprocess import pytest @@ -69,6 +70,10 @@ def test_stdout_relayed_per_line(monkeypatch): 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) From 35414a42a9042c90a1028983284b62221145ec9d Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 24 Jun 2026 12:32:24 -0400 Subject: [PATCH 05/10] Scale `test_most_beautiful_word` deadline for CI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flat `trio.fail_after(1)` wraps a whole actor spawn + IPC round-trip in `test_most_beautiful_word` — sub-second on a warm box, but slow/noisy CI runners (esp. macOS) blow the 1s deadline with a `trio.TooSlowError`. Scale the budget by `cpu_scaling_factor()`: a strict `1s` locally (factor `1.0`), CI/CPU-throttle headroom (~3x) on macOS. Review: PR #465 (copilot + self CI-triage) https://github.com/goodboy/tractor/pull/465#pullrequestreview-4548191641 (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- tests/test_spawning.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/test_spawning.py b/tests/test_spawning.py index 63a2fb8e..d12c9b28 100644 --- a/tests/test_spawning.py +++ b/tests/test_spawning.py @@ -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: From 7f6782b9be67b1de7ebdc9fb152cda68d4945331 Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 24 Jun 2026 13:11:40 -0400 Subject: [PATCH 06/10] Fix `test_simple_context` for `trio>=0.33` groups MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under `trio>=0.33`'s strict exception-groups a lone `error_parent` arrives wrapped in a 1-exc `ExceptionGroup` instead of collapsing to the bare exc, so it skips the `except error_parent` arm and lands in the group arm — where `is_multi_cancelled()` is (correctly) `False`, tripping the assert. Also accept `beg.subgroup(error_parent) is not None`. Review: PR #465 (copilot + self CI-triage) https://github.com/goodboy/tractor/pull/465#pullrequestreview-4548191641 (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- tests/test_context_stream_semantics.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/tests/test_context_stream_semantics.py b/tests/test_context_stream_semantics.py index 3f5f2cee..4e23dcae 100644 --- a/tests/test_context_stream_semantics.py +++ b/tests/test_context_stream_semantics.py @@ -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) From e076296cdaf6ef8bd4e6a32fada89b727cc3ccc8 Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 24 Jun 2026 19:04:35 -0400 Subject: [PATCH 07/10] Guard `stdout=PIPE`/`capture_*` footguns Two `supervise_run_process()` caller-kwarg footguns that previously failed obscurely now raise a clear, early `ValueError`: - a bare `stdout=subprocess.PIPE` override spawned NO drain reader (only `relay_stdout` does), so a child emitting >64KiB would deadlock on a full ~64KiB pipe buffer. - `trio`'s `capture_stdout`/`capture_stderr` alias our MANAGED `stdout`/`stderr` keys; forwarding them made `trio.run_process` raise an opaque "can't specify both" error, so reject them up-front pointing at `relay_*` / the `stdout=` override. Review: PR #465 (copilot-pull-request-reviewer) https://github.com/goodboy/tractor/pull/465#pullrequestreview-4548191641 (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- tractor/trionics/_subproc.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/tractor/trionics/_subproc.py b/tractor/trionics/_subproc.py index 4cd62200..ec894ba2 100644 --- a/tractor/trionics/_subproc.py +++ b/tractor/trionics/_subproc.py @@ -201,6 +201,20 @@ async def supervise_run_process( # 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 @@ -210,6 +224,16 @@ async def supervise_run_process( 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 From a3f8765264c85e3e4129638faa4299f61e0b32df Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 24 Jun 2026 19:05:55 -0400 Subject: [PATCH 08/10] Resolve relay `emit` only when relaying `supervise_run_process()` bound the relay `emit` method (`getattr(log, relay_level)`) unconditionally at entry, so a bad/typo'd `relay_level` raised `AttributeError` even for a call that requested NO relay (the common silent path). Bind it lazily behind `relay_stdout or relay_stderr` so only an actual relay request validates the level. Review: PR #465 (copilot-pull-request-reviewer) https://github.com/goodboy/tractor/pull/465#pullrequestreview-4548191641 (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- tractor/trionics/_subproc.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/tractor/trionics/_subproc.py b/tractor/trionics/_subproc.py index ec894ba2..ab287d52 100644 --- a/tractor/trionics/_subproc.py +++ b/tractor/trionics/_subproc.py @@ -190,7 +190,14 @@ async def supervise_run_process( ) ''' - emit: Callable[[str], None] = getattr(log, relay_level) + # 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 From 02cb789901594b4a34c0efe6ab86ba40cda50293 Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 24 Jun 2026 19:08:19 -0400 Subject: [PATCH 09/10] Bound retained `stderr` to a tail + de-dup decode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_relay_stream_lines()` kept ALL captured `stderr` for the rc!=0 `CalledProcessError` note, so a long-lived `check=True` child could buffer its entire stderr history in parent mem. Add an `accum_cap` (the new `_STDERR_NOTE_TAIL_BYTES`=16KiB) that trims the buffer to its TAIL — the bytes nearest the failure, which are what the note wants anyway. While here, factor the per-line decode/`rstrip` into a `_decode_line()` helper so the in-loop + trailing-flush emits no longer duplicate it. Review: PR #465 (copilot-pull-request-reviewer) https://github.com/goodboy/tractor/pull/465#pullrequestreview-4548191641 (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- tractor/trionics/_subproc.py | 37 ++++++++++++++++++++++++++++-------- 1 file changed, 29 insertions(+), 8 deletions(-) diff --git a/tractor/trionics/_subproc.py b/tractor/trionics/_subproc.py index ab287d52..cd466127 100644 --- a/tractor/trionics/_subproc.py +++ b/tractor/trionics/_subproc.py @@ -42,6 +42,22 @@ log = get_logger() _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, @@ -65,6 +81,7 @@ async def _relay_stream_lines( 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` @@ -95,15 +112,21 @@ async def _relay_stream_lines( 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: - line: str = raw.decode( - errors='replace', - ).rstrip('\r') - emit(f'[{tag}] {line}') + emit(f'[{tag}] {_decode_line(raw)}') # flush any trailing partial (un-newline-term'd) line @ EOF if ( @@ -111,10 +134,7 @@ async def _relay_stream_lines( and residual ): - line: str = residual.decode( - errors='replace', - ).rstrip('\r') - emit(f'[{tag}] {line}') + emit(f'[{tag}] {_decode_line(residual)}') async def supervise_run_process( @@ -293,6 +313,7 @@ async def supervise_run_process( emit=emit if relay_stderr else None, tag=f'{tag}:err', accum=stderr_accum, + accum_cap=_STDERR_NOTE_TAIL_BYTES, ) ) From 6764251e6b114cdc17d8749a7485509a0fc3c615 Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 24 Jun 2026 19:09:45 -0400 Subject: [PATCH 10/10] Clarify eg-wrap contract + tighten `stdout` typing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Doc/type accuracy in `supervise_run_process()`: - the docstring claimed the rc!=0 `CalledProcessError` had "no nursery-eg-wrapped CPE to catch/`collapse_eg`", but it IS raised as a task into the parent `tn`, so under `trio>=0.33` it surfaces `ExceptionGroup`-wrapped like any `tn.start()`-task raise. Say so + point at `except*` / `collapse_eg()` (matching the inline rc-check comment). - widen `stdout: int` to `int|None` — the param accepts an explicit `None` (inherit) per its own docstring. - note that stdout is NOT captured, so the CPE note carries only stderr (a cmd logging errors to stdout should `relay_stdout=True`). Review: PR #465 (copilot-pull-request-reviewer) https://github.com/goodboy/tractor/pull/465#pullrequestreview-4548191641 (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- tractor/trionics/_subproc.py | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/tractor/trionics/_subproc.py b/tractor/trionics/_subproc.py index cd466127..814fe0a4 100644 --- a/tractor/trionics/_subproc.py +++ b/tractor/trionics/_subproc.py @@ -160,7 +160,7 @@ async def supervise_run_process( # non-relay `stdout` override; defaults (via `_UNSET`) to # `DEVNULL` so we NEVER inherit (+ thus can't clobber) the # parent controlling-tty. - stdout: int = _UNSET, + stdout: int|None = _UNSET, task_status: trio.TaskStatus[ trio.Process @@ -178,11 +178,14 @@ async def supervise_run_process( - surfaces a rc!=0 `subprocess.CalledProcessError` DETERMINISTICALLY: we pass `check=False` to `trio` and - do our OWN post-drain rc-check, (re)building + raising a - BARE CPE (with a `.stderr` note) from this coro's body - AFTER the child exits — so there's no nursery-eg-wrapped - CPE to catch/`collapse_eg`, and the relay reader is never - race-cancelled mid-drain. + 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 @@ -339,6 +342,10 @@ async def supervise_run_process( 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,