Port `test_dynamic_pub_sub` off `run_in_actor`

The known-flaky dynamic pubsub test's 3 fire-and-forget spawn
sites (#477 removal),

- the forever-streaming `publisher` + N `consumer` one-shots now
  bg-schedule as `to_actor.run(fn, an=n)` tasks in a local `trio`
  task-nursery (`publisher`'s rendezvous name still derives from
  `fn.__name__`).
- the simulated user-cancel raise (`KeyboardInterrupt` /
  `TooSlowError` params) cancels the task-nursery, each one-shot
  reaping its subactor via `to_actor.run()`'s shielded
  `Portal.cancel_actor()`; `_run_and_match()`'s existing
  `BaseExceptionGroup.split()` walk covers the (possibly nested)
  relay shapes unchanged.
- spawns now issue concurrently rather than sequentially —
  comment on the fork-backend budget updated to match.

Gate: both params x4 runs green on `trio` + x1 on `mp_spawn`;
full module green.

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

tests/test_advanced_streaming.py

@@ -3,6 +3,7 @@ Advanced streaming patterns using bidirectional streams and contexts.
 
 '''
 from collections import Counter
+from functools import partial
 import itertools
 import platform
 from typing import Type
@@ -173,8 +174,8 @@ def test_dynamic_pub_sub(
     # test. Picked backend-aware: under `trio` backend spawn is
     # cheap (~1s for `cpus` actors) but fork-based backends pay
     # a per-spawn cost (forkserver round-trip + IPC peer-handshake)
-    # that can stack up over `cpus - 1` sequential `n.run_in_actor()`
+    # that can stack up over the `cpus - 1` one-shot
-    # calls — especially on UDS under cross-pytest contention
+    # (`to_actor.run()`) spawns — especially on UDS under cross-pytest contention
     # (#451 / #452). 4s was flaking right at the edge under fork
     # backends — bumped to 8s with diag-snapshot-on-timeout via
     # `fail_after_w_trace` so a borderline run still fails loud
@@ -214,33 +215,55 @@ def test_dynamic_pub_sub(
                     f'enter `fail_after_w_trace({fail_after_s})` scope'
                 )
                 try:
-                    async with tractor.open_nursery(
+                    async with (
-                        registry_addrs=[reg_addr],
+                        tractor.open_nursery(
-                        debug_mode=debug_mode,
+                            registry_addrs=[reg_addr],
-                    ) as n:
+                            debug_mode=debug_mode,
+                        ) as n,
+                        # bg-schedules the forever-streaming
+                        # one-shots below; the user-cancel raise
+                        # cancels them all, each reaping its
+                        # subactor via `to_actor.run()`'s
+                        # (shielded) `Portal.cancel_actor()`.
+                        trio.open_nursery() as tn,
+                    ):
                         test_log.cancel(
                             'test_dynamic_pub_sub: '
                             'actor nursery opened'
                         )
 
                         # name of this actor will be same as target func
-                        await n.run_in_actor(publisher)
+                        tn.start_soon(
+                            partial(
+                                tractor.to_actor.run,
+                                publisher,
+                                an=n,
+                            )
+                        )
 
                         for i, sub in zip(
                             range(cpus - 2),
                             itertools.cycle(_registry.keys())
                         ):
-                            await n.run_in_actor(
+                            tn.start_soon(
-                                consumer,
+                                partial(
-                                name=f'consumer_{sub}',
+                                    tractor.to_actor.run,
-                                subs=[sub],
+                                    consumer,
+                                    an=n,
+                                    name=f'consumer_{sub}',
+                                    subs=[sub],
+                                )
                             )
 
                         # make one dynamic subscriber
-                        await n.run_in_actor(
+                        tn.start_soon(
-                            consumer,
+                            partial(
-                            name='consumer_dynamic',
+                                tractor.to_actor.run,
-                            subs=list(_registry.keys()),
+                                consumer,
+                                an=n,
+                                name='consumer_dynamic',
+                                subs=list(_registry.keys()),
+                            )
                         )
 
                         # block until "cancelled by user"