Wait for active pub/sub cancellation targets

Replace `test_dynamic_pub_sub()`'s fixed startup sleep with an RPC
activity probe in the publisher actor. Track the publisher task and
wait until every launched consumer has installed its first
subscription before raising the user cancellation exception.

This keeps slow spawn backends from passing the regression by
cancelling actors which never reached the streaming workload.

Review: PR #484 (OpenCode)
https://github.com/goodboy/tractor/pull/484#pullrequestreview-5025383596

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

tests/test_advanced_streaming.py

@@ -25,6 +25,7 @@ _registry: dict[str, set[tractor.MsgStream]] = {
     'even': set(),
     'odd': set(),
 }
+_publisher_started: bool = False
 
 
 async def publisher(
@@ -33,11 +34,13 @@ async def publisher(
 
 ) -> None:
 
-    global _registry
+    global _publisher_started, _registry
 
     def is_even(i):
         return i % 2 == 0
 
+    _publisher_started = True
+    try:
         for val in itertools.count(seed):
 
             sub = 'even' if is_even(val) else 'odd'
@@ -49,6 +52,26 @@ async def publisher(
             # making it readable to a human user
             await trio.sleep(1/1000)
 
+    finally:
+        _publisher_started = False
+
+
+async def pubsub_active(
+    expected_subs: int,
+) -> bool:
+    '''
+    Report whether the publisher and all subscriber tasks are active.
+
+    Runs as an RPC task in the publisher actor, where `_registry` is
+    mutated by each `subscribe()` context after its consumer sends the
+    first subscription.
+
+    '''
+    return (
+        _publisher_started
+        and sum(map(len, _registry.values())) >= expected_subs
+    )
+
 
 @tractor.context
 async def subscribe(
@@ -270,8 +293,16 @@ def test_dynamic_pub_sub(
                             )
                         )
 
-                        # block until "cancelled by user"
+                        expected_subs: int = max(cpus - 2, 0) + 1
-                        await trio.sleep(3)
+                        async with tractor.wait_for_actor(
+                            'publisher',
+                        ) as portal:
+                            while not await portal.run(
+                                pubsub_active,
+                                expected_subs=expected_subs,
+                            ):
+                                await trio.sleep(0.01)
+
                         test_log.warning(
                             f'Raising user cancel exc: '
                             f'{expect_cancel_exc!r}'