Preserve cancellation over shielded `send_all()` errors

The frame-publication shield added in `88a23449` checks for
pending cancellation only after a successful write. If actor
teardown closes the stream first, `ClosedResourceError` escaped
as `TransportClosed` and could defeat the caller's cancel scope.

Check for pending cancellation on the transport-error path before
normalizing the close. Genuine stream errors retain their existing
translation when no cancellation is active.

The cancellation-first path can swap which nested debugger
intermediary renders as the immediate source vs. relay. Keep
assertions over both actor levels while accepting either valid role.

Prompt-IO: ai/prompt-io/opencode/20260820T143845Z_559fd0f1_prompt_io.md

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

ai/prompt-io/opencode/20260820T143845Z_559fd0f1_prompt_io.md

@@ -0,0 +1,43 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: 76c5d31c-5a2f-4503-9b16-410ee7f4fab3
+timestamp: 2026-08-20T14:38:50Z
+git_ref: 559fd0f1
+scope: code
+substantive: true
+raw_file: 20260820T143845Z_559fd0f1_prompt_io.raw.md
+---
+
+## Prompt
+
+Continue preparing PR #481 after the latest fix was pushed. Follow CI and
+proceed with clear next steps toward a green landing candidate.
+
+## Response summary
+
+Traced the remaining macOS UDS failure to cancellation racing transport
+teardown inside the shielded framed-send path. Preserve pending cancellation
+over a transport error caused by concurrent teardown, and add a deterministic
+regression for that ordering. A follow-up A/B run showed the corrected
+cancellation precedence changes which nested debugger intermediary is
+rendered as the immediate source versus relay, so retain coverage for both
+actor levels without pinning those racy roles. The adjusted UDS node passes
+three consecutive runs, and both debugger/tooling transport suites pass with
+39 passed and 6 skipped.
+
+## Files changed
+
+- `tractor/ipc/_transport.py` - deliver pending cancellation before
+  translating a shielded send's transport error.
+- `tests/ipc/test_each_tpt.py` - reproduce cancellation followed by local
+  stream closure during shielded frame publication.
+- `tests/devx/test_debugger.py` - accept either valid source/relay role for
+  each nested intermediary while retaining the actor and error assertions.
+
+## Human edits
+
+The human pushed the preceding fix, ran the proposed verification plan, and
+reported a repeated UDS debugger failure. That report prompted the A/B
+comparison and role-insensitive assertion. No direct source-line edits were
+made by the human.

ai/prompt-io/opencode/20260820T143845Z_559fd0f1_prompt_io.raw.md

@@ -0,0 +1,26 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-20T14:38:50Z
+git_ref: 559fd0f1
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+Continue preparing PR #481 after pushing the macOS debugger skip fix.
+Follow the replacement CI run and address any remaining PR-specific
+failure.
+
+> `git diff HEAD~1..HEAD -- tractor/ipc/_transport.py`
+
+> `git diff HEAD~1..HEAD -- tests/ipc/test_each_tpt.py`
+
+macOS UDS failed `test_reqresp_ontopof_streaming` when its two-second
+`move_on_after()` scope cancelled during `stream.send('ping')`. Commit
+`88a23449` shields framed `send_all()` and checks pending cancellation only
+after a successful write. Concurrent transport teardown instead closed the
+socket, causing `ClosedResourceError` to escape as `TransportClosed` before
+the pending cancellation could be delivered.
+
+Preserve structured cancellation precedence on the shielded send's
+transport-error path, and add a deterministic regression that cancels the
+sender before making the fake stream raise `ClosedResourceError`.

tests/devx/test_debugger.py

@@ -892,20 +892,28 @@ def test_multi_nested_subactors_error_through_nurseries(
     # happening but ONLY WHEN RUN FROM THE TEST, bc when i try to
     # run the test script manually the correct output ALWAYS seems
     # to be in the last `str(child.before.decode())` output !?!?
+    transcript: str = '\n'.join(transcript_parts)
     if (
         not is_forking_spawner
         and
         last_send_char == 'q'
     ):
-        expect_patts += [
+        # Cancellation can swap which intermediary is rendered as
-            # expect the pdb-quit exc.
+        # the immediate source vs. relay. Require both actor levels
-            "bdb.BdbQuit",
+        # below without pinning those racy roles.
-            # BUT WHY these dude!?
+        expect_patts.append('bdb.BdbQuit')
-            "src_uid=('spawn_until_0'",
+        for uid in (
-            "relay_uid=('spawn_until_1'",
+            'spawn_until_0',
-        ]
+            'spawn_until_1',
+        ):
+            assert any(
+                role in transcript
+                for role in (
+                    f"src_uid=('{uid}'",
+                    f"relay_uid=('{uid}'",
+                )
+            )
 
-    transcript: str = '\n'.join(transcript_parts)
     for part in expect_patts:
         assert part in transcript
 

tests/ipc/test_each_tpt.py

@@ -134,6 +134,79 @@ def test_cancelled_transport_send_completes_frame():
     trio.run(main)
 
 
+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.
+
+    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.
+
+    '''
+    class ClosingStream:
+        def __init__(self) -> None:
+            self.send_entered = trio.Event()
+            self.release = trio.Event()
+
+        async def send_all(
+            self,
+            data: bytes,
+        ) -> None:
+            assert data
+            self.send_entered.set()
+            await self.release.wait()
+            raise trio.ClosedResourceError(
+                'this socket was already closed'
+            )
+
+    async def main() -> None:
+        stream = ClosingStream()
+        transport = object.__new__(MsgpackTransport)
+        transport.stream = stream
+        transport._send_lock = trio.StrictFIFOLock()
+        sender_done = trio.Event()
+        sender_scopes: list[trio.CancelScope] = []
+        cancelled_caught: bool = False
+
+        msg = tractor.msg.Start(
+            ns=__name__,
+            func='add_one',
+            kwargs={'n': 1},
+            uid=('root', 'test'),
+            cid='close-during-cancelled-send',
+        )
+
+        async def send() -> None:
+            nonlocal cancelled_caught
+            with trio.CancelScope() as cs:
+                sender_scopes.append(cs)
+                await transport.send(msg)
+
+            cancelled_caught = cs.cancelled_caught
+            sender_done.set()
+
+        async with trio.open_nursery() as tn:
+            tn.start_soon(send)
+            await stream.send_entered.wait()
+            sender_scopes[0].cancel()
+            await wait_all_tasks_blocked()
+
+            assert not sender_done.is_set()
+            stream.release.set()
+            await sender_done.wait()
+
+        assert cancelled_caught
+
+    trio.run(main)
+
+
 @pytest.fixture
 def bindspace_dir_str() -> str:
 

tractor/ipc/_transport.py

@@ -526,6 +526,12 @@ class MsgpackTransport(MsgTransport):
                 trio.BrokenResourceError,
                 trio.ClosedResourceError,
             ) as _re:
+                # A shielded send can race outer cancellation with
+                # stream teardown. If teardown closes the stream, let
+                # the pending cancellation retain precedence instead
+                # of converting that close into `TransportClosed`.
+                await trio.lowlevel.checkpoint_if_cancelled()
+
                 trans_err = _re
                 tpt_name: str = f'{type(self).__name__!r}'