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,25 +279,4 @@ 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/hung-dump.xsh

@@ -1,89 +0,0 @@
-"""
-`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/devx/test_tooling.py

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

tests/discovery/test_multi_program.py

@@ -122,13 +122,7 @@ def test_register_duplicate_name(
                         'test_register_duplicate_name: '
                         '`wait_for_actor` returned'
                     )
-                    assert (
-                        portal.channel.aid.uid
-                        in (
-                            p2.channel.aid.uid,
-                            p1.channel.aid.uid,
-                        )
-                    )
+                    assert portal.channel.uid in (p2.channel.uid, p1.channel.uid)
 
                 log.cancel(
                     'test_register_duplicate_name: '
@@ -248,14 +242,8 @@ 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.aid.uid
-                        for p in portals
-                    }
-                    assert (
-                        portal.channel.aid.uid
-                        in expected_uids
-                    )
+                    expected_uids = {p.channel.uid for p in portals}
+                    assert portal.channel.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,10 +62,7 @@ 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.aid.uid
-    )
+    uid: tuple[str, str]|None = None if not actor else actor.uid
     print(f'{uid} ENC HOOK')
 
     match obj:
@@ -98,10 +95,7 @@ def dec_nsp(
     actor: Actor = tractor.current_actor(
         err_on_no_runtime=False,
     )
-    uid: tuple[str, str]|None = (
-        None if not actor
-        else actor.aid.uid
-    )
+    uid: tuple[str, str]|None = None if not actor else actor.uid
     print(
         f'{uid}\n'
         'CUSTOM DECODE\n'
@@ -425,8 +419,7 @@ async def send_back_values(
     and ensure we can round trip a func ref with our parent.
 
     '''
-    _aid = tractor.current_actor().aid
-    uid: tuple = _aid.uid
+    uid: tuple = tractor.current_actor().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.aid.uid}')
+            print(f'setting sub to {new_subs} for {ctx.chan.uid}')
 
             # remove old subs
             for sub in remove:
@@ -84,8 +84,7 @@ async def consumer(
     subs: list[str],
 ) -> None:
 
-    _aid = tractor.current_actor().aid
-    uid = _aid.uid
+    uid = tractor.current_actor().uid
 
     async with tractor.wait_for_actor('publisher') as portal:
         async with portal.open_context(subscribe) as (ctx, first):

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.cancel_called
+    assert n.cancelled
 
 
 @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.cancel_called is True
+        assert an.cancelled is True
         assert not an._children
     else:
         pytest.fail("Should have gotten a remote assertion error?")
@@ -821,28 +821,21 @@ 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 trio.open_nursery() as tn:
+            async with (
+
+                # XXX ?TODO? why no work!?
+                # tractor.trionics.collapse_eg(),
+                trio.open_nursery(
+                    strict_exception_groups=False,
+                ) as tn,
+            ):
                 await tn.start(spawn_and_sleep_forever)
                 if 'mp' in spawn_backend:
                     time.sleep(0.1)
                 os.kill(pid, signal.SIGINT)
 
-    # 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:
+    with pytest.raises(KeyboardInterrupt):
         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):

tests/test_context_stream_semantics.py

@@ -311,7 +311,7 @@ def test_parent_cancels(
         ctx: Context,
     ) -> None:
         actor: Actor = current_actor()
-        uid: tuple = actor.aid.uid
+        uid: tuple = actor.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().aid.uid
+                    current_actor().uid
                     ==
-                    root.aid.uid
+                    root.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().aid.uid
+                    assert tuple(res.canceller) == current_actor().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.aid.uid)}.file'
+    tmp_file: Path = tmp_path / f'{" ".join(ctx._actor.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.aid.uid
+                            assert ctxc.canceller == ctx.chan.uid
                             raise
 
                         except trio.TooSlowError:
@@ -1300,7 +1300,7 @@ async def caching_ep(
             },
 
             # lock around current actor task access
-            key=tractor.current_actor().aid.uid,
+            key=tractor.current_actor().uid,
 
         ) as (cache_hit, (clients, chan)),
     ):

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.aid.name == 'root'
+            assert portal.channel.uid[0] == 'root'
 
 
 @tractor_test

tests/test_spawning.py

@@ -65,7 +65,7 @@ async def spawn(
 
             assert len(an._children) == 1
             assert (
-                portal.channel.aid.uid
+                portal.channel.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[str, str] = current_actor().aid.uid
+    our_uid: tuple = current_actor().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.aid.uid}\n'
+                f'actor: {ctx.chan.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.aid.name}:{task}\n'
+            f'task: {actor.uid[0]}:{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.aid.uid}\n'
+                        f'actor: {ctx.chan.uid}\n'
                         f'ctx id: {ctx.cid}'
                     )
 

tractor/_testing/addr.py

@@ -72,12 +72,7 @@ def get_rando_addr(
             #   `test_tpt_bind_addrs::bind-subset-reg`),
             # - across parallel pytest sessions, the pid bias
             #   makes coincident port choices unlikely.
-            # 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 + (
+            port: int = 1000 + (
                 random.randint(0, 8999) + os.getpid()
             ) % 9000
             testrun_reg_addr = (

tractor/_testing/pytest.py

@@ -491,16 +491,6 @@ 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.aid.uid
+            ctx_in_debug.chan.uid
             if ctx_in_debug
             else None
         )
-        if in_debug == current_actor().aid.uid:
+        if in_debug == current_actor().uid:
             log.debug(
                 msg
                 +

tractor/experimental/_pubsub.py

@@ -38,7 +38,6 @@ import wrapt
 
 from ..log import get_logger
 from .._context import Context
-from ..msg import Yield
 
 
 __all__ = ['pub']
@@ -89,18 +88,7 @@ async def fan_out_to_ctxs(
             if ctx_payloads:
                 for ctx, payload in ctx_payloads:
                     try:
-                        # 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)
-                        )
+                        await ctx.send_yield(payload)
                     except (
                         # That's right, anything you can think of...
                         trio.ClosedResourceError, ConnectionResetError,
@@ -129,7 +117,7 @@ def modify_subs(
 
     Effectively a symbol subscription api.
     """
-    log.info(f"{ctx.chan.aid.uid} changed subscription to {topics}")
+    log.info(f"{ctx.chan.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.aid.uid
+                            ipc.chan.uid
                             if not is_started_send_side
-                            else ipc._actor.aid.uid
+                            else ipc._actor.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.aid.uid}\n'
+                        f'<= {ctx.chan.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.aid.uid}\n'
+                        f'<= {ctx.chan.uid}\n'
                         f'  |_{ctx._nsf}()\n\n'
                         f'=> {ctx._task}\n'
                         f'  |_{ctx._stream}\n\n'