Compare commits

...

20 Commits

Author SHA1 Message Date
Gud Boi 3bc6680982 Use `cpu_perf_headroom()` for timing-test deadlines
`cpu_perf_headroom()` is a strict SUPERSET of `cpu_scaling_factor()`
— it folds in static cpu-freq scaling + the slow-CI bump AND the
sustained-load power-cap throttle probe. Point the timing-deadline
call sites at it so they're robust to sustained throttling too (the
gremlin behind mass `trio` deadline-miss flakes),

- `test_legacy_one_way_streaming`: `time_quad_ex` cancel-deadline +
  `test_a_quadruple_example`'s `this_fast` smoke bound.
- `test_cancellation`: `test_cancel_via_SIGINT_other_task` +
  `test_fast_graceful_cancel_*` deadlines.

`cpu_perf_headroom >= cpu_scaling_factor` always, so never less
headroom (CI stays green); `cpu_scaling_factor()` remains the
internal static component.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-06-24 20:08:03 -04:00
Gud Boi 23580e9cb8 Add `cpu-perf-check` sustained-throttle gate script
Standalone CLI companion to `cpu_perf_headroom()` (20cb99ec): idle
freq snapshots LIE — every static knob (`governor`, EPP,
`platform_profile`, `scaling_max_freq`) can read "performance"
while a firmware/EC power cap (AMD PPT/STAPM + friends) clamps the
package to ~30% the moment a sustained multi-core load lands,
masquerading as a `trio`-backend deadline-miss "regression" on
byte-identical code.

Deats,
- burns every core for `CPU_PERF_SECS` (default 4s) and samples the
  ACHIEVED `scaling_cur_freq` steady-state (post boost-ramp) vs the
  package max ceiling,
- exits 0 when the sustained fraction clears
  `CPU_PERF_HEALTHY_FRAC` (default 0.45), 1 when throttled — so it
  gates a suite run: `scripts/cpu-perf-check && pytest tests/ ...`,
- prints the static knobs first (to show they all read fine) then
  the remediation list on failure (`platform_profile` bounce, USB-C
  PD replug, `ryzenadj`, reboot) w/ the key reminder: do NOT bump
  test budgets — the box is slow, not the code.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-06-24 20:08:03 -04:00
Gud Boi 2a7c46add2 Add `cpu_perf_headroom()` for throttle-aware deadlines
Mass `trio` deadline-miss failures on byte-identical code turned
out to be a firmware/EC power-cap (AMD PPT/STAPM) clamping the
all-core sustained clock while every static knob (`governor`,
`scaling_max_freq`, EPP, platform-profile) still read "performance"
— invisible to the existing `cpu_scaling_factor()` check. See
`scripts/cpu-perf-check` + the
`ai/conc-anal/trio_033_cancel_cascade_slowdown_depth3_issue.md`
notes.

Deats,
- add `_measure_sustained_headroom()` to `tests/conftest.py`: a
  one-shot ~0.9s all-core burn (explicit `fork`-ctx `mp` procs)
  sampling achieved-vs-max freq AFTER the boost window; under a 0.6
  gate it returns the full inverse fraction (capped 4x), else 1.0;
  best-effort 1.0 on non-linux or any error,
- add `cpu_perf_headroom()`: `max()` of the static scaling factor
  and the (session-cached) sustained probe,
- inflate deadline budgets by it in `test_dynamic_pub_sub`, both
  `test_clustering` cases, the
  `test_multi_nested_subactors_error_through_nurseries` pexpect
  waits + `test_nested_multierrors`,
- `xfail(strict=False)` `test_nested_multierrors` depth=3 under
  throttle: the deep tree trips tractor's INTERNAL reap deadlines
  (`soft_kill`/`hard_kill` `terminate_after=1.6`) minting a
  `Cancelled` inside the runtime — not fixable by test-budget
  inflation; auto-clears once the box un-throttles.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-06-24 20:08:03 -04:00
Gud Boi 6764251e6b Clarify eg-wrap contract + tighten `stdout` typing
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
2026-06-24 19:09:45 -04:00
Gud Boi 02cb789901 Bound retained `stderr` to a tail + de-dup decode
`_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
2026-06-24 19:08:19 -04:00
Gud Boi a3f8765264 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
2026-06-24 19:05:55 -04:00
Gud Boi e076296cda 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
2026-06-24 19:04:35 -04:00
Gud Boi 7f6782b9be Fix `test_simple_context` for `trio>=0.33` groups
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
2026-06-24 18:19:16 -04:00
Gud Boi 35414a42a9 Scale `test_most_beautiful_word` deadline for CI
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
2026-06-24 18:19:16 -04:00
Gud Boi 8fcb9eb7d6 Skip `test_parent_tty_isolated` off-Linux (`/proc`-only) 2026-06-24 18:19:16 -04:00
Gud Boi 14329ec1d6 Strip ANSI + accept `_create(...)` in devx tests
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 3854cf5ecb)
(cherry picked from commit c07cf2546b)
2026-06-24 18:19:16 -04:00
Gud Boi 008824d650 Add `supervise_run_process` to `trionics._subproc`
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.<level>` 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.<relay_level>`
  (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 f595acc76c)
(cherry picked from commit 6df9ee11bc)
2026-06-24 18:19:16 -04:00
Gud Boi 6255c17db9 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 0e3e008b0c)
(cherry picked from commit 13ed668512)
2026-06-24 18:19:16 -04:00
Bd 6a1149d401
Merge pull request #467 from goodboy/custom_log_levels_api
Add `add_log_level()` custom log-level API
2026-06-24 18:18:37 -04:00
Gud Boi 88cc68edd3 Fix `add_log_level()` re-registration value drift
The bound `StackLevelAdapter.<name>` emit method captured its
level via a `_level: int = value` default-arg at bind time, so
a later `add_log_level(name, <new-value>)` re-registration left
the method emitting at the *stale* original level. Read
`CUSTOM_LEVELS[name_up]` at call time instead — the method binds
once but always tracks the current registered value.

Also,
- correct the `add_log_level()` docstring's "Idempotent" note
  to describe the call-time value lookup (was the misleading
  "no-op-ish refresh (won't clobber an already-bound method)").
- sync stale `_reap.py` doc/comment markers `tractor[` ->
  `_subactor[` to match the actual `_proctitle._def_prefix`
  proc-title prefix (doc-only drift; the code markers already
  referenced `_proctitle._def_prefix`).

Review: PR #467 (copilot-pull-request-reviewer)
https://github.com/goodboy/tractor/pull/467#pullrequestreview-4562945515

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-06-24 18:03:28 -04:00
Gud Boi 3727ce9e6f Bump lockfile for `xonsh>=0.23.8` release 2026-06-24 16:20:50 -04:00
Gud Boi 90c316f90c Code-style, couple newline/ws tweaks
(cherry picked from commit 8526985c97)
(cherry picked from commit 0952b33a9e)
2026-06-24 16:20:50 -04:00
Gud Boi 6db796ed83 Pin to latest `xonsh` release
(cherry picked from commit c4cad921b9)
(cherry picked from commit ade15b4204)
2026-06-24 16:20:50 -04:00
Gud Boi eb7ef3f0bf Hoist proc-title prefix to `_def_prefix` const
Make the sub-actor proc-title prefix a single
authoritative constant (`_proctitle._def_prefix`) so
the reap-recognition markers and `xontrib` banner pick
it up automatically — one place to flip the prefix
shape going fwd.

Deats,
- `_proctitle._def_prefix: str = '_subactor'`. New
  module-level const consumed by everything that needs
  to know the prefix.
- `set_actor_proctitle(actor, prefix=_def_prefix)`:
  takes an explicit `prefix` arg (default = the const)
  so callers can override per-spawn if they want.
- Default proc-title format:
  `'tractor[<reprol>]'` → `f'{prefix}[<reprol>]'`
  i.e. `_subactor[<reprol>]` by default.
- `_testing/_reap.py`: cmdline + comm markers source
  the prefix from `_proctitle._def_prefix` instead of
  the hardcoded `'tractor['`. So
  `_is_tractor_subactor()` tracks the const
  automatically.
- `xontrib/tractor_diag.xsh`: `acli.reap` orphan-mode
  banner now interpolates the
  `_TRACTOR_PROC_CMDLINE_MARKERS` tuple directly so
  the human-readable mode line stays in sync if the
  prefix shape changes again.

(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 3a45dbd503)
(cherry picked from commit fd8d39c0ce)
2026-06-24 16:20:50 -04:00
Gud Boi bb0b2ffa3c Add `add_log_level()` factory + register `IO`=21
Follow-up to f595acc7 (`supervise_run_process`) which
called `log.io(...)` for std-stream relay assuming an
`IO=21` level existed. Add the registration via a new
factory + tests covering both the factory and the new
level.

`add_log_level()` factory,
- One call wires the four (otherwise hand-synced) pieces:
  - `CUSTOM_LEVELS[NAME]` — drives the `stacklevel` bump
    in `StackLevelAdapter.log()` + `get_logger()`'s
    per-level audit.
  - `logging.addLevelName()` — stdlib name registration.
  - `STD_PALETTE[NAME]` + `BOLD_PALETTE['bold'][NAME]` —
    color entries consumed by `get_console_log()`'s
    `ColoredFormatter` build.
  - Same-named (lowercase) emit method bound on
    `StackLevelAdapter` so `log.<name>('msg')` works +
    `get_logger()`'s per-level method audit passes.
- Idempotent: re-registering an existing name is a
  no-op-ish refresh that won't clobber an already-bound
  method.
- Method binding uses a default-arg `_level=value` so
  the level int is captured (not late-bound across
  multiple registrations).

`IO=21` level (first user),
- Purple. Used by `tractor.trionics._subproc`'s
  std-stream relay (see f595acc7).
- Value 21 picked to sit just ABOVE stdlib `INFO`=20 so
  it's SHOWN BY DEFAULT at usual `info`/`devx` console
  levels — a `runtime`=15 relay would be silently
  filtered (footgun for daemon supervisors whose whole
  point is visibility). Still distinctly labeled +
  filterable.

Tests (`tests/test_log_sys.py`),
- `test_io_custom_level_registered`: validates the IO
  level is fully wired (`CUSTOM_LEVELS`, `addLevelName`,
  both palettes, `StackLevelAdapter.io()` callable);
  emits a record + sanity-asserts `21 >= INFO(20)`.
- `test_add_log_level_pluggable`: registers a fresh
  `XLVL=19` (cyan) via `add_log_level()`, asserts all
  four wires + the bound `xlog.xlvl()` emit, then
  try/finally cleans up the module-global mutations so
  later `get_logger()` audits don't trip on a
  half-removed level.

(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 7bd7dd50c7)
(cherry picked from commit 93558fe3c9)
2026-06-24 16:20:50 -04:00
26 changed files with 1467 additions and 72 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

@ -104,7 +104,7 @@ testing = [
repl = [
"pyperclip>=1.9.0",
"prompt-toolkit>=3.0.50",
"xonsh>=0.23.0",
"xonsh>=0.23.8",
"psutil>=7.0.0",
]
lint = [

View File

@ -0,0 +1,159 @@
#!/usr/bin/env python3
# tractor: distributed structured concurrency.
# Copyright 2018-eternity Tyler Goodlet.
#
# SPDX-License-Identifier: AGPL-3.0-or-later
'''
`cpu-perf-check` — sustained-load CPU throttle detector.
Idle freq snapshots LIE. A laptop can read
`governor=performance`, `EPP=performance`,
`platform_profile=performance`, `scaling_max_freq=<full>`
and momentarily clock a P-core at 5GHz — while a
firmware/EC power cap (AMD PPT/STAPM and friends) clamps
the whole package to ~1.5GHz the instant a sustained
multi-core load lands. That throttle masquerades as a
`trio`-backend test *regression*: a wave of `fail_after` /
`TooSlowError` / `Cancelled(source='deadline')` deadline
misses on spawn-heavy tests, on byte-identical code that
was green yesterday.
The existing `tests/conftest.py:cpu_scaling_factor()` only
reads STATIC `scaling_max_freq` vs `*_pstate_max_freq`, so
it returns `1.0` (no throttle) during exactly this failure
— it can't see the cap. This script complements it by
BURNING every core for a few seconds and sampling the
ACHIEVED `scaling_cur_freq`, which is the only thing that
exposes the clamp.
Exit code: `0` if sustained perf looks restored, `1` if
throttled — so it gates a test run:
py313/bin/python scripts/cpu-perf-check && pytest tests/ ...
Tunables (env-overridable):
CPU_PERF_SECS load duration (default 4.0)
CPU_PERF_HEALTHY_FRAC sustained/max floor (default 0.45)
'''
from __future__ import annotations
import glob
import os
import time
import multiprocessing as mp
def _read(path: str) -> str | None:
try:
with open(path) as f:
return f.read().strip()
except OSError:
return None
def _cur_freqs_mhz() -> list[int]:
out: list[int] = []
for f in glob.glob(
'/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_cur_freq'
):
if (v := _read(f)):
out.append(int(v) // 1000)
return out
def _pkg_max_mhz() -> int:
'''
Highest per-core ceiling across the package — the
P-core max on hybrid parts.
'''
mxs: list[int] = []
for f in glob.glob(
'/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_max_freq'
):
if (v := _read(f)):
mxs.append(int(v) // 1000)
return max(mxs) if mxs else 0
def _burn(stop: float) -> None:
x: int = 1
while time.perf_counter() < stop:
x += x * x ^ 0x5
def main(
secs: float = float(os.environ.get('CPU_PERF_SECS', 4.0)),
# sustained aggregate must clear this fraction of the
# package max-freq ceiling. Throttled (~1.5GHz of 5.1GHz)
# ~= 0.29; a healthy all-core load easily clears 0.5.
healthy_frac: float = float(
os.environ.get('CPU_PERF_HEALTHY_FRAC', 0.45)
),
) -> int:
if not glob.glob('/sys/devices/system/cpu/cpu0/cpufreq'):
print('no cpufreq sysfs (non-linux?) — skipping, assume OK')
return 0
b: str = '/sys/devices/system/cpu/cpu0/cpufreq/'
pkg_max: int = _pkg_max_mhz()
print('=== static knobs (ALL can read fine while throttled) ===')
print(f' governor : {_read(b + "scaling_governor")}')
print(f' EPP : {_read(b + "energy_performance_preference")}')
print(f' platform_profile : '
f'{_read("/sys/firmware/acpi/platform_profile")}')
print(f' pkg max freq : {pkg_max} MHz')
ncpu: int = os.cpu_count() or 1
print(f'\n=== sustained {ncpu}-core load ({secs:.0f}s) — the real test ===')
stop: float = time.perf_counter() + secs
procs = [
mp.Process(target=_burn, args=(stop,))
for _ in range(ncpu)
]
for p in procs:
p.start()
# skip the initial ~0.6s ramp, then sample steady-state
samples: list[int] = []
time.sleep(0.6)
while time.perf_counter() < stop - 0.2:
if (fr := _cur_freqs_mhz()):
samples.append(sum(fr) // len(fr))
time.sleep(0.3)
for p in procs:
p.join()
if not (samples and pkg_max):
print(' could not sample cur freq — assume OK')
return 0
sustained: int = sum(samples) // len(samples)
frac: float = sustained / pkg_max
print(f' aggregate cur-freq samples: {samples}')
print(f' sustained avg : {sustained} MHz '
f'({frac * 100:.0f}% of {pkg_max} MHz max)')
if frac < healthy_frac:
print(
f'\n ❌ THROTTLED — sustained {sustained}MHz is only '
f'{frac * 100:.0f}% of max (< {healthy_frac * 100:.0f}%).\n'
f' Power cap (PPT/STAPM) still engaged. Fixes:\n'
f' - bounce /sys/firmware/acpi/platform_profile\n'
f' (balanced -> performance)\n'
f' - unplug/replug USB-C to re-negotiate PD\n'
f' - ryzenadj to lift STAPM/PPT\n'
f' - else reboot\n'
f' Do NOT bump test budgets — the box is slow, not the code.'
)
return 1
print(
f'\n ✅ PERF OK — sustained {sustained}MHz holds '
f'{frac * 100:.0f}% of max. Cap looks lifted; safe to run tests.'
)
return 0
if __name__ == '__main__':
raise SystemExit(main())

View File

@ -134,6 +134,139 @@ def cpu_scaling_factor() -> float:
return factor
# session-cached sustained-load throttle multiplier — measured
# once (lazily) on the first `cpu_perf_headroom()` call. `None`
# = not-yet-measured.
_sustained_headroom: float|None = None
def _measure_sustained_headroom(
secs: float = 0.9,
# a healthy all-core sustained clock holds AT/ABOVE this
# fraction of the package single-core max ceiling (boost sags
# under full multi-core load even un-throttled, but not far);
# at/above it we assume no throttle and return 1.0.
throttle_gate: float = 0.6,
max_headroom: float = 4.,
) -> float:
'''
One-shot all-core burn returning a latency multiplier
(>= 1.0) that reflects *sustained-load* CPU throttle.
Catches the firmware/EC power-cap clamp (AMD PPT/STAPM &
friends) that pins achieved `scaling_cur_freq` to a fraction
of the ceiling under multi-core load while EVERY static knob
(`governor`, `scaling_max_freq`, `EPP`, `platform_profile`)
still reads "full performance". That cap is INVISIBLE to
`cpu_scaling_factor()` and is the gremlin behind mass `trio`
deadline-miss failures on byte-identical code see
`scripts/cpu-perf-check`.
Best-effort: returns 1.0 on non-linux / missing sysfs / any
error so it can never break a test run.
'''
import glob
import multiprocessing as mp
def _read_mhz(path: str) -> int|None:
try:
return int(open(path).read()) // 1000
except OSError:
return None
try:
maxs: list[int] = [
v for f in glob.glob(
'/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_max_freq'
)
if (v := _read_mhz(f)) is not None
]
pkg_max: int = max(maxs) if maxs else 0
if not pkg_max:
return 1.
def _burn(stop: float) -> None:
x: int = 1
while time.perf_counter() < stop:
x += x * x ^ 0x5
# explicit `fork` ctx so we're immune to whatever global
# mp start-method tractor/the suite may have set (`spawn`
# would re-exec + re-import 24x — slow and pointless here).
ctx = mp.get_context('fork')
ncpu: int = os.cpu_count() or 1
stop: float = time.perf_counter() + secs
procs = [
ctx.Process(target=_burn, args=(stop,), daemon=True)
for _ in range(ncpu)
]
for p in procs:
p.start()
# skip the ~0.4s boost window so we sample the steady
# state AFTER any power-cap has engaged.
samples: list[int] = []
time.sleep(0.4)
while time.perf_counter() < stop - 0.1:
curs: list[int] = [
v for f in glob.glob(
'/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_cur_freq'
)
if (v := _read_mhz(f)) is not None
]
if curs:
samples.append(sum(curs) // len(curs))
time.sleep(0.15)
for p in procs:
p.join()
if not samples:
return 1.
frac: float = (sum(samples) // len(samples)) / pkg_max
# below the gate we read it as a power-cap throttle. The
# spawn/IPC/fork-bound work these budgets guard slows ~1:1
# with the achieved-vs-max freq ratio, so compensate by the
# FULL inverse fraction (a boost-discounted factor
# under-shoots and still trips the marginal cases).
if frac >= throttle_gate:
return 1.
return min(max_headroom, 1. / frac)
except Exception:
return 1.
def cpu_perf_headroom() -> float:
'''
Latency-headroom multiplier (>= 1.0) covering BOTH cpu-perf
throttle classes multiply a test's deadline by it, e.g.
`timeout *= cpu_perf_headroom()`:
- static cpu-freq scaling via `cpu_scaling_factor()`
(governor/policy lowered the `scaling_max_freq` ceiling).
- sustained-load power-cap throttle via
`_measure_sustained_headroom()` (firmware/EC PPT/STAPM
clamps achieved freq under load while every static knob
reads "performance"; INVISIBLE to the static check). This
is the gremlin behind mass `trio` deadline-miss failures
on unchanged code see
`ai/conc-anal/trio_033_cancel_cascade_slowdown_depth3_issue.md`.
The sustained probe runs ONCE per session (cached); the cost
is a ~0.9s all-core burn on first call only.
'''
global _sustained_headroom
static: float = cpu_scaling_factor()
if _non_linux:
return static
if _sustained_headroom is None:
_sustained_headroom = _measure_sustained_headroom()
return max(static, _sustained_headroom)
# NOTE, the `--ll`/`--tl` CLI flags + the `loglevel`, `test_log`
# and `testing_pkg_name` fixtures have been factored into the
# `tractor._testing.pytest` plugin (loaded via the `-p` entry in

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

@ -794,6 +794,14 @@ def test_multi_nested_subactors_error_through_nurseries(
loglevel='pdb',
)
last_send_char: str|None = None
# inflate pexpect waits under CPU throttle — incl. the
# sustained-load power-cap invisible to static freq reads — so
# a slow-to-boot child REPL doesn't trip a false `TIMEOUT`.
# See `scripts/cpu-perf-check`.
from ..conftest import cpu_perf_headroom
headroom: float = cpu_perf_headroom()
for (
i,
send_char,
@ -817,6 +825,9 @@ def test_multi_nested_subactors_error_through_nurseries(
if is_forking_spawner:
timeout += 4
if headroom != 1.:
timeout *= headroom
try:
child.expect(
PROMPT,
@ -1186,7 +1197,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 +1220,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

@ -8,9 +8,9 @@ after `Actor` construction, so any spawned sub-actor process
should:
- have `argv[0]` (== `/proc/<pid>/cmdline`) start with
`tractor[<aid.reprol()>]`
- have `/proc/<pid>/comm` start with `tractor[` (kernel
truncates to ~15 bytes)
`<_def_prefix>[<aid.reprol()>]` (currently `_subactor[]`)
- have `/proc/<pid>/comm` start with `<_def_prefix>[`
(kernel truncates to ~15 bytes)
- be detected as a tractor sub-actor by
`_is_tractor_subactor(pid)` via the cmdline marker.
@ -27,7 +27,10 @@ import trio
import tractor
from tractor.runtime._runtime import Actor
from tractor.devx._proctitle import set_actor_proctitle
from tractor.devx._proctitle import (
set_actor_proctitle,
_def_prefix,
)
from tractor._testing._reap import (
_is_tractor_subactor,
_read_cmdline,
@ -41,8 +44,9 @@ _non_linux: bool = platform.system() != 'Linux'
def test_set_actor_proctitle_format():
'''
`set_actor_proctitle()` returns the canonical
`tractor[<aid.reprol()>]` form and actually mutates
the running proc's title.
`<_def_prefix>[<aid.reprol()>]` form (currently
`_subactor[]`) and actually mutates the running
proc's title.
'''
pytest.importorskip(
@ -60,12 +64,14 @@ def test_set_actor_proctitle_format():
)
title: str = set_actor_proctitle(actor)
# canonical wrapping: `tractor[<aid.reprol()>]`. We
# compare against the runtime-computed `reprol()`
# rather than a hard-coded value so the test stays
# decoupled from `Aid.reprol()`'s internal format
# (currently `<name>@<pid>`, but could evolve).
expected: str = f'tractor[{actor.aid.reprol()}]'
# canonical wrapping: `<_def_prefix>[<aid.reprol()>]`.
# We source BOTH the prefix (`_def_prefix`) and the
# runtime-computed `reprol()` rather than hard-coding,
# so the test stays decoupled from the prefix shape
# (flipped to `_subactor` in `3a45dbd5`) AND from
# `Aid.reprol()`'s internal format (currently
# `<name>@<pid>`, but could evolve).
expected: str = f'{_def_prefix}[{actor.aid.reprol()}]'
assert title == expected
# sanity: the actor's name must be in the title
# somewhere (so a future `reprol()` change that
@ -140,15 +146,17 @@ def test_subactor_proctitle_visible_via_proc():
)
pid, info = matched[0]
# canonical proctitle prefix in cmdline (full form)
assert info['cmdline'].startswith('tractor[proctitle_boi@'), (
f'cmdline missing `tractor[proctitle_boi@…]` prefix: '
# canonical proctitle prefix in cmdline (full form);
# prefix sourced from `_def_prefix` so it tracks the
# `3a45dbd5` flip (`tractor[` -> `_subactor[`).
assert info['cmdline'].startswith(f'{_def_prefix}[proctitle_boi@'), (
f'cmdline missing `{_def_prefix}[proctitle_boi@…]` prefix: '
f'{info["cmdline"]!r}'
)
# comm is kernel-truncated to ~15 bytes — just check the
# `tractor[` prefix made it.
assert info['comm'].startswith('tractor['), (
f'comm missing `tractor[` prefix: {info["comm"]!r}'
# `<_def_prefix>[` prefix made it.
assert info['comm'].startswith(f'{_def_prefix}['), (
f'comm missing `{_def_prefix}[` prefix: {info["comm"]!r}'
)
# intrinsic-signal detector should match.
assert info['is_tractor'] is True

View File

@ -188,11 +188,18 @@ def test_dynamic_pub_sub(
# sits forever until external SIGINT. The `afk_alarm_w_trace`
# outer guard below is the AFK-safety counterpart (SIGALRM
# raises in the main thread regardless of trio scope state).
fail_after_s: int = (
fail_after_s: float = (
8
if is_forking_spawner
else 20
)
# inflate under CPU throttle — incl. the sustained-load
# power-cap invisible to static freq reads — so a slow box
# doesn't trip the deadline. See `scripts/cpu-perf-check`.
from .conftest import cpu_perf_headroom
headroom: float = cpu_perf_headroom()
if headroom != 1.:
fail_after_s *= headroom
async def main():
# bug-class-3 breadcrumb: tag each level of the cancel path

View File

@ -596,6 +596,15 @@ async def test_nested_multierrors(
# depth=3, BOTH variants will reliably `xpass` and
# pytest will yell — our signal to drop the marker. See
# `ai/conc-anal/cancel_cascade_too_slow_under_main_thread_forkserver_issue.md`.
#
# Probe CPU throttle ONCE up-front (folds in the sustained-load
# power-cap that static freq reads miss): used BOTH to inflate
# the deadline budget below AND to xfail depth=3, whose failure
# mode under throttle is a runtime-internal reap deadline — not
# a test-budget miss. See `scripts/cpu-perf-check`.
from .conftest import cpu_perf_headroom
headroom: float = cpu_perf_headroom()
if start_method == 'main_thread_forkserver':
request.node.add_marker(
pytest.mark.xfail(
@ -609,6 +618,34 @@ async def test_nested_multierrors(
)
)
# Under CPU throttle (incl. the sustained-load power-cap that
# static freq reads miss) the DEEP depth=3 tree trips tractor's
# INTERNAL reap deadlines (`soft_kill`/`hard_kill`
# `move_on_after`/`terminate_after=1.6`) before slow subprocs
# exit, injecting a `Cancelled(source='deadline')` into the BEG
# — the SAME shape-mismatch class as the MTF xfail above, and
# NOT fixable by inflating the test-level budget (the Cancelled
# is minted inside the runtime, not by our `fail_after`).
# xfail(strict=False) so it auto-clears the moment the box is
# un-throttled (`headroom == 1.`); depth=1's shallow tree stays
# under those internal deadlines so it just rides the budget
# inflation below. See `scripts/cpu-perf-check`.
elif (
depth == 3
and
headroom != 1.
):
request.node.add_marker(
pytest.mark.xfail(
strict=False,
reason=(
'CPU throttled — tractor reap deadline injects '
'Cancelled into BEG; see conc-anal/'
'trio_033_cancel_cascade_slowdown_depth3_issue.md'
),
)
)
# Per-backend/-depth budgets: in the non-hang case the
# whole spawn + cancel-cascade should complete in well
# under these. On the borderline hang case the
@ -636,11 +673,14 @@ async def test_nested_multierrors(
case ('main_thread_forkserver', 3):
timeout = 30
# headroom for CPU-freq scaling AND/OR slow CI so the inner
# snapshot-capturing budget doesn't fire spuriously on a
# sluggish runner; see `cpu_scaling_factor()`.
from .conftest import cpu_scaling_factor
timeout *= cpu_scaling_factor()
# inflate the budget by the throttle headroom probed above so
# a slow box doesn't masquerade as a deadline regression.
# NOTE, `headroom = cpu_perf_headroom()` (set above) is the
# SUPERSET of `cpu_scaling_factor()` — it folds in the static
# cpu-freq scaling + slow-CI bump AND the sustained-load
# throttle probe this depth-3 cascade was the poster child for.
if headroom != 1.:
timeout *= headroom
async with fail_after_w_trace(timeout):
try:
@ -748,7 +788,7 @@ def test_cancel_via_SIGINT_other_task(
started from a seperate ``trio`` child task.
'''
from .conftest import cpu_scaling_factor
from .conftest import cpu_perf_headroom
pid: int = os.getpid()
timeout: float = (
@ -758,8 +798,9 @@ def test_cancel_via_SIGINT_other_task(
if _friggin_windows: # smh
timeout += 1
# add latency headroom for CPU freq scaling (auto-cpufreq et al.)
headroom: float = cpu_scaling_factor()
# latency headroom for static cpu-freq scaling + sustained-load
# throttle + CI (auto-cpufreq et al.); see `cpu_perf_headroom()`.
headroom: float = cpu_perf_headroom()
if headroom != 1.:
timeout *= headroom
@ -955,11 +996,11 @@ def test_fast_graceful_cancel_when_spawn_task_in_soft_proc_wait_for_daemon(
if _friggin_windows: # smh
timeout += 1
# CPU-scaling / CI latency headroom — macOS CI especially is
# slow for this graceful-vs-hard-reap timing race; see
# `cpu_scaling_factor()`.
from .conftest import cpu_scaling_factor
timeout *= cpu_scaling_factor()
# CPU-scaling / sustained-throttle / CI latency headroom — macOS
# CI especially is slow for this graceful-vs-hard-reap timing
# race; see `cpu_perf_headroom()`.
from .conftest import cpu_perf_headroom
timeout *= cpu_perf_headroom()
async def main():
start = time.time()

View File

@ -24,8 +24,14 @@ def test_empty_mngrs_input_raises(
'actor-cluster teardown hangs intermittently on UDS'
)
# inflate under CPU throttle — incl. the sustained-load
# power-cap invisible to static freq reads. See
# `scripts/cpu-perf-check`.
from .conftest import cpu_perf_headroom
fail_after_s: float = 3 * cpu_perf_headroom()
async def main():
with trio.fail_after(3):
with trio.fail_after(fail_after_s):
async with (
open_actor_cluster(
modules=[__name__],
@ -93,6 +99,13 @@ async def test_streaming_to_actor_cluster(
10 if is_forking_spawner
else 6
)
# inflate under CPU throttle — incl. the sustained-load
# power-cap invisible to static freq reads. See
# `scripts/cpu-perf-check`.
from .conftest import cpu_perf_headroom
headroom: float = cpu_perf_headroom()
if headroom != 1.:
delay *= headroom
with trio.fail_after(delay):
async with (
open_actor_cluster(modules=[__name__]) as portals,

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

@ -326,11 +326,11 @@ def time_quad_ex(
):
timeout += 1
# inflate the cancel-deadline for CPU-freq scaling AND/OR CI
# latency (see `cpu_scaling_factor()`) so the example isn't
# cancelled mid-stream on a throttled/CI runner.
from .conftest import cpu_scaling_factor
timeout *= cpu_scaling_factor()
# inflate the cancel-deadline for static cpu-freq scaling +
# sustained-load throttle + CI latency (see `cpu_perf_headroom()`)
# so the example isn't cancelled mid-stream on a throttled/CI box.
from .conftest import cpu_perf_headroom
timeout *= cpu_perf_headroom()
start: float = time.time()
results: list[int] = trio.run(partial(
@ -379,8 +379,8 @@ def test_a_quadruple_example(
# https://github.com/AdnanHodzic/auto-cpufreq?tab=readme-ov-file#example-config-file-contents
#
# HENCE this below latency-headroom compensation logic..
from .conftest import cpu_scaling_factor
headroom: float = cpu_scaling_factor()
from .conftest import cpu_perf_headroom
headroom: float = cpu_perf_headroom()
if headroom != 1.:
this_fast = this_fast_on_linux * headroom
test_log.warning(

View File

@ -162,6 +162,66 @@ def test_implicit_mod_name_applied_for_child(
assert submod.log.logger in sub_logs
def test_io_custom_level_registered():
'''
The `IO`(21) level (registered via `add_log_level()` at
import, for `tractor.trionics._subproc`'s std-stream relay)
is fully wired and SHOWN BY DEFAULT at `info`-level consoles
since `21 >= INFO(20)`.
'''
import logging
assert log.CUSTOM_LEVELS.get('IO') == 21
assert logging.getLevelName(21) == 'IO'
assert log.STD_PALETTE.get('IO')
assert log.BOLD_PALETTE['bold'].get('IO')
iolog = log.get_logger('io_lvl_test')
assert callable(getattr(iolog, 'io', None))
# emit must not raise
iolog.io('hello from the IO level')
# 21 >= INFO(20) -> shown when console set to `info`
assert 21 >= logging.INFO
def test_add_log_level_pluggable():
'''
`add_log_level()` is the single pluggable entry-point: one
call wires `CUSTOM_LEVELS` + `addLevelName` + both palettes +
a same-named `StackLevelAdapter` emit method (so
`get_logger()`'s per-level audit passes).
'''
import logging
name: str = 'XLVL'
val: int = 19
try:
log.add_log_level(name, val, 'cyan')
assert log.CUSTOM_LEVELS[name] == val
assert logging.getLevelName(val) == name
assert log.STD_PALETTE[name] == 'cyan'
assert log.BOLD_PALETTE['bold'][name] == 'bold_cyan'
# the audit in `get_logger()` (asserts a method per
# `CUSTOM_LEVELS` entry) must still pass.
xlog = log.get_logger('xlvl_test')
emit = getattr(xlog, name.lower(), None)
assert callable(emit)
emit('hello from a plugged-in level')
finally:
# best-effort cleanup of our module-global mutations so
# later `get_logger()` audits don't see a half-removed
# level.
log.CUSTOM_LEVELS.pop(name, None)
log.STD_PALETTE.pop(name, None)
log.BOLD_PALETTE['bold'].pop(name, None)
if hasattr(log.StackLevelAdapter, name.lower()):
delattr(log.StackLevelAdapter, name.lower())
# TODO, moar tests against existing feats:
# ------ - ------
# - [ ] color settings?

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

@ -155,7 +155,6 @@ async def maybe_block_bp(
os.environ.pop('PYTHONBREAKPOINT', None)
@acm
async def open_root_actor(
*,
@ -186,6 +185,7 @@ async def open_root_actor(
# enables the multi-process debugger support
debug_mode: bool = False,
maybe_enable_greenback: bool = False, # `.pause_from_sync()/breakpoint()` support
# ^XXX NOTE^ the perf implications of use,
# https://greenback.readthedocs.io/en/latest/principle.html#performance
enable_stack_on_sig: bool = False,

View File

@ -90,7 +90,6 @@ keys are caller-defined).
'''
from __future__ import annotations
import os
import pathlib
import re
@ -99,6 +98,9 @@ import stat
import sys
import time
from tractor.devx import _proctitle
# `/dev/shm` is the POSIX-shm filesystem on Linux + FreeBSD.
# macOS uses `shm_open` syscalls without a fs-visible path,
# so the shm helpers refuse to run there.
@ -214,25 +216,25 @@ def _read_comm(pid: int) -> str:
# regardless of cwd / venv path / launch context. Used by
# `_is_tractor_subactor()` below.
#
# - cmdline `tractor[`: matches the `setproctitle`-set form
# (`tractor[<aid.reprol()>]`) — set in
# `_actor_child_main` for ALL backends, mutates argv via
# libc so visible in `/proc/<pid>/cmdline`.
# - cmdline `_subactor[` (`_proctitle._def_prefix`): matches
# the `setproctitle`-set form (`_subactor[<aid.reprol()>]`)
# — set in `_actor_child_main` for ALL backends, mutates
# argv via libc so visible in `/proc/<pid>/cmdline`.
# - cmdline `tractor._child`: matches the legacy
# `python -m tractor._child --uid (...)` form. Catches
# procs that died before `_actor_child_main` got to call
# `setproctitle()` — argv from exec is still kernel-
# visible at that point.
# - comm `tractor[`: same proctitle-set form, but visible
# - comm `_subactor[`: same proctitle-set form, but visible
# via `/proc/<pid>/comm` (kernel-truncated to ~15 bytes,
# `tractor[doggy:`). Critical for ZOMBIES — kernel
# `_subactor[doggy:`). Critical for ZOMBIES — kernel
# preserves `comm` past task-exit until parent reaps,
# while `cmdline` for zombies often reads as empty.
_TRACTOR_PROC_CMDLINE_MARKERS: tuple[str, ...] = (
'tractor._child',
'tractor[',
_proctitle._def_prefix,
)
_TRACTOR_PROC_COMM_MARKER: str = 'tractor['
_TRACTOR_PROC_COMM_MARKER: str = _proctitle._def_prefix
def _is_tractor_subactor(pid: int) -> bool:
@ -251,7 +253,7 @@ def _is_tractor_subactor(pid: int) -> bool:
'''
# 1. cmdline match — catches both `setproctitle`-set
# `tractor[<aid>]` (live) AND legacy `python -m
# `_subactor[<aid>]` (live) AND legacy `python -m
# tractor._child` (any) form.
cmdline: str = _read_cmdline(pid)
if any(m in cmdline for m in _TRACTOR_PROC_CMDLINE_MARKERS):

View File

@ -24,7 +24,10 @@ which" at a glance without needing to read full
`/proc/<pid>/cmdline`.
Format:
``tractor[<aid.reprol()>]`` e.g. ``tractor[doggy@1027301b]``
``<_def_prefix>[<aid.reprol()>]`` e.g. ``_subactor[doggy@1027301b]``
(prefix from the `_def_prefix` const, flipped `tractor` ->
`_subactor` so sub-actor procs are visually distinct from the
root in `ps`/`htop` and the reap-recognition markers.)
Uses the canonical `Aid.reprol()` form
(``<name>@<uuid_short>``) so the proc-title matches the
@ -52,7 +55,13 @@ except ImportError:
_stp = None
def set_actor_proctitle(actor: 'Actor') -> str | None:
_def_prefix: str = '_subactor'
def set_actor_proctitle(
actor: 'Actor',
prefix: str = _def_prefix,
) -> str | None:
'''
Set the calling process's proc-title to identify it as a
tractor sub-actor.
@ -69,6 +78,6 @@ def set_actor_proctitle(actor: 'Actor') -> str | None:
if _stp is None:
return None
title: str = f'tractor[{actor.aid.reprol()}]'
title: str = f'{prefix}[{actor.aid.reprol()}]'
_stp.setproctitle(title)
return title

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

@ -262,6 +262,68 @@ class StackLevelAdapter(LoggerAdapter):
)
def add_log_level(
name: str,
value: int,
color: str = 'white',
) -> None:
'''
Register a new custom log level with `tractor`'s logging
machinery in ONE call the single pluggable entry-point that
keeps the (otherwise hand-synced) pieces consistent:
- `CUSTOM_LEVELS[name]` (drives the `stacklevel` bump in
`StackLevelAdapter.log()` + the `get_logger()` audit).
- `logging.addLevelName()` registration.
- `STD_PALETTE`/`BOLD_PALETTE` color entries (consumed when
`get_console_log()` builds its `ColoredFormatter`).
- a same-named (lowercase) emit method bound on
`StackLevelAdapter` so `log.<name>('msg')` works (and so
`get_logger()`'s per-level method audit passes).
Idempotent: re-registering an existing name refreshes its
value + color in place. The bound `StackLevelAdapter.<name>`
method reads the current `CUSTOM_LEVELS` value at call time, so
it tracks the new level across re-registration (the method
object is bound once, never re-bound/clobbered).
'''
name_up: str = name.upper()
name_lo: str = name.lower()
CUSTOM_LEVELS[name_up] = value
logging.addLevelName(value, name_up)
STD_PALETTE[name_up] = color
BOLD_PALETTE['bold'][name_up] = f'bold_{color}'
if not hasattr(StackLevelAdapter, name_lo):
# read the level from `CUSTOM_LEVELS[name_up]` at CALL time
# (NOT captured at bind time) so a later
# `add_log_level(name, <new-value>)` re-registration stays
# consistent with this once-bound method. The closure binds
# *this* call's `name_up` local (stable, never reassigned),
# so there's no late-binding hazard. Delegates to `.log()`
# exactly like the hand-written level methods above.
def _emit(
self,
msg: str,
) -> None:
return self.log(CUSTOM_LEVELS[name_up], msg)
_emit.__name__ = name_lo
_emit.__qualname__ = f'StackLevelAdapter.{name_lo}'
setattr(StackLevelAdapter, name_lo, _emit)
# `IO`: child-subproc std-stream relay (see
# `tractor.trionics._subproc`). Value 21 sits just ABOVE
# `INFO`(20) so it's SHOWN BY DEFAULT at the usual `info`/`devx`
# console levels (a `runtime`(15) relay would be silently
# filtered) yet still distinctly labelled/colored + separately
# filterable.
add_log_level('IO', 21, 'purple')
# TODO IDEAs:
# -[ ] move to `.devx.pformat`?
# -[ ] do per task-name and actor-name color coding

View File

@ -139,7 +139,7 @@ _RUNTIME_VARS_DEFAULTS: dict[str, Any] = {
# `debug_mode: bool` settings
'_debug_mode': False, # bool
'repl_fixture': False, # |AbstractContextManager[bool]
'use_greenback': False, # `.pause_from_sync()`/`breakpoint()`
'use_stackscope': False, # trio-task-stack dumps on SIGUSR1

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

View File

@ -797,7 +797,7 @@ dev = [
{ name = "pytest-timeout", specifier = ">=2.3" },
{ name = "stackscope", specifier = ">=0.2.2,<0.3" },
{ name = "typing-extensions", specifier = ">=4.14.1" },
{ name = "xonsh", specifier = ">=0.23.0" },
{ name = "xonsh", specifier = ">=0.23.8" },
]
devx = [
{ name = "stackscope", specifier = ">=0.2.2,<0.3" },
@ -809,7 +809,7 @@ repl = [
{ name = "prompt-toolkit", specifier = ">=3.0.50" },
{ name = "psutil", specifier = ">=7.0.0" },
{ name = "pyperclip", specifier = ">=1.9.0" },
{ name = "xonsh", specifier = ">=0.23.0" },
{ name = "xonsh", specifier = ">=0.23.8" },
]
subints = [{ name = "msgspec", marker = "python_full_version >= '3.14'", specifier = ">=0.21.0" }]
sync-pause = [{ name = "greenback", marker = "python_full_version == '3.13.*'", specifier = ">=1.2.1,<2" }]

View File

@ -488,6 +488,7 @@ def _tractor_reap(args):
reap,
reap_shm,
reap_uds,
_TRACTOR_PROC_CMDLINE_MARKERS,
)
rc: int = 0
@ -500,9 +501,8 @@ def _tractor_reap(args):
else:
pids = find_orphans()
mode = (
'orphans (PPid==1, intrinsic '
'cmdline/comm match — `tractor[…]` or '
'`tractor._child`)'
f'orphans (PPid==1, intrinsic '
f'cmdline/comm match — {_TRACTOR_PROC_CMDLINE_MARKERS}'
)
if not pids: