Close interrupted `MsgTransport.send()` streams

`SendStream.send_all()` can raise `trio.Cancelled` after writing an
arbitrary prefix of the four-byte length header and payload. The
peer can no longer distinguish a following msg boundary.

Close the stream under a shield before propagating cancellation so
callers can not append another msg to an indeterminate byte stream.

Caught-during: review remediation
Found-via: prospective P2 cancellation review

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

Prompt-IO: ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.md

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

ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.md

@@ -0,0 +1,36 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: ses_3e4c90d3eafeqHEtRYSIHgHhpA
+timestamp: 2026-08-18T19:30:01Z
+git_ref: bf06b4f8
+scope: code
+substantive: true
+raw_file: 20260818T193001Z_bf06b4f8_prompt_io.raw.md
+---
+
+## Prompt
+
+Fix cancellation during an indeterminate transport write without
+allowing a second RPC msg to corrupt the stream. Keep the transport
+fix separate from the higher-level `to_actor` remediation and explain
+the length-prefixed protocol rather than using unexplained "framed
+send" terminology.
+
+## Response summary
+
+Close a msgpack transport stream when cancellation interrupts its
+length-prefixed `send_all()` operation. The stream can no longer be
+safely reused because Trio cannot report how many bytes were written.
+
+## Files changed
+
+- `tractor/ipc/_transport.py` - close an interrupted send stream.
+- `tests/ipc/test_each_tpt.py` - cover cancellation during the write.
+
+## Human edits
+
+The human required this transport edge-case fix to land as its own
+behavioral commit with a detailed message. During staged review, the
+human also rejected the unexplained "framed send" wording and asked
+for terminology tied directly to the actual transport operation.

ai/prompt-io/opencode/20260818T193001Z_bf06b4f8_prompt_io.raw.md

@@ -0,0 +1,19 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-18T19:30:01Z
+git_ref: bf06b4f8
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+Prospective review found that cancellation can interrupt
+`MsgpackTransport.send()` after `send_all()` writes only part of its
+length-prefixed msg. Sending a cancellation request afterward can
+append another msg to the indeterminate stream and desynchronize the
+peer decoder.
+
+> `git diff HEAD~1..HEAD -- tractor/ipc/_transport.py tests/ipc/test_each_tpt.py`
+
+Close the stream under a cancellation shield when `send_all()` is
+cancelled. Cover the behavior with a fake stream that checkpoints
+inside the write and records forced closure.

tests/ipc/test_each_tpt.py

@@ -17,9 +17,65 @@ import trio
 import tractor
 from tractor import Actor
 from tractor.discovery import _addr
+from tractor.ipc._transport import MsgpackTransport
 from tractor.runtime import _state
 
 
+
+def test_cancelled_transport_send_closes_stream():
+    '''
+    Discard a transport after cancellation interrupts a framed send.
+
+    Trio's `SendStream.send_all()` may write an arbitrary frame prefix
+    before raising `Cancelled`. Sending another IPC msg afterward
+    would append a second frame and desynchronize the peer decoder.
+    The fake stream checkpoints after recording send entry; cancelling
+    its nursery deterministically interrupts that unknown-publication
+    window. Its close assertion proves the transport is made unusable
+    before another framed msg can be attempted.
+
+    '''
+    class PartialSendStream:
+        def __init__(self) -> None:
+            self.send_entered = trio.Event()
+            self.closed = False
+
+        async def send_all(
+            self,
+            data: bytes,
+        ) -> None:
+            assert data
+            self.send_entered.set()
+            await trio.sleep_forever()
+
+        async def aclose(self) -> None:
+            self.closed = True
+
+    async def main() -> None:
+        stream = PartialSendStream()
+        transport = object.__new__(MsgpackTransport)
+        transport.stream = stream
+        transport._send_lock = trio.StrictFIFOLock()
+
+        async with trio.open_nursery() as tn:
+            tn.start_soon(
+                transport.send,
+                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()
+
+        assert stream.closed
+
+    trio.run(main)
+
+
 @pytest.fixture
 def bindspace_dir_str() -> str:
 

tractor/ipc/_transport.py

@@ -499,6 +499,12 @@ class MsgpackTransport(MsgTransport):
             size: bytes = struct.pack("<I", len(bytes_data))
             try:
                 return await self.stream.send_all(size + bytes_data)
+            except trio.Cancelled:
+                # `send_all()` may have written a partial frame. The
+                # stream can not safely carry another framed msg.
+                with trio.CancelScope(shield=True):
+                    await self.stream.aclose()
+                raise
             except (
                 trio.BrokenResourceError,
                 trio.ClosedResourceError,