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

pyproject.toml

@@ -279,4 +279,25 @@ log_cli = false
 
 # https://docs.pytest.org/en/stable/reference/reference.html#confval-console_output_style
 console_output_style = 'progress'
+
+# `pexpect` (used only by the `devx` debugger REPL tests) allocates
+# its child pty via stdlib `pty.py:os.forkpty()`, which CPython
+# 3.12+ flags as unsafe in a multi-threaded process. This is NOT
+# ours to fix at source — we never call `forkpty()`; pexpect does.
+# UPSTREAM to actually resolve: pexpect would need to avoid
+# `forkpty()` under multithreading (or perform the pty spawn before
+# any thread starts); until then this single third-party warning is
+# the one we filter rather than fix.
+filterwarnings = [
+  'ignore:This process \(pid=\d+\) is multi-threaded, use of forkpty\(\):DeprecationWarning',
+
+  # `tests/test_legacy_one_way_streaming.py` exists *specifically*
+  # to exercise the DEPRECATED `@tractor.stream` async-gen API, so
+  # its decoration-time `DeprecationWarning` is expected — and since
+  # it fires at import (module-level decorators) it can't be caught
+  # by a per-test mark, hence this test-wide ini filter. Real
+  # library users still see the warning; drop this once the legacy
+  # `@tractor.stream` / `Portal.open_stream_from()` API is removed.
+  'ignore:`@tractor.stream decorated funcs:DeprecationWarning',
+]
 # ------ tool.pytest ------

scripts/cpu-perf-check

@@ -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())

scripts/hung-dump.xsh

@@ -0,0 +1,89 @@
+"""
+`hung-dump`: snapshot diagnostic state for a hung
+`pytest`/`tractor` process tree.
+
+For each pid (and all `pgrep -P` descendants), prints
+- `ps` forest header,
+- `/proc/<pid>/wchan` + `/proc/<pid>/stack` (kernel-side
+  blocked-on syscall),
+- `sudo py-spy dump` (python-side stack).
+
+Usage (xonsh):
+    source-foreign xonsh ./scripts/hung-dump.xsh
+    hung-dump <pid> [<pid> ...]
+
+Or just:
+    source ./scripts/hung-dump.xsh
+
+Tip — pipe to a paste buffer:
+    hung-dump 1336765 |t /tmp/hung.log
+"""
+
+def _hung_dump(args):
+    import subprocess as sp
+    from pathlib import Path
+
+    if not args:
+        print('usage: hung-dump <pid> [<pid> ...]')
+        return 1
+
+    def kids(pid):
+        try:
+            out = sp.check_output(
+                ['pgrep', '-P', str(pid)],
+                text=True,
+            )
+        except sp.CalledProcessError:
+            return []
+        return [int(p) for p in out.split() if p]
+
+    def tree(pid):
+        out = [pid]
+        for k in kids(pid):
+            out.extend(tree(k))
+        return out
+
+    pids = [
+        p
+        for r in (int(a) for a in args)
+        for p in tree(r)
+    ]
+
+    print(f'# tree: {pids}')
+    print('\n## ps forest')
+    $[ps -o pid,ppid,pgid,stat,cmd -p @(','.join(map(str, pids)))]
+
+    for p in pids:
+        print(f'\n## pid {p}')
+        for f in ('wchan', 'stack'):
+            path = Path(f'/proc/{p}/{f}')
+            try:
+                txt = path.read_text().rstrip()
+                print(f'-- /proc/{p}/{f} --\n{txt}')
+            except PermissionError:
+                # `stack` requires CAP_SYS_PTRACE; retry
+                # via sudo -n so creds-cached users don't
+                # block on prompt.
+                try:
+                    txt = sp.check_output(
+                        ['sudo', '-n', 'cat', str(path)],
+                        text=True,
+                        stderr=sp.DEVNULL,
+                    ).rstrip()
+                    print(f'-- /proc/{p}/{f} (via sudo) --\n{txt}')
+                except sp.CalledProcessError:
+                    print(
+                        f'-- /proc/{p}/{f}: '
+                        'PermissionError (try `sudo true` first) --'
+                    )
+            except FileNotFoundError:
+                print(f'-- /proc/{p}/{f}: proc gone --')
+
+        print(f'-- py-spy {p} --')
+        try:
+            $[sudo -n py-spy dump --pid @(p)]
+        except Exception as e:
+            print(f'  (py-spy failed: {e})')
+
+
+aliases['hung-dump'] = _hung_dump

tests/conftest.py

@@ -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

tests/devx/test_debugger.py

@@ -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,

tests/devx/test_tooling.py

@@ -92,7 +92,7 @@ def test_shield_pause(
     expect(
         child,
         # end-of-tree delimiter
-        "end-of-\('root'",
+        r"end-of-\('root'",
     )
     _before: str = assert_before(
         child,
@@ -157,7 +157,7 @@ def test_shield_pause(
             expect(
                 child,
                 # end-of-subactor's-tree delimiter
-                "end-of-\('hanger'",
+                r"end-of-\('hanger'",
             )
             _before: str = assert_before(
                 child,

tests/discovery/test_multi_program.py

@@ -122,7 +122,13 @@ def test_register_duplicate_name(
                         'test_register_duplicate_name: '
                         '`wait_for_actor` returned'
                     )
-                    assert portal.channel.uid in (p2.channel.uid, p1.channel.uid)
+                    assert (
+                        portal.channel.aid.uid
+                        in (
+                            p2.channel.aid.uid,
+                            p1.channel.aid.uid,
+                        )
+                    )
 
                 log.cancel(
                     'test_register_duplicate_name: '
@@ -242,8 +248,14 @@ def test_dup_name_cancel_cascade_escalates_to_hard_kill(
                 # name; doesn't matter which one (registrar will
                 # have last-wins semantics under same-name).
                 async with tractor.wait_for_actor('doggy') as portal:
-                    expected_uids = {p.channel.uid for p in portals}
-                    assert portal.channel.uid in expected_uids
+                    expected_uids = {
+                        p.channel.aid.uid
+                        for p in portals
+                    }
+                    assert (
+                        portal.channel.aid.uid
+                        in expected_uids
+                    )
 
                 # critical section: this MUST return within
                 # `fail_after_s` even when one or more cancel-RPC

tests/msg/test_ext_types_msgspec.py

@@ -62,7 +62,10 @@ def enc_nsp(obj: Any) -> Any:
     actor: Actor = tractor.current_actor(
         err_on_no_runtime=False,
     )
-    uid: tuple[str, str]|None = None if not actor else actor.uid
+    uid: tuple[str, str]|None = (
+        None if not actor
+        else actor.aid.uid
+    )
     print(f'{uid} ENC HOOK')
 
     match obj:
@@ -95,7 +98,10 @@ def dec_nsp(
     actor: Actor = tractor.current_actor(
         err_on_no_runtime=False,
     )
-    uid: tuple[str, str]|None = None if not actor else actor.uid
+    uid: tuple[str, str]|None = (
+        None if not actor
+        else actor.aid.uid
+    )
     print(
         f'{uid}\n'
         'CUSTOM DECODE\n'
@@ -419,7 +425,8 @@ async def send_back_values(
     and ensure we can round trip a func ref with our parent.
 
     '''
-    uid: tuple = tractor.current_actor().uid
+    _aid = tractor.current_actor().aid
+    uid: tuple = _aid.uid
 
     # init state in sub-actor should be default
     chk_codec_applied(

tests/test_advanced_streaming.py

@@ -69,7 +69,7 @@ async def subscribe(
             new_subs = set(new_subs)
             remove = new_subs - _registry.keys()
 
-            print(f'setting sub to {new_subs} for {ctx.chan.uid}')
+            print(f'setting sub to {new_subs} for {ctx.chan.aid.uid}')
 
             # remove old subs
             for sub in remove:
@@ -84,7 +84,8 @@ async def consumer(
     subs: list[str],
 ) -> None:
 
-    uid = tractor.current_actor().uid
+    _aid = tractor.current_actor().aid
+    uid = _aid.uid
 
     async with tractor.wait_for_actor('publisher') as portal:
         async with portal.open_context(subscribe) as (ctx, first):
@@ -188,11 +189,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

tests/test_cancellation.py

@@ -351,7 +351,7 @@ async def test_cancel_infinite_streamer(
 
     # we support trio's cancellation system
     assert cancel_scope.cancelled_caught
-    assert n.cancelled
+    assert n.cancel_called
 
 
 @pytest.mark.parametrize(
@@ -465,7 +465,7 @@ async def test_some_cancels_all(
         elif isinstance(err, tractor.RemoteActorError):
             assert err.boxed_type == err_type
 
-        assert an.cancelled is True
+        assert an.cancel_called is True
         assert not an._children
     else:
         pytest.fail("Should have gotten a remote assertion error?")
@@ -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
 
@@ -780,21 +821,28 @@ def test_cancel_via_SIGINT_other_task(
     async def main():
         # should never timeout since SIGINT should cancel the current program
         with trio.fail_after(timeout):
-            async with (
-
-                # XXX ?TODO? why no work!?
-                # tractor.trionics.collapse_eg(),
-                trio.open_nursery(
-                    strict_exception_groups=False,
-                ) as tn,
-            ):
+            async with trio.open_nursery() as tn:
                 await tn.start(spawn_and_sleep_forever)
                 if 'mp' in spawn_backend:
                     time.sleep(0.1)
                 os.kill(pid, signal.SIGINT)
 
-    with pytest.raises(KeyboardInterrupt):
+    # SIGINT -> `KeyboardInterrupt`; under `trio>=0.25`'s strict
+    # exception-groups the KI surfaces wrapped in a (cancel-padded)
+    # `BaseExceptionGroup` rather than bare — so accept either form
+    # (replaces the now-deprecated `strict_exception_groups=False`,
+    # and `collapse_eg()` can't help since the group is multi-exc:
+    # the KI rides alongside the child-task `Cancelled`s).
+    with pytest.raises(BaseException) as excinfo:
         trio.run(main)
+    exc = excinfo.value
+    assert (
+        isinstance(exc, KeyboardInterrupt)
+        or (
+            isinstance(exc, BaseExceptionGroup)
+            and exc.subgroup(KeyboardInterrupt) is not None
+        )
+    )
 
 
 async def spin_for(period=3):
@@ -955,11 +1003,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()

tests/test_clustering.py

@@ -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,

tests/test_context_stream_semantics.py

@@ -311,7 +311,7 @@ def test_parent_cancels(
         ctx: Context,
     ) -> None:
         actor: Actor = current_actor()
-        uid: tuple = actor.uid
+        uid: tuple = actor.aid.uid
         _ctxc: ContextCancelled|None = None
 
         if (
@@ -712,9 +712,9 @@ async def test_parent_exits_ctx_after_child_enters_stream(
                 assert (
                     ctxc.canceller
                     ==
-                    current_actor().uid
+                    current_actor().aid.uid
                     ==
-                    root.uid
+                    root.aid.uid
                 )
 
                 # channel should still be up
@@ -1219,7 +1219,7 @@ def test_maybe_allow_overruns_stream(
 
                 if cancel_ctx:
                     assert isinstance(res, ContextCancelled)
-                    assert tuple(res.canceller) == current_actor().uid
+                    assert tuple(res.canceller) == current_actor().aid.uid
 
                 else:
                     print(f'RX ROOT SIDE RESULT {res}')

tests/test_infected_asyncio.py

@@ -956,7 +956,7 @@ async def manage_file(
     '''
 
     tmp_path: Path = Path(tmp_path_str)
-    tmp_file: Path = tmp_path / f'{" ".join(ctx._actor.uid)}.file'
+    tmp_file: Path = tmp_path / f'{" ".join(ctx._actor.aid.uid)}.file'
 
     # create a the tmp file and tell the parent where it's at
     assert not tmp_file.is_file()
@@ -1180,7 +1180,7 @@ def test_sigint_closes_lifetime_stack(
                             ):
                                 await ctx.wait_for_result()
                         except tractor.ContextCancelled as ctxc:
-                            assert ctxc.canceller == ctx.chan.uid
+                            assert ctxc.canceller == ctx.chan.aid.uid
                             raise
 
                         except trio.TooSlowError:
@@ -1300,7 +1300,7 @@ async def caching_ep(
             },
 
             # lock around current actor task access
-            key=tractor.current_actor().uid,
+            key=tractor.current_actor().aid.uid,
 
         ) as (cache_hit, (clients, chan)),
     ):

tests/test_legacy_one_way_streaming.py

@@ -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(

tests/test_local.py

@@ -34,7 +34,7 @@ async def test_self_is_registered(reg_addr):
     assert actor.is_registrar
     with trio.fail_after(0.2):
         async with tractor.wait_for_actor('root') as portal:
-            assert portal.channel.uid[0] == 'root'
+            assert portal.channel.aid.name == 'root'
 
 
 @tractor_test

tests/test_spawning.py

@@ -65,7 +65,7 @@ async def spawn(
 
             assert len(an._children) == 1
             assert (
-                portal.channel.uid
+                portal.channel.aid.uid
                 in
                 tractor.current_actor().ipc_server._peers
             )

tractor/_exceptions.py

@@ -1223,7 +1223,7 @@ def pack_error(
         str,
         str | tuple[str, str]
     ] = {}
-    our_uid: tuple = current_actor().uid
+    our_uid: tuple[str, str] = current_actor().aid.uid
 
     if (
         isinstance(exc, RemoteActorError)

tractor/_streaming.py

@@ -418,7 +418,7 @@ class MsgStream(trio.abc.Channel):
             # it can't traverse the transport.
             log.warning(
                 f'Stream was already destroyed?\n'
-                f'actor: {ctx.chan.uid}\n'
+                f'actor: {ctx.chan.aid.uid}\n'
                 f'ctx id: {ctx.cid}'
             )
             drained.append(re)
@@ -700,7 +700,7 @@ async def open_stream_from_ctx(
         task: str = trio.lowlevel.current_task().name
         raise RuntimeError(
             'Stream opened after `Context.cancel()` called..?\n'
-            f'task: {actor.uid[0]}:{task}\n'
+            f'task: {actor.aid.name}:{task}\n'
             f'{ctx}'
         )
 
@@ -823,7 +823,7 @@ async def open_stream_from_ctx(
                 except KeyError:
                     log.warning(
                         f'Stream was already destroyed?\n'
-                        f'actor: {ctx.chan.uid}\n'
+                        f'actor: {ctx.chan.aid.uid}\n'
                         f'ctx id: {ctx.cid}'
                     )
 

tractor/_testing/addr.py

@@ -72,7 +72,12 @@ def get_rando_addr(
             #   `test_tpt_bind_addrs::bind-subset-reg`),
             # - across parallel pytest sessions, the pid bias
             #   makes coincident port choices unlikely.
-            port: int = 1000 + (
+            # NB. floor at 1024, NOT 1000: ports <1024 are PRIVILEGED
+            # on linux (`net.ipv4.ip_unprivileged_port_start = 1024`),
+            # so a non-root `.bind()` on one raises `PermissionError`.
+            # The old `1000 +` floor could roll 1000-1023 -> flaky
+            # `[Errno 13] Permission denied` registry-listener binds.
+            port: int = 1024 + (
                 random.randint(0, 8999) + os.getpid()
             ) % 9000
             testrun_reg_addr = (

tractor/_testing/pytest.py

@@ -491,6 +491,16 @@ def pytest_configure(
         'cases (e.g. the `subint` GIL-starvation class documented '
         'in `ai/conc-anal/subint_sigint_starvation_issue.md`).'
     )
+    config.addinivalue_line(
+        'markers',
+        'has_nested_actors: test spawns nested (>1-level) subactor '
+        'trees.'
+    )
+    config.addinivalue_line(
+        'markers',
+        'trio: legacy mark for tests meant to run under the `trio` '
+        'spawn backend (e.g. `test_local.py`).'
+    )
 
     # `--enable-stackscope`: install SIGUSR1 → trio task-tree
     # dump in pytest itself + propagate to every subactor via

tractor/devx/debug/_sync.py

@@ -92,11 +92,11 @@ async def maybe_wait_for_debugger(
         # tearing down.
         ctx_in_debug: Context|None = Lock.ctx_in_debug
         in_debug: tuple[str, str]|None = (
-            ctx_in_debug.chan.uid
+            ctx_in_debug.chan.aid.uid
             if ctx_in_debug
             else None
         )
-        if in_debug == current_actor().uid:
+        if in_debug == current_actor().aid.uid:
             log.debug(
                 msg
                 +

tractor/experimental/_pubsub.py

@@ -38,6 +38,7 @@ import wrapt
 
 from ..log import get_logger
 from .._context import Context
+from ..msg import Yield
 
 
 __all__ = ['pub']
@@ -88,7 +89,18 @@ async def fan_out_to_ctxs(
             if ctx_payloads:
                 for ctx, payload in ctx_payloads:
                     try:
-                        await ctx.send_yield(payload)
+                        # NOTE: inline the (now-deprecated)
+                        # `Context.send_yield()` body to drop its
+                        # `DeprecationWarning` without a behaviour change.
+                        # The "proper" fix is migrating BOTH sides to
+                        # `open_context()`/`open_stream()`, but the
+                        # subscriber still uses the legacy
+                        # `Portal.open_stream_from()` (no `started()`
+                        # handshake) so `Context.open_stream()` can't be
+                        # used here yet — see this module's rework TODO.
+                        await ctx.chan.send(
+                            Yield(cid=ctx.cid, pld=payload)
+                        )
                     except (
                         # That's right, anything you can think of...
                         trio.ClosedResourceError, ConnectionResetError,
@@ -117,7 +129,7 @@ def modify_subs(
 
     Effectively a symbol subscription api.
     """
-    log.info(f"{ctx.chan.uid} changed subscription to {topics}")
+    log.info(f"{ctx.chan.aid.uid} changed subscription to {topics}")
 
     # update map from each symbol to requesting client's chan
     for topic in topics:

tractor/msg/_ops.py

@@ -345,9 +345,9 @@ class PldRx(Struct):
                         exc=mte,
                         cid=msg.cid,
                         src_uid=(
-                            ipc.chan.uid
+                            ipc.chan.aid.uid
                             if not is_started_send_side
-                            else ipc._actor.uid
+                            else ipc._actor.aid.uid
                         ),
                     )
                     mte._ipc_msg = err_msg
@@ -711,7 +711,7 @@ async def drain_to_final_msg(
                         'Cancelling `MsgStream` drain since '
                         f'{reason}\n'
                         f'\n'
-                        f'<= {ctx.chan.uid}\n'
+                        f'<= {ctx.chan.aid.uid}\n'
                         f'  |_{ctx._nsf}()\n'
                         f'\n'
                         f'=> {ctx._task}\n'
@@ -726,7 +726,7 @@ async def drain_to_final_msg(
                 else:
                     report: str = (
                         'Ignoring "yield" msg during `ctx.result()` drain..\n'
-                        f'<= {ctx.chan.uid}\n'
+                        f'<= {ctx.chan.aid.uid}\n'
                         f'  |_{ctx._nsf}()\n\n'
                         f'=> {ctx._task}\n'
                         f'  |_{ctx._stream}\n\n'