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/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/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
-            self.send_entered.set()
+            if not self.wire:
-            await trio.sleep_forever()
+                self.wire.extend(data[:2])
+                self.send_entered.set()
+                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
+
+        first_msg = 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(
-                transport.send,
+                send_first,
-                tractor.msg.Start(
-                    ns=__name__,
-                    func='add_one',
-                    kwargs={'n': 1},
-                    uid=('root', 'test'),
-                    cid='partial-send',
-                ),
             )
             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,