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

@@ -1,37 +0,0 @@
----
-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

@@ -1,26 +0,0 @@
----
-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

@@ -1,43 +0,0 @@
----
-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

@@ -1,31 +0,0 @@
----
-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,7 +27,6 @@ from pexpect.exceptions import (
 import tractor
 
 from .conftest import (
-    ansi_strip,
     do_ctlc,
     PROMPT,
     _pause_msg,
@@ -795,7 +794,6 @@ 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
@@ -835,9 +833,6 @@ 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)
@@ -847,9 +842,6 @@ 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'
@@ -896,12 +888,11 @@ def test_multi_nested_subactors_error_through_nurseries(
             "relay_uid=('spawn_until_1'",
         ]
 
-    transcript: str = '\n'.join(transcript_parts)
+    assert_before(
-    for part in expect_patts:
+        child,
-        assert part in transcript
+        expect_patts,
-
+    )
-    assert child.flag_eof
+    expect(child, EOF)
-    assert not child.isalive()
 
 
 # @pytest.mark.timeout(15)
@@ -1292,8 +1283,13 @@ def test_ctxep_pauses_n_maybe_ipc_breaks(
             )
             child.sendline('c')
             child.expect(EOF)
-            assert child.flag_eof
+            assert_before(
-            assert not child.isalive()
+                child,
+                ["tractor._exceptions.RemoteActorError: remote task raised a 'BdbQuit'",
+                 "bdb.BdbQuit",
+                 "('bp_boi'",
+                ]
+            )
             break  # end-of-test
 
         child.sendline('c')
@@ -1342,8 +1338,10 @@ def test_ctxep_pauses_n_maybe_ipc_breaks(
                 expect_prompt=False,
             )
             child.expect(EOF)
-            assert child.flag_eof
+            assert_before(
-            assert not child.isalive()
+                child,
+                ['KeyboardInterrupt'],
+            )
 
 
 def test_crash_handling_within_cancelled_root_actor(

tests/devx/test_tooling.py

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

tests/ipc/test_each_tpt.py

@@ -7,7 +7,6 @@ import os
 from pathlib import Path
 import socket
 import stat
-import struct
 import sys
 import tempfile
 from types import SimpleNamespace
@@ -15,7 +14,6 @@ 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
@@ -24,113 +22,57 @@ from tractor.runtime import _state
 
 
 
-def test_cancelled_transport_send_completes_frame():
+def test_cancelled_transport_send_closes_stream():
     '''
-    Finish an in-flight frame before delivering sender cancellation.
+    Discard a transport after cancellation interrupts a framed send.
 
-    A cancelled `send_all()` may leave an arbitrary frame prefix on the
+    Trio's `SendStream.send_all()` may write an arbitrary frame prefix
-    wire. Closing the actor-wide stream avoids decoder corruption but
+    before raising `Cancelled`. Sending another IPC msg afterward
-    also destroys unrelated contexts using that channel. The fake
+    would append a second frame and desynchronize the peer decoder.
-    stream publishes two header bytes and blocks, letting this test
+    The fake stream checkpoints after recording send entry; cancelling
-    cancel the sender inside frame publication. The sender must remain
+    its nursery deterministically interrupts that unknown-publication
-    blocked until the complete frame is written, then observe pending
+    window. Its close assertion proves the transport is made unusable
-    cancellation; a second complete frame proves channel reuse remains
+    before another framed msg can be attempted.
-    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 self.release.wait()
+            await trio.sleep_forever()
-                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(
+        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',
-        )
+                ),
-        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()
-            sender_scopes[0].cancel()
-            await wait_all_tasks_blocked()
-
-            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()
 
+        assert stream.closed
+
     trio.run(main)
 
 

tests/test_context_stream_semantics.py

@@ -11,14 +11,9 @@ 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
@@ -31,7 +26,6 @@ from tractor import (
 from tractor._exceptions import (
     StreamOverrun,
     ContextCancelled,
-    TransportClosed,
 )
 from tractor.runtime._state import current_ipc_ctx
 
@@ -79,88 +73,6 @@ 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,16 +2021,9 @@ class Context:
                     await chan.send(err_msg)
                     return True
 
-                # XXX: the local consumer may have closed its side of
+                # XXX: local consumer has closed their side of
-                # the IPC, in which case context/channel teardown owns
+                # the IPC so cancel the far end streaming task
-                # cancellation of the far-end streaming task. The same
+                except trio.BrokenResourceError:
-                # 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,30 +498,13 @@ class MsgpackTransport(MsgTransport):
             # https://stackoverflow.com/a/54027962
             size: bytes = struct.pack("<I", len(bytes_data))
             try:
-                # Every IPC msg is length-prefixed and all contexts
+                return await self.stream.send_all(size + bytes_data)
-                # on this actor pair share one transport stream. If
+            except trio.Cancelled:
-                # cancellation interrupts `send_all()`, an unknown
+                # `send_all()` may have written a partial frame. The
-                # frame prefix may already be on the wire; allowing
+                # stream can not safely carry another framed msg.
-                # 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.send_all(size + bytes_data)
+                    await self.stream.aclose()
-
+                raise
-                await trio.lowlevel.checkpoint_if_cancelled()
-                return None
-
             except (
                 trio.BrokenResourceError,
                 trio.ClosedResourceError,