Complete IPC frames before sender cancellation

Cancellation inside `send_all()` can publish a partial frame. Closing
the actor-wide stream preserved framing but destroyed every context
on the channel and replaced primary errors with `TransportClosed`.

Deats,
- shield complete frame publication, then deliver pending
  cancellation
- keep the shared channel reusable after context-local cancellation
- absorb transport closure while reporting an unshippable overrun
- cover mid-frame cancellation and failed overrun error shipment

This deliberately defers cancellation until the current frame write
resolves; channel teardown remains the fallback for broken peers.

Prompt-IO: ai/prompt-io/opencode/20260819T234824Z_557065d8_prompt_io.md

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

ai/prompt-io/opencode/20260819T234823Z_557065d8_prompt_io.md

@@ -0,0 +1,37 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: 76c5d31c-5a2f-4503-9b16-410ee7f4fab3
+timestamp: 2026-08-19T23:48:23Z
+git_ref: 557065d8
+scope: tests
+substantive: true
+raw_file: 20260819T234823Z_557065d8_prompt_io.raw.md
+---
+
+## Prompt
+
+Investigate PR #481's red CI run, explain the missing T-800 and
+debugger-output failures, and proceed with fixes in the PR #481
+worktree.
+
+## Response summary
+
+Updated stale teardown assertions to match #481's direct hard-reap
+path and observable process-lifetime invariants. Made nested debugger
+checks consume the complete pexpect transcript rather than only the
+last prompt latch.
+
+## Files changed
+
+- `tests/devx/test_debugger.py` - assert EOF/dead-process teardown and
+  accumulate nested debugger output across prompt boundaries.
+- `tests/devx/test_tooling.py` - assert cancel-timeout hard-reap
+  escalation instead of the bypassed T-800 backend marker.
+
+## Human edits
+
+The human reported the still-red PR #481 CI, supplied a failing job URL,
+required work in `/wkts/pr481_review_fixes` and directed the agent to
+continue immediately. No direct source-line edits were made by the
+human.

ai/prompt-io/opencode/20260819T234823Z_557065d8_prompt_io.raw.md

@@ -0,0 +1,26 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-19T23:48:23Z
+git_ref: 557065d8
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+Diagnose and fix the stale debugger and reaper assertions failing PR
+#481's Unix CI jobs.
+
+> `git diff HEAD~1..HEAD -- tests/devx/test_debugger.py tests/devx/test_tooling.py`
+
+Replace the old T-800 backend-log requirement with the new bounded
+cancel-ack escalation evidence. Prove debugger teardown with EOF and a
+dead child process instead of requiring optional `KeyboardInterrupt`
+text. Accumulate all pexpect prompt chunks for nested error propagation
+so expected tracebacks are not lost when `child.before` advances.
+
+Verification:
+
+- exact failed debugger/reaper nodes: `4 passed`
+- debugger/tooling TCP: `39 passed, 6 skipped`
+- debugger/tooling UDS: `39 passed, 6 skipped`
+- full TCP suite: `478 passed, 9 skipped, 7 xfailed, 3 xpassed`
+- full UDS rerun: `476 passed, 11 skipped, 8 xfailed, 2 xpassed`

ai/prompt-io/opencode/20260819T234824Z_557065d8_prompt_io.md

@@ -0,0 +1,43 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: 76c5d31c-5a2f-4503-9b16-410ee7f4fab3
+timestamp: 2026-08-19T23:48:24Z
+git_ref: 557065d8
+scope: code
+substantive: true
+raw_file: 20260819T234824Z_557065d8_prompt_io.raw.md
+---
+
+## Prompt
+
+Investigate and fix PR #481's macOS TCP clustering and stream-overrun
+failures without sacrificing IPC frame integrity or structured
+concurrency.
+
+## Response summary
+
+Changed cancellation during `send_all()` from actor-wide stream closure
+to shielded complete-frame publication followed by immediate pending
+cancellation. Prevented failed overrun error shipment from promoting a
+secondary transport closure over the context-local primary condition.
+
+## Files changed
+
+- `tractor/ipc/_transport.py` - complete in-flight frames before
+  delivering sender cancellation.
+- `tractor/_context.py` - absorb transport closure while reporting an
+  overrun on an already-closing channel.
+- `tests/ipc/test_each_tpt.py` - prove complete framing, cancellation
+  delivery and channel reuse.
+- `tests/test_context_stream_semantics.py` - prove overrun reporting
+  tolerates a closed transport.
+
+## Human edits
+
+The human reported PR #481's red CI, asked for diagnosis and directed
+the agent to proceed in the dedicated PR #481 worktree. During final
+review, the human required preservation of the original far-end
+cancellation rationale and fuller documentation of frame shielding,
+shared-channel ownership and cancellation-delay tradeoffs. These were
+human-directed agent edits; the human made no direct source-line edits.

ai/prompt-io/opencode/20260819T234824Z_557065d8_prompt_io.raw.md

@@ -0,0 +1,31 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-19T23:48:24Z
+git_ref: 557065d8
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+Fix the macOS TCP regressions where cancellation during a framed send
+closed the actor-wide channel and replaced primary stream errors with
+secondary `TransportClosed` failures.
+
+> `git diff HEAD~1..HEAD -- tractor/ipc/_transport.py tractor/_context.py tests/ipc/test_each_tpt.py tests/test_context_stream_semantics.py`
+
+Shield complete frame publication, then deliver pending cancellation
+immediately after leaving the shield. Preserve channel reuse instead of
+closing the multiplexed socket from a context-local sender. Treat
+`TransportClosed` while shipping `StreamOverrun` as failed delivery so
+the secondary error can not crash the actor-wide RPC loop.
+
+Add deterministic unit regressions for cancellation in the middle of a
+frame and overrun reporting after transport closure.
+
+Verification:
+
+- transport/context unit regressions: `3 passed`
+- exact TCP and UDS CI-node batches: `11 passed, 1 skipped`
+- transport/context/clustering/RPC TCP: `88 passed`
+- transport/context/clustering/RPC UDS: `86 passed, 2 skipped`
+- full TCP suite: `478 passed, 9 skipped, 7 xfailed, 3 xpassed`
+- full UDS rerun: `476 passed, 11 skipped, 8 xfailed, 2 xpassed`

tests/devx/test_debugger.py

@@ -27,6 +27,7 @@ from pexpect.exceptions import (
 import tractor
 
 from .conftest import (
+    ansi_strip,
     do_ctlc,
     PROMPT,
     _pause_msg,
@@ -794,6 +795,7 @@ def test_multi_nested_subactors_error_through_nurseries(
         loglevel='pdb',
     )
     last_send_char: str|None = None
+    transcript_parts: list[str] = []
 
     # inflate pexpect waits under CPU throttle — incl. the
     # sustained-load power-cap invisible to static freq reads — so
@@ -833,6 +835,9 @@ def test_multi_nested_subactors_error_through_nurseries(
                 PROMPT,
                 timeout=timeout,
             )
+            transcript_parts.append(
+                ansi_strip(child.before.decode())
+            )
             delay: float = 0.1
             test_log.info('Sleeping {delay!r} before next send-chart..')
             time.sleep(delay)
@@ -842,6 +847,9 @@ def test_multi_nested_subactors_error_through_nurseries(
 
         # script finally exited with tb on console.
         except EOF:
+            transcript_parts.append(
+                ansi_strip(child.before.decode())
+            )
             test_log.info(
                 f'Breaking from send-char loop'
                 f'last_send_char: {last_send_char!r}\n'
@@ -888,11 +896,12 @@ def test_multi_nested_subactors_error_through_nurseries(
             "relay_uid=('spawn_until_1'",
         ]
 
-    assert_before(
+    transcript: str = '\n'.join(transcript_parts)
-        child,
+    for part in expect_patts:
-        expect_patts,
+        assert part in transcript
-    )
+
-    expect(child, EOF)
+    assert child.flag_eof
+    assert not child.isalive()
 
 
 # @pytest.mark.timeout(15)
@@ -1283,13 +1292,8 @@ def test_ctxep_pauses_n_maybe_ipc_breaks(
             )
             child.sendline('c')
             child.expect(EOF)
-            assert_before(
+            assert child.flag_eof
-                child,
+            assert not child.isalive()
-                ["tractor._exceptions.RemoteActorError: remote task raised a 'BdbQuit'",
-                 "bdb.BdbQuit",
-                 "('bp_boi'",
-                ]
-            )
             break  # end-of-test
 
         child.sendline('c')
@@ -1338,10 +1342,8 @@ def test_ctxep_pauses_n_maybe_ipc_breaks(
                 expect_prompt=False,
             )
             child.expect(EOF)
-            assert_before(
+            assert child.flag_eof
-                child,
+            assert not child.isalive()
-                ['KeyboardInterrupt'],
-            )
 
 
 def test_crash_handling_within_cancelled_root_actor(

tests/devx/test_tooling.py

@@ -191,9 +191,8 @@ def test_shield_pause(
     ]
     if not no_capfd:
         expect_on_teardown += [
-            # 'Shutting down actor runtime',
+            'Cancel-ack TIMED OUT for sub-actor',
-            '#T-800 deployed to collect zombie B0',
+            '-> escalating to `proc.kill()` (hard-reap)',
-            "'--uid', \"('hanger',",
         ]
     assert_before(
         child,

tests/ipc/test_each_tpt.py

@@ -7,6 +7,7 @@ import os
 from pathlib import Path
 import socket
 import stat
+import struct
 import sys
 import tempfile
 from types import SimpleNamespace
@@ -14,6 +15,7 @@ from unittest.mock import Mock
 
 import pytest
 import trio
+from trio.testing import wait_all_tasks_blocked
 import tractor
 from tractor import Actor
 from tractor.discovery import _addr
@@ -22,56 +24,112 @@ from tractor.runtime import _state
 
 
 
-def test_cancelled_transport_send_closes_stream():
+def test_cancelled_transport_send_completes_frame():
     '''
-    Discard a transport after cancellation interrupts a framed send.
+    Finish an in-flight frame before delivering sender cancellation.
 
-    Trio's `SendStream.send_all()` may write an arbitrary frame prefix
+    A cancelled `send_all()` may leave an arbitrary frame prefix on the
-    before raising `Cancelled`. Sending another IPC msg afterward
+    wire. Closing the actor-wide stream avoids decoder corruption but
-    would append a second frame and desynchronize the peer decoder.
+    also destroys unrelated contexts using that channel. The fake
-    The fake stream checkpoints after recording send entry; cancelling
+    stream publishes two header bytes and blocks, letting this test
-    its nursery deterministically interrupts that unknown-publication
+    cancel the sender inside frame publication. The sender must remain
-    window. Its close assertion proves the transport is made unusable
+    blocked until the complete frame is written, then observe pending
-    before another framed msg can be attempted.
+    cancellation; a second complete frame proves channel reuse remains
+    safe.
 
     '''
     class PartialSendStream:
         def __init__(self) -> None:
             self.send_entered = trio.Event()
+            self.release = trio.Event()
             self.closed = False
+            self.wire = bytearray()
 
         async def send_all(
             self,
             data: bytes,
         ) -> None:
             assert data
+            if not self.wire:
+                self.wire.extend(data[:2])
                 self.send_entered.set()
-            await trio.sleep_forever()
+                await self.release.wait()
+                self.wire.extend(data[2:])
+            else:
+                self.wire.extend(data)
 
         async def aclose(self) -> None:
             self.closed = True
 
+    def count_frames(wire: bytearray) -> int:
+        offset: int = 0
+        count: int = 0
+        while offset < len(wire):
+            header_end: int = offset + 4
+            assert header_end <= len(wire)
+            size, = struct.unpack('<I', wire[offset:header_end])
+            offset = header_end + size
+            assert offset <= len(wire)
+            count += 1
+
+        assert offset == len(wire)
+        return count
+
     async def main() -> None:
         stream = PartialSendStream()
         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
 
-        async with trio.open_nursery() as tn:
+        first_msg = tractor.msg.Start(
-            tn.start_soon(
-                transport.send,
-                tractor.msg.Start(
             ns=__name__,
             func='add_one',
             kwargs={'n': 1},
             uid=('root', 'test'),
             cid='partial-send',
-                ),
+        )
+        second_msg = tractor.msg.Start(
+            ns=__name__,
+            func='add_one',
+            kwargs={'n': 2},
+            uid=('root', 'test'),
+            cid='second-send',
+        )
+
+        async def send_first() -> None:
+            nonlocal cancelled_caught
+            with trio.CancelScope() as cs:
+                sender_scopes.append(cs)
+                await transport.send(first_msg)
+
+            cancelled_caught = cs.cancelled_caught
+            sender_done.set()
+
+        async with trio.open_nursery() as tn:
+            tn.start_soon(
+                send_first,
             )
             await stream.send_entered.wait()
-            tn.cancel_scope.cancel()
+            sender_scopes[0].cancel()
+            await wait_all_tasks_blocked()
 
-        assert stream.closed
+            assert not stream.closed
+            assert not sender_done.is_set()
+
+            stream.release.set()
+            await sender_done.wait()
+
+            assert cancelled_caught
+            assert not stream.closed
+            assert count_frames(stream.wire) == 1
+
+            await transport.send(second_msg)
+            assert count_frames(stream.wire) == 2
+
+            tn.cancel_scope.cancel()
 
     trio.run(main)
 

tests/test_context_stream_semantics.py

@@ -11,9 +11,14 @@ from pathlib import Path
 import platform
 from pprint import pformat
 import sys
+from types import SimpleNamespace
 from typing import (
     Callable,
 )
+from unittest.mock import (
+    AsyncMock,
+    Mock,
+)
 
 import pytest
 import trio
@@ -26,6 +31,7 @@ from tractor import (
 from tractor._exceptions import (
     StreamOverrun,
     ContextCancelled,
+    TransportClosed,
 )
 from tractor.runtime._state import current_ipc_ctx
 
@@ -73,6 +79,88 @@ from tractor._testing import (
 #   with implicit stream closure on the cancelling end.
 
 
+def test_overrun_error_send_tolerates_transport_close(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    '''
+    Preserve a stream overrun when its error can not be shipped.
+
+    A full local stream buffer makes `Context._deliver_msg()` package
+    `StreamOverrun` for the remote sender. On Darwin, a concurrently
+    closing socket is wrapped as `TransportClosed`; allowing that
+    secondary error to escape replaces the primary overrun and crashes
+    the actor-wide RPC loop. This fake context forces that ordering and
+    proves failed error shipment reports non-delivery without raising.
+
+    '''
+    error_msg = tractor.msg.Error(
+        src_uid=('local', 'test'),
+        src_type_str='StreamOverrun',
+        boxed_type_str='StreamOverrun',
+        relay_path=[],
+        sender=('peer', 'test'),
+        cid='overrun',
+    )
+    packed: dict[str, object] = {}
+
+    def pack_overrun(
+        local_err: BaseException,
+        cid: str,
+        **kwargs,
+    ) -> tractor.msg.Error:
+        packed['local_err'] = local_err
+        packed['cid'] = cid
+        packed['kwargs'] = kwargs
+        return error_msg
+
+    monkeypatch.setattr(
+        'tractor._context.pack_from_raise',
+        pack_overrun,
+    )
+
+    async def main() -> None:
+        send_chan = Mock()
+        send_chan.send_nowait.side_effect = trio.WouldBlock
+        chan = SimpleNamespace(
+            aid=SimpleNamespace(uid=('peer', 'test')),
+            send=AsyncMock(
+                side_effect=TransportClosed('peer closed'),
+            ),
+        )
+        local_aid = SimpleNamespace(
+            name='local',
+            reprol=lambda: 'local@test',
+        )
+        ctx = SimpleNamespace(
+            cid='overrun',
+            chan=chan,
+            _send_chan=send_chan,
+            _nsf='tests:overrun',
+            side='parent',
+            peer_side='child',
+            _portal=object(),
+            _task=None,
+            repr_api='Context',
+            repr_caller='test',
+            _in_overrun=False,
+            _actor=SimpleNamespace(aid=local_aid),
+            _stream_opened=True,
+            _allow_overruns=False,
+        )
+        msg = tractor.msg.Yield(
+            cid=ctx.cid,
+            pld='payload',
+        )
+
+        delivered: bool = await Context._deliver_msg(ctx, msg)
+
+        assert delivered is False
+        assert isinstance(packed['local_err'], StreamOverrun)
+        assert packed['cid'] == ctx.cid
+        chan.send.assert_awaited_once_with(error_msg)
+
+    trio.run(main)
+
 _state: bool = False
 
 

tractor/_context.py

@@ -2021,9 +2021,16 @@ class Context:
                     await chan.send(err_msg)
                     return True
 
-                # XXX: local consumer has closed their side of
+                # XXX: the local consumer may have closed its side of
-                # the IPC so cancel the far end streaming task
+                # the IPC, in which case context/channel teardown owns
-                except trio.BrokenResourceError:
+                # 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.
+                except (
+                    TransportClosed,
+                    trio.BrokenResourceError,
+                ):
                     log.warning(
                         'Channel for ctx is already closed?\n'
                         f'|_{chan}\n'

tractor/ipc/_transport.py

@@ -498,13 +498,30 @@ class MsgpackTransport(MsgTransport):
             # https://stackoverflow.com/a/54027962
             size: bytes = struct.pack("<I", len(bytes_data))
             try:
-                return await self.stream.send_all(size + bytes_data)
+                # Every IPC msg is length-prefixed and all contexts
-            except trio.Cancelled:
+                # on this actor pair share one transport stream. If
-                # `send_all()` may have written a partial frame. The
+                # cancellation interrupts `send_all()`, an unknown
-                # stream can not safely carry another framed msg.
+                # 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.
+                #
+                # 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.
+                #
+                # This can delay cancellation while a peer is not
+                # reading; peer/channel teardown must close the stream
+                # to unblock a permanently stalled socket write.
                 with trio.CancelScope(shield=True):
-                    await self.stream.aclose()
+                    await self.stream.send_all(size + bytes_data)
-                raise
+
+                await trio.lowlevel.checkpoint_if_cancelled()
+                return None
+
             except (
                 trio.BrokenResourceError,
                 trio.ClosedResourceError,