Port `test_registrar` off `run_in_actor`

Two sites migrated (#477 removal),

- `test_trynamic_trio`: donny + gretchen each wait on the *other*
  to register, so they must run CONCURRENTLY — was two
  non-blocking `run_in_actor()`s awaited after; now two
  `to_actor.run()` one-shots scheduled into a local `trio`
  task-nursery.
- the unregister-on-cancel cluster test: its non-streaming branch
  spawned `run_in_actor(trio.sleep_forever)` purely to keep each
  subactor alive + registered — a `start_actor()` daemon does that
  without a "main" task, so the spawn loop collapses to the same
  `start_actor()` the streaming branch already used.

Suite: 16 passed.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

tests/discovery/test_registrar.py

@@ -152,23 +152,27 @@ async def test_trynamic_trio(
     for the directed subs.
 
     '''
-    async with tractor.open_nursery() as n:
+    async with (
+        tractor.open_nursery() as n,
+        trio.open_nursery() as tn,
+    ):
         print("Alright... Action!")
 
-        donny = await n.run_in_actor(
+        # donny + gretchen each wait on the *other* to register, so
+        # they must run CONCURRENTLY — schedule both one-shots into a
+        # local task-nursery (was two non-blocking `run_in_actor()`s).
+        async def _direct(this_name: str, other_actor: str):
+            res = await tractor.to_actor.run(
                 ria_fn,
-            other_actor='gretchen',
+                an=n,
+                other_actor=other_actor,
                 reg_addr=reg_addr,
-            name='donny',
+                name=this_name,
             )
-        gretchen = await n.run_in_actor(
-            ria_fn,
-            other_actor='donny',
-            reg_addr=reg_addr,
-            name='gretchen',
-        )
-        print(await gretchen.result())
-        print(await donny.result())
+            print(res)
+
+        tn.start_soon(_direct, 'donny', 'gretchen')
+        tn.start_soon(_direct, 'gretchen', 'donny')
         print("CUTTTT CUUTT CUT!!?! Donny!! You're supposed to say...")
 
 
@@ -270,13 +274,15 @@ async def spawn_and_check_registry(
                         portals = {}
                         for i in range(3):
                             name = f'a{i}'
-                            if with_streaming:
+                            # a daemon subactor is alive + registered
+                            # without a "main" task; the streaming
+                            # branch below uses the module funcs, the
+                            # non-streaming case just needs it up (was
+                            # `run_in_actor(trio.sleep_forever)`).
                             portals[name] = await an.start_actor(
-                                    name=name, enable_modules=[__name__])
-
-                            else:  # no streaming
-                                portals[name] = await an.run_in_actor(
-                                    trio.sleep_forever, name=name)
+                                name=name,
+                                enable_modules=[__name__],
+                            )
 
                         # wait on last actor to come up
                         async with tractor.wait_for_actor(name):

tests/test_infected_asyncio.py

@@ -24,6 +24,7 @@ from tractor import (
     current_actor,
     Actor,
     to_asyncio,
+    to_actor,
     RemoteActorError,
     ContextCancelled,
 )
@@ -110,8 +111,9 @@ def test_trio_cancels_aio_on_actor_side(
                 registry_addrs=[reg_addr],
                 debug_mode=debug_mode,
             ) as an:
-                await an.run_in_actor(
+                await to_actor.run(
                     trio_cancels_single_aio_task,
+                    an=an,
                     infect_asyncio=True,
                 )
 
@@ -157,6 +159,28 @@ async def asyncio_actor(
         raise
 
 
+@tractor.context
+async def sleep_forever_aio_ctx(
+    ctx: tractor.Context,
+    expect_err: str = 'trio.Cancelled',
+) -> None:
+    '''
+    `@context` shim so a parent can spawn a forever-sleeping
+    infected-`asyncio` task via `Portal.open_context()` and cancel it
+    (via `Portal.cancel_actor()` or an enclosing `trio` cancel scope),
+    asserting the graceful `trio.Cancelled` teardown.
+
+    Replaces the legacy `ActorNursery.run_in_actor()` spawn the
+    aio-cancel tests below used to rely on (removed with #477).
+
+    '''
+    await ctx.started()
+    await asyncio_actor(
+        target='aio_sleep_forever',
+        expect_err=expect_err,
+    )
+
+
 def test_aio_simple_error(
     reg_addr: tuple[str, int],
     debug_mode: bool,
@@ -172,8 +196,9 @@ def test_aio_simple_error(
             registry_addrs=[reg_addr],
             debug_mode=debug_mode,
         ) as an:
-            await an.run_in_actor(
+            await to_actor.run(
                 asyncio_actor,
+                an=an,
                 target='sleep_and_err',
                 expect_err='AssertionError',
                 infect_asyncio=True,
@@ -207,18 +232,34 @@ def test_tractor_cancels_aio(
 
     '''
     async def main():
+        # anti-hang wall-clock cap: a per-test `trio.fail_after`
+        # is the blessed guard here since `pytest-timeout`'s
+        # global cap is intentionally off (see the `pyproject`
+        # NOTE — it breaks trio under fork backends). Generous +
+        # CPU-headroom-scaled bc this is an anti-hang guard, not
+        # a perf assertion; a wedged ria-reaper once hung this
+        # test forever (the `._ria_nursery`-removal regression).
+        from .conftest import cpu_perf_headroom
+        with trio.fail_after(9 * cpu_perf_headroom()):
             async with tractor.open_nursery(
                 debug_mode=debug_mode,
                 registry_addrs=[reg_addr],
             ) as an:
-            portal = await an.run_in_actor(
-                asyncio_actor,
-                target='aio_sleep_forever',
-                expect_err='trio.Cancelled',
+                p: tractor.Portal = await an.start_actor(
+                    'aio_daemon',
+                    enable_modules=[__name__],
                     infect_asyncio=True,
                 )
-            # cancel the entire remote runtime
-            await portal.cancel_actor()
+                async with (
+                    # `.cancel_actor()` below tears the ctx down
+                    expect_ctxc(yay=True),
+                    p.open_context(
+                        sleep_forever_aio_ctx,
+                    ) as (ctx, first),
+                ):
+                    # cancel the entire remote runtime while its
+                    # infected-`asyncio` task sleeps forever
+                    await p.cancel_actor()
 
     trio.run(main)
 
@@ -236,13 +277,19 @@ def test_trio_cancels_aio(
         with trio.move_on_after(1):
             async with tractor.open_nursery(
                 registry_addrs=[reg_addr],
-            ) as tn:
-                await tn.run_in_actor(
-                    asyncio_actor,
-                    target='aio_sleep_forever',
-                    expect_err='trio.Cancelled',
+            ) as an:
+                p: tractor.Portal = await an.start_actor(
+                    'aio_daemon',
+                    enable_modules=[__name__],
                     infect_asyncio=True,
                 )
+                async with p.open_context(
+                    sleep_forever_aio_ctx,
+                ) as (ctx, first):
+                    # block until the enclosing `move_on_after`
+                    # cancels this `trio` scope, tearing down the
+                    # infected-aio task via ctx cancellation
+                    await trio.sleep_forever()
 
     trio.run(main)
 
@@ -392,17 +439,16 @@ def test_aio_cancelled_from_aio_causes_trio_cancelled(
         async with tractor.open_nursery(
             registry_addrs=[reg_addr],
         ) as an:
-            p: tractor.Portal = await an.run_in_actor(
+            # `to_actor.run()` blocks on the one-shot's result and
+            # relays the remote error here in the caller's task.
+            with trio.fail_after(1 + delay):
+                await to_actor.run(
                     asyncio_actor,
+                    an=an,
                     target='aio_cancel',
                     expect_err='tractor.to_asyncio.AsyncioCancelled',
                     infect_asyncio=True,
                 )
-            # NOTE: normally the `an.__aexit__()` waits on the
-            # portal's result but we do it explicitly here
-            # to avoid indent levels.
-            with trio.fail_after(1 + delay):
-                await p.wait_for_result()
 
     with pytest.raises(
         expected_exception=(RemoteActorError, ExceptionGroup),
@@ -603,13 +649,13 @@ def test_basic_interloop_channel_stream(
             async with tractor.open_nursery(
                 registry_addrs=[reg_addr],
             ) as an:
-                portal = await an.run_in_actor(
+                # should raise RAE diectly
+                await to_actor.run(
                     stream_from_aio,
+                    an=an,
                     infect_asyncio=True,
                     fan_out=fan_out,
                 )
-                # should raise RAE diectly
-                await portal.result()
 
     trio.run(main)
 
@@ -622,13 +668,13 @@ def test_trio_error_cancels_intertask_chan(
         async with tractor.open_nursery(
             registry_addrs=[reg_addr],
         ) as an:
-            portal = await an.run_in_actor(
+            # should trigger remote actor error
+            await to_actor.run(
                 stream_from_aio,
+                an=an,
                 trio_raise_err=True,
                 infect_asyncio=True,
             )
-            # should trigger remote actor error
-            await portal.result()
 
     with pytest.raises(RemoteActorError) as excinfo:
         trio.run(main)
@@ -658,14 +704,14 @@ def test_trio_closes_early_causes_aio_checkpoint_raise(
                 # enable_stack_on_sig=True,
                 registry_addrs=[reg_addr],
             ) as an:
-                portal = await an.run_in_actor(
+                # should raise RAE diectly
+                print('waiting on final infected subactor result..')
+                res: None = await to_actor.run(
                     stream_from_aio,
+                    an=an,
                     trio_exit_early=True,
                     infect_asyncio=True,
                 )
-                # should raise RAE diectly
-                print('waiting on final infected subactor result..')
-                res: None = await portal.wait_for_result()
                 assert res is None
                 print(f'infected subactor returned result: {res!r}\n')
 
@@ -709,15 +755,15 @@ def test_aio_exits_early_relays_AsyncioTaskExited(
                 debug_mode=debug_mode,
                 # enable_stack_on_sig=True,
             ) as an:
-                portal = await an.run_in_actor(
+                # should raise RAE diectly
+                print('waiting on final infected subactor result..')
+                res: None = await to_actor.run(
                     stream_from_aio,
+                    an=an,
                     infect_asyncio=True,
                     trio_exit_early=False,
                     aio_exit_early=True,
                 )
-                # should raise RAE diectly
-                print('waiting on final infected subactor result..')
-                res: None = await portal.wait_for_result()
                 assert res is None
                 print(f'infected subactor returned result: {res!r}\n')
 
@@ -749,17 +795,19 @@ def test_aio_errors_and_channel_propagates_and_closes(
             registry_addrs=[reg_addr],
             debug_mode=debug_mode,
         ) as an:
-            portal = await an.run_in_actor(
+            # should trigger RAE directly, not an eg.
+            await to_actor.run(
                 stream_from_aio,
+                an=an,
                 aio_raise_err=True,
                 infect_asyncio=True,
             )
-            # should trigger RAE directly, not an eg.
-            await portal.result()
 
     with pytest.raises(
-        # NOTE: bc we directly wait on `Portal.result()` instead
-        # of capturing it inside the `ActorNursery` machinery.
+        # NOTE: bc `to_actor.run()` blocks on + relays the result
+        # in the caller's task (not captured inside the
+        # `ActorNursery` teardown machinery) we get a direct RAE,
+        # not an eg.
         expected_exception=RemoteActorError,
     ) as excinfo:
         trio.run(main)

tests/test_pubsub.py

@@ -176,10 +176,13 @@ def test_multi_actor_subs_arbiter_pub(
 
     async def main():
 
-        async with tractor.open_nursery(
+        async with (
+            tractor.open_nursery(
                 registry_addrs=[reg_addr],
                 enable_modules=[__name__],
-        ) as n:
+            ) as n,
+            trio.open_nursery() as tn,
+        ):
 
             name = 'root'
 
@@ -191,18 +194,37 @@ def test_multi_actor_subs_arbiter_pub(
                 )
                 name = 'streamer'
 
-            even_portal = await n.run_in_actor(
+            # spawn the two subscriber actors as daemons and run
+            # `subs()` on each as a background task (was the legacy
+            # `run_in_actor()`); keep the portals for the explicit
+            # `cancel_actor()` teardown below. Each runner swallows
+            # the teardown error that `cancel_actor()` relays.
+            async def _run_subs(
+                portal: tractor.Portal,
+                which: list[str],
+            ) -> None:
+                try:
+                    await portal.run(
                         subs,
-                which=['even'],
-                name='evens',
-                pub_actor_name=name
+                        which=which,
+                        pub_actor_name=name,
                     )
-            odd_portal = await n.run_in_actor(
-                subs,
-                which=['odd'],
-                name='odds',
-                pub_actor_name=name
+                except (
+                    tractor.RemoteActorError,
+                    tractor.ContextCancelled,
+                ):
+                    pass  # expected once we `cancel_actor()` below
+
+            even_portal = await n.start_actor(
+                'evens',
+                enable_modules=[__name__],
             )
+            odd_portal = await n.start_actor(
+                'odds',
+                enable_modules=[__name__],
+            )
+            tn.start_soon(_run_subs, even_portal, ['even'])
+            tn.start_soon(_run_subs, odd_portal, ['odd'])
 
             async with tractor.wait_for_actor('evens'):
                 # block until 2nd actor is initialized
@@ -257,6 +279,9 @@ def test_multi_actor_subs_arbiter_pub(
             else:
                 await master_portal.cancel_actor()
 
+            # drop the bg `subs()` runners now the subs are cancelled
+            tn.cancel_scope.cancel()
+
     trio.run(main)
 
 

tests/test_rpc.py

@@ -111,8 +111,9 @@ def test_rpc_errors(
 
             actor = tractor.current_actor()
             assert actor.is_registrar
-            await n.run_in_actor(
+            await tractor.to_actor.run(
                 sleep_back_actor,
+                an=n,
                 actor_name=subactor_requests_to,
 
                 name='subactor',

tests/test_runtime.py

@@ -78,13 +78,12 @@ async def test_lifetime_stack_wipes_tmpfile(
             async with tractor.open_nursery(
                 loglevel=loglevel,
             ) as an:
-                await (  # inlined `tractor.Portal`
-                    await an.run_in_actor(
+                await tractor.to_actor.run(
                     crash_and_clean_tmpdir,
+                    an=an,
                     tmp_file_path=path,
                     error=error_in_child,
                 )
-                ).result()
     except (
         tractor.RemoteActorError,
         BaseExceptionGroup,

tests/test_spawning.py

@@ -48,9 +48,11 @@ async def spawn(
             actor: tractor.Actor = tractor.current_actor()
             assert actor.is_registrar == should_be_root
 
-            # spawns subproc here
-            portal: tractor.Portal = await an.run_in_actor(
-                fn=spawn,
+            # recursively spawn this same `spawn()` fn as the lone
+            # task of a one-shot child subactor and get its result.
+            result = await tractor.to_actor.run(
+                spawn,
+                an=an,
 
                 # spawning args
                 name='sub-actor',
@@ -62,16 +64,6 @@ async def spawn(
                 data=data_to_pass_down,
                 reg_addr=reg_addr,
             )
-
-            assert len(an._children) == 1
-            assert (
-                portal.channel.aid.uid
-                in
-                tractor.current_actor().ipc_server._peers
-            )
-
-            # get result from child subactor
-            result = await portal.result()
             assert result == 10
             return result
     else:
@@ -79,7 +71,7 @@ async def spawn(
         return 10
 
 
-def test_run_in_actor_same_func_in_child(
+def test_to_actor_run_same_func_in_child(
     reg_addr: tuple,
     debug_mode: bool,
 ):
@@ -159,21 +151,16 @@ async def test_most_beautiful_word(
         async with tractor.open_nursery(
             debug_mode=debug_mode,
         ) as an:
-            portal = await an.run_in_actor(
+            res: Any = await tractor.to_actor.run(
                 cellar_door,
+                an=an,
                 return_value=return_value,
                 name='some_linguist',
             )
-
-            res: Any = await portal.wait_for_result()
-            assert res == return_value
-    # The ``async with`` will unblock here since the 'some_linguist'
-    # actor has completed its main task ``cellar_door``.
-
-    # this should pull the cached final result already captured during
-    # the nursery block exit.
-    res: Any = await portal.wait_for_result()
             assert res == return_value
+    # The ``async with`` unblocks here — the 'some_linguist'
+    # one-shot actor completed its lone task ``cellar_door`` and
+    # was reaped by `to_actor.run()`.
     print(res)
 
 
@@ -216,8 +203,9 @@ def test_loglevel_propagated_to_subactor(
             registry_addrs=[reg_addr],
 
         ) as tn:
-            await tn.run_in_actor(
+            await tractor.to_actor.run(
                 check_loglevel,
+                an=tn,
                 loglevel=level,
                 level=level,
             )
@@ -267,11 +255,11 @@ async def check_parent_main_inheritance(
     return has_data
 
 
-def test_run_in_actor_can_skip_parent_main_inheritance(
+def test_to_actor_run_can_skip_parent_main_inheritance(
     start_method: str,  # <- only support on `trio` backend rn.
 ):
     '''
-    Verify ``inherit_parent_main=False`` on ``run_in_actor()``
+    Verify ``inherit_parent_main=False`` on ``to_actor.run()``
     prevents parent ``__main__`` data from reaching the child.
 
     '''
@@ -284,21 +272,21 @@ def test_run_in_actor_can_skip_parent_main_inheritance(
         async with tractor.open_nursery(start_method='trio') as an:
 
             # Default: child receives parent __main__ bootstrap data
-            replaying = await an.run_in_actor(
+            await tractor.to_actor.run(
                 check_parent_main_inheritance,
+                an=an,
                 name='replaying-parent-main',
                 expect_inherited=True,
             )
-            await replaying.result()
 
             # Opt-out: child gets no parent __main__ data
-            isolated = await an.run_in_actor(
+            await tractor.to_actor.run(
                 check_parent_main_inheritance,
+                an=an,
                 name='isolated-parent-main',
                 inherit_parent_main=False,
                 expect_inherited=False,
             )
-            await isolated.result()
 
     trio.run(main)