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(
+            print(res)
-            ria_fn,
+
-            other_actor='donny',
+        tn.start_soon(_direct, 'donny', 'gretchen')
-            reg_addr=reg_addr,
+        tn.start_soon(_direct, 'gretchen', 'donny')
-            name='gretchen',
-        )
-        print(await gretchen.result())
-        print(await donny.result())
         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__])
+                                name=name,
-
+                                enable_modules=[__name__],
-                            else:  # no streaming
+                            )
-                                portals[name] = await an.run_in_actor(
-                                    trio.sleep_forever, name=name)
 
                         # wait on last actor to come up
                         async with tractor.wait_for_actor(name):