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,18 +263,9 @@ def test_simple_context(
         except error_parent:
             pass
         except BaseExceptionGroup as beg:
-            # XXX: on windows — and under `trio>=0.33`'s strict
+            # XXX: on windows it seems we may have to expect the group error
-            # 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 (
+            assert is_multi_cancelled(beg)
-                is_multi_cancelled(beg)
-                or
-                beg.subgroup(error_parent) is not None
-            )
     else:
         trio.run(main)
 

tests/test_spawning.py

@@ -150,12 +150,7 @@ async def test_most_beautiful_word(
     The main ``tractor`` routine.
 
     '''
-    # actor spawn + IPC round-trip is comfortably sub-second on a
+    with trio.fail_after(1):
-    # 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,7 +14,6 @@ Hermetic `trio`-only coverage (no actor-runtime needed):
 
 '''
 from functools import partial
-import platform
 import subprocess
 
 import pytest
@@ -70,10 +69,6 @@ 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 `_subactor[` (`_proctitle._def_prefix`): matches
+# - cmdline `tractor[`: matches the `setproctitle`-set form
-#   the `setproctitle`-set form (`_subactor[<aid.reprol()>]`)
+#   (`tractor[<aid.reprol()>]`) — set in
-#   — set in `_actor_child_main` for ALL backends, mutates
+#   `_actor_child_main` for ALL backends, mutates argv via
-#   argv via libc so visible in `/proc/<pid>/cmdline`.
+#   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 `_subactor[`: same proctitle-set form, but visible
+# - comm `tractor[`: same proctitle-set form, but visible
 #   via `/proc/<pid>/comm` (kernel-truncated to ~15 bytes,
-#   `_subactor[doggy:`). Critical for ZOMBIES — kernel
+#   `tractor[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
-    #    `_subactor[<aid>]` (live) AND legacy `python -m
+    #    `tractor[<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,11 +281,8 @@ 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 refreshes its
+    Idempotent: re-registering an existing name is a no-op-ish
-    value + color in place. The bound `StackLevelAdapter.<name>`
+    refresh (won't clobber an already-bound method).
-    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()
@@ -297,18 +294,16 @@ def add_log_level(
     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
+        # bind via default-arg so `value` is captured (not
-        # (NOT captured at bind time) so a later
+        # late-bound); delegates to `.log()` exactly like the
-        # `add_log_level(name, <new-value>)` re-registration stays
+        # hand-written level methods above.
-        # 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(CUSTOM_LEVELS[name_up], msg)
+            return self.log(_level, msg)
 
         _emit.__name__ = name_lo
         _emit.__qualname__ = f'StackLevelAdapter.{name_lo}'

tractor/trionics/_subproc.py

@@ -42,22 +42,6 @@ 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,
@@ -81,7 +65,6 @@ 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`
@@ -112,21 +95,15 @@ 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:
-                emit(f'[{tag}] {_decode_line(raw)}')
+                line: str = raw.decode(
+                    errors='replace',
+                ).rstrip('\r')
+                emit(f'[{tag}] {line}')
 
     # flush any trailing partial (un-newline-term'd) line @ EOF
     if (
@@ -134,7 +111,10 @@ async def _relay_stream_lines(
         and
         residual
     ):
-        emit(f'[{tag}] {_decode_line(residual)}')
+        line: str = residual.decode(
+            errors='replace',
+        ).rstrip('\r')
+        emit(f'[{tag}] {line}')
 
 
 async def supervise_run_process(
@@ -160,7 +140,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|None = _UNSET,
+    stdout: int = _UNSET,
 
     task_status: trio.TaskStatus[
         trio.Process
@@ -178,14 +158,11 @@ 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 the
+      do our OWN post-drain rc-check, (re)building + raising a
-      CPE (with a `.stderr` note) from this coro's body AFTER the
+      BARE CPE (with a `.stderr` note) from this coro's body
-      child exits — so it's never wrapped by `trio.run_process`'s
+      AFTER the child exits — so there's no nursery-eg-wrapped
-      INTERNAL nursery nor race-cancelled mid-drain. It IS still
+      CPE to catch/`collapse_eg`, and the relay reader is never
-      raised as a task into the *parent* `tn`, so — like any
+      race-cancelled mid-drain.
-      `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
@@ -213,14 +190,7 @@ async def supervise_run_process(
             )
 
     '''
-    # resolve the relay emit-method ONLY when actually relaying so
+    emit: Callable[[str], None] = getattr(log, relay_level)
-    # 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
@@ -231,20 +201,6 @@ 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
 
@@ -254,16 +210,6 @@ 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
@@ -316,7 +262,6 @@ 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,
                 )
             )
 
@@ -342,10 +287,6 @@ 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,