Strengthen context and debugger regressions

Older review threads found that context startup mocks did not identify
the internal cancel RPC, the overrun packer seam lacked rationale and
the debugger test no longer asserted its KeyboardInterrupt transcript.

Assert exact startup/cancel RPC ordering, explain the stable error spy,
add terse test typing and restore the terminal interrupt check after
EOF.

Review: PR #481 (goodboy)
https://github.com/goodboy/tractor/pull/481#pullrequestreview-5012942328

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

tests/_helpers.py

@@ -6,11 +6,12 @@ from pathlib import Path
 from types import TracebackType
 
 import tractor
+import trio
 
 
 class CancellationMarkers:
     '''
-    Mark a test endpoint's lifetime without cleanup checkpoints.
+    Mark a test endpoint and require cancellation-driven teardown.
 
     '''
     def __init__(
@@ -30,6 +31,8 @@ class CancellationMarkers:
         exc_value: BaseException|None,
         traceback: TracebackType|None,
     ) -> None:
+        assert exc_type is trio.Cancelled
+        assert isinstance(exc_value, trio.Cancelled)
         Path(self.cancelled_path).touch()
 
 

tests/devx/test_debugger.py

@@ -1359,6 +1359,8 @@ def test_ctxep_pauses_n_maybe_ipc_breaks(
                 expect_prompt=False,
             )
             child.expect(EOF)
+            before += ansi_strip(child.before.decode())
+            assert 'KeyboardInterrupt' in before
             assert child.flag_eof
             assert not child.isalive()
 

tests/ipc/test_each_tpt.py

@@ -33,18 +33,19 @@ def test_cancelled_transport_send_completes_frame():
 
     A cancelled `send_all()` may leave an arbitrary frame prefix on the
     wire. Closing the actor-wide stream avoids decoder corruption but
-    also destroys unrelated contexts using that channel. The fake
-    stream publishes two header bytes and blocks, letting this test
-    cancel the sender inside frame publication. The sender must remain
-    blocked until the complete frame is written, then observe pending
-    cancellation; a second complete frame proves channel reuse remains
-    safe.
+    also destroys unrelated contexts using that channel. On its first
+    call, the fake stream publishes two header bytes and blocks until
+    the parent test releases it. This lets the parent cancel the sender
+    while frame publication is suspended. The sender must remain inside
+    `send_first()` until the complete frame is written, then observe
+    pending cancellation; a second sender proves sibling contexts can
+    safely reuse the still frame-aligned stream.
 
     '''
     class PartialSendStream:
         def __init__(self) -> None:
-            self.send_entered = trio.Event()
-            self.release = trio.Event()
+            self.send_all_entered = trio.Event()
+            self.send_all_release = trio.Event()
             self.closed = False
             self.wire = bytearray()
 
@@ -55,8 +56,8 @@ def test_cancelled_transport_send_completes_frame():
             assert data
             if not self.wire:
                 self.wire.extend(data[:2])
-                self.send_entered.set()
-                await self.release.wait()
+                self.send_all_entered.set()
+                await self.send_all_release.wait()
                 self.wire.extend(data[2:])
             else:
                 self.wire.extend(data)
@@ -115,21 +116,29 @@ def test_cancelled_transport_send_completes_frame():
             tn.start_soon(
                 send_first,
             )
-            await stream.send_entered.wait()
+            await stream.send_all_entered.wait()
             sender_scopes[0].cancel()
             await wait_all_tasks_blocked()
 
             assert not stream.closed
+            # Cancellation is pending, but complete-frame shielding
+            # keeps `send_first()` suspended in `.send_all()`.
             assert not sender_done.is_set()
 
-            stream.release.set()
+            # Let the underlying frame write finish after the parent
+            # has requested sender cancellation.
+            stream.send_all_release.set()
             await sender_done.wait()
 
             assert cancelled_caught
             assert not stream.closed
+            # The initial two-byte prefix was completed into one valid
+            # frame before cancellation reached `send_first()`.
             assert count_frames(stream.wire) == 1
 
             await transport.send(second_msg)
+            # A sibling sender can append and decode another frame only
+            # because the first cancellation preserved stream alignment.
             assert count_frames(stream.wire) == 2
 
             tn.cancel_scope.cancel()
@@ -139,14 +148,15 @@ def test_cancelled_transport_send_completes_frame():
 
 def test_transport_send_deadline_closes_partial_frame():
     '''
-    Bound one shielded frame without exposing a corrupt stream.
+    Destroy a stalled partial frame before another sender can append.
 
     Ordinary cancellation cannot interrupt complete-frame publication.
-    Actor-wide cancellation instead passes its absolute deadline into
-    this operation. The fake stream writes a partial header and stalls;
-    when the send's own deadline fires, the transport must close the
-    stream before releasing its shared send lock and report the channel
-    unusable.
+    Bounded actor/context cancellation instead passes its absolute
+    deadline into this operation. The fake stream writes a partial
+    header and stalls; when the send's own deadline fires, the transport
+    must close the stream before releasing its shared send lock. This
+    prevents the next sender from appending bytes which a decoder would
+    treat as the remainder of the corrupt first frame.
 
     '''
     class StalledStream:
@@ -186,9 +196,9 @@ def test_transport_send_deadline_closes_partial_frame():
                 send_deadline=1,
             )
 
-        assert stream.closed
-        assert len(stream.wire) == 2
-        assert not transport._send_lock.locked()
+        assert stream.closed  # partial-frame timeout destroys stream
+        assert len(stream.wire) == 2  # only a header fragment was sent
+        assert not transport._send_lock.locked()  # cleanup released lock
 
     trio.run(
         main,
@@ -200,30 +210,35 @@ def test_cancelled_transport_send_preserves_cancellation():
     '''
     Prefer sender cancellation when teardown closes the stream.
 
-    `MsgpackTransport.send()` shields frame publication. Before this
-    regression fix, if an outer scope cancelled the sender and actor
-    teardown then made `send_all()` raise `ClosedResourceError`, the
-    transport error escaped instead of the pending cancellation. That
-    defeated `move_on_after()` and failed otherwise orderly teardown.
+    `MsgpackTransport.send()` shields frame publication at
+    `tractor.ipc._transport:MsgpackTransport.send`. Before this fix, an
+    outer `move_on_after()`/cancel scope could cancel `Channel.send()`
+    while actor teardown closed the shared stream. The resulting
+    `ClosedResourceError` escaped from the transport handler instead of
+    its `checkpoint_if_cancelled()` redelivering pending cancellation.
 
     The fake stream blocks inside the shield until the test cancels the
-    sender, then raises the same close error seen on macOS UDS. Observing
-    `CancelScope.cancelled_caught` proves cancellation wins once the
-    shield unwinds.
+    sender. Parent-controlled release then simulates actor teardown
+    closing the socket and raises the `ClosedResourceError` observed on
+    macOS UDS. `CancelScope.cancelled_caught` proves the handler's
+    checkpoint preserved cancellation as the primary outcome instead of
+    leaking that secondary close error.
 
     '''
     class ClosingStream:
         def __init__(self) -> None:
-            self.send_entered = trio.Event()
-            self.release = trio.Event()
+            self.send_all_entered = trio.Event()
+            self.send_all_release = trio.Event()
 
         async def send_all(
             self,
             data: bytes,
         ) -> None:
             assert data
-            self.send_entered.set()
-            await self.release.wait()
+            self.send_all_entered.set()
+            await self.send_all_release.wait()
+            # Model actor teardown closing the shared transport while
+            # this sender is still inside the complete-frame shield.
             raise trio.ClosedResourceError(
                 'this socket was already closed'
             )
@@ -256,12 +271,12 @@ def test_cancelled_transport_send_preserves_cancellation():
 
         async with trio.open_nursery() as tn:
             tn.start_soon(send)
-            await stream.send_entered.wait()
+            await stream.send_all_entered.wait()
             sender_scopes[0].cancel()
             await wait_all_tasks_blocked()
 
             assert not sender_done.is_set()
-            stream.release.set()
+            stream.send_all_release.set()
             await sender_done.wait()
 
         assert cancelled_caught

tests/test_context_stream_semantics.py

@@ -86,7 +86,7 @@ from ._helpers import (
 
 def test_overrun_error_send_tolerates_transport_close(
     monkeypatch: pytest.MonkeyPatch,
-):
+) -> None:
     '''
     Preserve a stream overrun when its error can not be shipped.
 
@@ -108,10 +108,13 @@ def test_overrun_error_send_tolerates_transport_close(
     )
     packed: dict[str, object] = {}
 
+    # Spy on the generated `StreamOverrun` and return a stable wire
+    # `Error`; the real packer adds traceback/relay details unrelated to
+    # this test's secondary transport-close contract.
     def pack_overrun(
         local_err: BaseException,
         cid: str,
-        **kwargs,
+        **kwargs: object,
     ) -> tractor.msg.Error:
         packed['local_err'] = local_err
         packed['cid'] = cid
@@ -275,7 +278,7 @@ async def simple_setup_teardown(
         _state = False
 
 
-async def assert_state(value: bool):
+async def assert_state(value: bool) -> None:
     global _state
     assert _state == value
 
@@ -286,7 +289,7 @@ async def test_cancel_during_context_startup(
     tmp_path: Path,
     start_method: str,
     debug_mode: bool,
-):
+) -> None:
     '''
     Cancel a context after sending `Start` but before its ack.
 
@@ -303,6 +306,7 @@ async def test_cancel_during_context_startup(
     started_path = tmp_path / 'startup_started'
     cancelled_path = tmp_path / 'startup_cancelled'
     start_sent = trio.Event()
+    start_funcs: list[str] = []
     original_send = tractor.Channel.send
 
     async def delay_after_start(
@@ -317,7 +321,11 @@ async def test_cancel_during_context_startup(
             hide_tb=hide_tb,
             send_deadline=send_deadline,
         )
+        # The patched method keeps `Channel.send()`'s broad message
+        # contract. It sees both the requested endpoint `Start` and the
+        # internal `self._cancel_task` startup RPC used for cleanup.
         if isinstance(payload, tractor.msg.Start):
+            start_funcs.append(payload.func)
             if payload.func == 'startup_cancel_target':
                 start_sent.set()
             await trio.sleep_forever()
@@ -333,7 +341,7 @@ async def test_cancel_during_context_startup(
             raise AssertionError('context startup should be cancelled')
 
     async with tractor.open_nursery() as an:
-        actor = tractor.current_actor()
+        actor: Actor = tractor.current_actor()
         portal: tractor.Portal = await an.start_actor(
             'startup_cancel_worker',
             enable_modules=[__name__],
@@ -365,6 +373,10 @@ async def test_cancel_during_context_startup(
             'return_one',
         ) == 1
         assert non_registration_contexts(actor) == contexts_before
+        assert start_funcs == [
+            'startup_cancel_target',
+            '_cancel_task',
+        ]
         await portal.cancel_actor()
 
 
@@ -372,7 +384,7 @@ async def test_cancel_during_context_startup(
 async def test_start_serialization_error_cleans_context(
     start_method: str,
     debug_mode: bool,
-):
+) -> None:
     '''
     Deallocate caller state when `Start` can not be serialized.
 
@@ -385,7 +397,7 @@ async def test_start_serialization_error_cleans_context(
 
     '''
     async with tractor.open_nursery() as an:
-        actor = tractor.current_actor()
+        actor: Actor = tractor.current_actor()
         portal: tractor.Portal = await an.start_actor(
             'serialization_error_worker',
             enable_modules=[__name__],
@@ -414,7 +426,7 @@ async def test_start_serialization_error_cleans_context(
 async def test_start_module_error_cleans_context(
     start_method: str,
     debug_mode: bool,
-):
+) -> None:
     '''
     Deallocate caller state after a remote startup rejection.
 
@@ -427,7 +439,7 @@ async def test_start_module_error_cleans_context(
 
     '''
     async with tractor.open_nursery() as an:
-        actor = tractor.current_actor()
+        actor: Actor = tractor.current_actor()
         portal: tractor.Portal = await an.start_actor(
             'module_error_worker',
         )

tests/test_to_actor.py

@@ -77,9 +77,9 @@ async def mark_task_cancellation(
 
 
 async def echo_startup_control(
-    _cancel_on_startup: str,
+    cancel_on_startup: str,
 ) -> str:
-    return _cancel_on_startup
+    return cancel_on_startup
 
 
 async def collect_args(
@@ -88,13 +88,6 @@ async def collect_args(
     return args
 
 
-async def collect_call(
-    *args: object,
-    **kwargs: object,
-) -> tuple[tuple[object, ...], dict[str, object]]:
-    return args, kwargs
-
-
 def test_public_module_alias() -> None:
     '''
     Keep the public trampoline alias separate from its private module.
@@ -587,9 +580,12 @@ async def test_reuse_existing_actor_via_portal(
     Pass `portal=` to schedule the one-shot task in an
     already-running actor; no spawn, no implicit reap.
 
-    The low-level `Portal.run_from_ns()` assertion also proves its
-    target kwargs remain separate from the private startup-cancel
-    policy used by context cleanup.
+    The low-level call uses `__name__` to select the remote module and
+    `'echo_startup_control'` to select its function. Public
+    `Portal.run_from_ns()` packages `cancel_on_startup` inside the
+    target `kwargs` passed to private `Portal._run_from_ns()`. Receiving
+    `'target_value'` back proves the value reached the target instead of
+    binding the private Boolean startup-cancellation policy parameter.
 
     '''
     async with tractor.open_nursery() as an:
@@ -612,7 +608,7 @@ async def test_reuse_existing_actor_via_portal(
         assert await portal.run_from_ns(
             __name__,
             'echo_startup_control',
-            _cancel_on_startup='target_value',
+            cancel_on_startup='target_value',
         ) == 'target_value'
         assert non_registration_contexts(actor) == contexts_before
 
@@ -631,6 +627,8 @@ async def test_concurrent_one_shots_from_task_nursery(
     nursery scheduling multiple one-shot calls against
     a shared caller-managed actor-nursery; error
     collection thus lives entirely in caller-code.
+    A proposed distilled task-manager API is tracked in #485:
+    https://github.com/goodboy/tractor/issues/485
 
     '''
     results: dict[int, int] = {}
@@ -658,7 +656,7 @@ async def test_concurrent_one_shots_from_task_nursery(
     }
 
 
-def test_rejects_sync_fn():
+def test_rejects_sync_fn() -> None:
     '''
     Non-async callables error BEFORE any spawn (or even
     runtime-boot) happens.
@@ -667,7 +665,10 @@ def test_rejects_sync_fn():
     def not_async() -> None:
         ...
 
-    with pytest.raises(TypeError):
+    with pytest.raises(
+        TypeError,
+        match='must be a non-streaming async function',
+    ):
         trio.run(
             partial(
                 to_actor.run,
@@ -676,7 +677,7 @@ def test_rejects_sync_fn():
         )
 
 
-def test_rejects_streaming_fn():
+def test_rejects_streaming_fn() -> None:
     '''
     Async-gen (streaming) fns are not one-shot-able,
     same constraint as `Portal.run()`.
@@ -685,7 +686,10 @@ def test_rejects_streaming_fn():
     async def agen():
         yield 1
 
-    with pytest.raises(TypeError):
+    with pytest.raises(
+        TypeError,
+        match='must be a non-streaming async function',
+    ):
         trio.run(
             partial(
                 to_actor.run,
@@ -738,7 +742,7 @@ def test_partial_placeholder_normalization(
         )
 
 
-def test_nested_partial_normalization():
+def test_nested_partial_normalization() -> None:
     '''
     Flatten every retained `functools.partial` layer before RPC.
 
@@ -750,6 +754,12 @@ def test_nested_partial_normalization():
     direct nested-partial call.
 
     '''
+    async def collect_call(
+        *args: object,
+        **kwargs: object,
+    ) -> tuple[tuple[object, ...], dict[str, object]]:
+        return args, kwargs
+
     inner = partial(
         collect_call,
         1,
@@ -771,13 +781,15 @@ def test_nested_partial_normalization():
     assert kwargs == {'label': 'outer'}
 
 
-def test_rejects_portal_and_an_combo():
+def test_rejects_portal_and_an_combo() -> None:
     '''
-    `portal=` and `an=` are mutually exclusive
-    placement options.
+    `portal=` and `an=` are mutually exclusive actor-lifetime handles.
 
     '''
-    with pytest.raises(ValueError):
+    with pytest.raises(
+        ValueError,
+        match='Pass at most ONE of `portal` or `an`',
+    ):
         trio.run(
             partial(
                 to_actor.run,
@@ -790,7 +802,7 @@ def test_rejects_portal_and_an_combo():
 
 
 @pytest.mark.parametrize(
-    'placement',
+    'lifetime_mode',
     ['an', 'portal'],
 )
 @pytest.mark.parametrize(
@@ -801,28 +813,31 @@ def test_rejects_portal_and_an_combo():
     ],
     ids=['empty', 'configured'],
 )
-def test_rejects_runtime_kwargs_with_placement(
-    placement: str,
+def test_rejects_runtime_kwargs_with_lifetime_mode(
+    lifetime_mode: str,
     runtime_kwargs: dict,
-):
+) -> None:
     '''
     `runtime_kwargs` only applies when the call opens
     its own private actor-nursery; passing it alongside
-    a placement opt is an error, never silently
+    an actor-lifetime handle is an error, never silently
     ignored. In particular, an empty dict still means the
     caller provided this mutually exclusive option; testing
-    both placement modes prevents truthiness checks from
+    both lifetime modes prevents truthiness checks from
     accepting it before any actor runtime is started.
 
     '''
-    with pytest.raises(ValueError):
+    with pytest.raises(
+        ValueError,
+        match='`runtime_kwargs` only applies',
+    ):
         trio.run(
             partial(
                 to_actor.run,
                 add_one,
                 1,
                 **{
-                    placement: object(),
+                    lifetime_mode: object(),
                     'runtime_kwargs': runtime_kwargs,
                 },
             )
@@ -880,12 +895,13 @@ async def test_portal_task_cancelled_with_local_caller(
     Couple a reused portal's remote task to its local caller.
 
     The former `Portal.run()` path abandoned its remote task when the
-    local `to_actor.run()` caller was cancelled. The target writes
-    one file after starting and another from its cancellation
-    `finally`. Cancelling the local task nursery and observing the
-    second file proves `Portal.open_context()` propagated
-    cancellation before the caller exited. A subsequent call proves
-    the caller-owned actor was not cancelled with that task.
+    local `to_actor.run()` caller was cancelled. `CancellationMarkers`
+    writes one file after entry and writes the second only after its
+    synchronous exit verifies the remote task received `trio.Cancelled`.
+    Cancelling the local task nursery and observing that marker proves
+    `Portal.open_context()` propagated cancellation before the caller
+    exited. A subsequent call proves the caller-owned actor was not
+    cancelled with that task.
 
     '''
     started_path = tmp_path / 'started'
@@ -918,12 +934,16 @@ async def test_portal_task_cancelled_with_local_caller(
             tn.cancel_scope.cancel()
 
         assert cancelled_path.exists()
+        # Remote-task cancellation must restore the exact application
+        # context snapshot captured before the one-shot call.
         assert non_registration_contexts(actor) == contexts_before
         assert await to_actor.run(
             add_one,
             1,
             portal=portal,
         ) == 2
+        # Reusing the actor for a later successful call must also leave
+        # no local or remote context registry entries behind.
         assert non_registration_contexts(actor) == contexts_before
 
         await portal.cancel_actor()
@@ -958,7 +978,10 @@ async def test_context_trampoline_preserves_module_allowlist(
                 portal=portal,
             )
 
-        assert excinfo.value.boxed_type is tractor.ModuleNotExposed
+        err = excinfo.value
+        assert err.boxed_type is tractor.ModuleNotExposed
+        assert add_one.__module__ in str(err)
+        assert 'Make sure you exposed the target module' in str(err)
         assert non_registration_contexts(actor) == contexts_before
         await portal.cancel_actor()
 

tractor/_context.py

@@ -1113,6 +1113,10 @@ class Context:
                 # NOTE: we're telling the far end actor to cancel a task
                 # corresponding to *this actor*. The far end local channel
                 # instance is passed to `Actor._cancel_task()` implicitly.
+                # Use private `Portal._run_from_ns()` because cancellation
+                # needs its internal `cancel_on_startup=False` policy and
+                # the transaction's shared absolute `send_deadline`; public
+                # `run_from_ns()` exposes neither control.
                 await self._portal._run_from_ns(
                     'self',
                     '_cancel_task',
@@ -1955,6 +1959,9 @@ class Context:
         # the sender; the main motivation is that using bp can block the
         # msg handling loop which calls into this method!
         except trio.WouldBlock:
+            # `send_chan.send_nowait(msg)` found the local receive feeder
+            # full. With overruns disabled below, report that primary
+            # local overflow to the far-end sender as `StreamOverrun`.
 
             # XXX: always push an error even if the local receiver
             # is in overrun state - i.e. if an 'error' msg is
@@ -2027,12 +2034,13 @@ class Context:
                     await chan.send(err_msg)
                     return True
 
-                # XXX: the local consumer may have closed its side of
-                # the IPC, in which case context/channel teardown owns
-                # cancellation of the far-end streaming task. The same
-                # shipment can raise `TransportClosed` when either peer
-                # has already closed the shared IPC channel. In both
-                # cases the primary overrun can no longer be reported.
+                # The `StreamOverrun` shipment can fail secondarily when
+                # context/channel teardown has already closed shared IPC.
+                # Local stream closure may surface as
+                # `BrokenResourceError`; either peer closing the transport
+                # can surface as `TransportClosed`. In both cases teardown
+                # owns far-end cancellation and the primary overrun can no
+                # longer be delivered.
                 except (
                     TransportClosed,
                     trio.BrokenResourceError,

tractor/ipc/_transport.py

@@ -448,9 +448,11 @@ class MsgpackTransport(MsgTransport):
         If `strict_types == True` then a `MsgTypeError` will be raised on any
         invalid msg type
 
-        `send_deadline` bounds publication of this complete frame. A
-        timeout destroys the stream because a partial prefix may have
-        reached the wire.
+        `send_deadline` bounds publication of one complete
+        length-prefixed frame. If it expires after a prefix or payload
+        fragment reaches the wire, a later sender would append bytes the
+        peer decoder treats as the remainder of that corrupt frame. The
+        stream is therefore destroyed before releasing `._send_lock`.
 
         '''
         __tracebackhide__: bool = hide_tb
@@ -505,18 +507,17 @@ class MsgpackTransport(MsgTransport):
             try:
                 # Every IPC msg is length-prefixed and all contexts
                 # on this actor pair share one transport stream. If
-                # cancellation interrupts `send_all()`, an unknown
-                # frame prefix may already be on the wire; allowing
-                # the next sender to append would corrupt framing.
-                # Closing the stream avoids that corruption but lets
-                # one context-local cancellation destroy every sibling
-                # context using the channel.
+                # the send deadline interrupts `send_all()`, an unknown
+                # frame prefix may already be on the wire; allowing the
+                # next sender to append would corrupt framing.
                 #
-                # Keep the `._send_lock` and defer cancellation only
-                # for complete frame publication. Broken/closed stream
-                # failures still escape to the handlers below. Once
-                # the frame is complete, the explicit checkpoint
-                # immediately delivers any pending cancellation.
+                # The enclosing `async with self._send_lock` retains the
+                # lock through shielded publication and any forced-close
+                # cleanup. Context-manager exit releases it only after a
+                # complete frame or destruction of the corrupt stream.
+                # Ordinary outer cancellation remains shielded until
+                # complete publication; only expiry of `send_deadline`
+                # intentionally closes a partial stream here.
                 #
                 # Ordinary sends may delay cancellation while the remote
                 # peer actor is not reading. Bounded actor/context cancel
@@ -539,6 +540,8 @@ class MsgpackTransport(MsgTransport):
                         f'deadline of {send_deadline!r}'
                     )
 
+                # The frame is complete and still aligned. Redeliver any
+                # pending outer cancellation before normal lock release.
                 await trio.lowlevel.checkpoint_if_cancelled()
                 return None
 

tractor/runtime/_runtime.py

@@ -794,8 +794,9 @@ class Actor:
         cancel_on_startup: bool = True,
 
         # Optional absolute deadline for publishing this exact `Start`
-        # frame. Used by actor-wide cancel RPCs whose outer timeout
-        # cannot penetrate complete-frame transport shielding.
+        # frame. Used by bounded actor/context cancel RPCs whose outer
+        # timeout cannot penetrate `Channel.send()` forwarding into the
+        # shield in `MsgpackTransport.send()`.
         send_deadline: float = float('inf'),
 
     ) -> Context: