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

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)
 

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:

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)
 

tractor/_testing/_reap.py

@@ -216,18 +216,18 @@ 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, ...] = (
@@ -253,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):

tractor/log.py

@@ -281,8 +281,11 @@ def add_log_level(
       `StackLevelAdapter` so `log.<name>('msg')` works (and so
       `get_logger()`'s per-level method audit passes).
 
-    Idempotent: re-registering an existing name is a no-op-ish
-    refresh (won't clobber an already-bound method).
+    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()
@@ -294,16 +297,18 @@ def add_log_level(
     BOLD_PALETTE['bold'][name_up] = f'bold_{color}'
 
     if not hasattr(StackLevelAdapter, name_lo):
-        # bind via default-arg so `value` is captured (not
-        # late-bound); delegates to `.log()` exactly like the
-        # hand-written level methods above.
+        # 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,
-            *,
-            _level: int = value,
         ) -> None:
-            return self.log(_level, msg)
+            return self.log(CUSTOM_LEVELS[name_up], msg)
 
         _emit.__name__ = name_lo
         _emit.__qualname__ = f'StackLevelAdapter.{name_lo}'

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(
@@ -140,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
@@ -158,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
@@ -190,7 +213,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
@@ -201,6 +231,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 +254,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
@@ -262,6 +316,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,
                 )
             )
 
@@ -287,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,