diff --git a/ai/prompt-io/opencode/20260819T234824Z_557065d8_prompt_io.md b/ai/prompt-io/opencode/20260819T234824Z_557065d8_prompt_io.md new file mode 100644 index 00000000..f7e96d12 --- /dev/null +++ b/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. diff --git a/ai/prompt-io/opencode/20260819T234824Z_557065d8_prompt_io.raw.md b/ai/prompt-io/opencode/20260819T234824Z_557065d8_prompt_io.raw.md new file mode 100644 index 00000000..43319582 --- /dev/null +++ b/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` diff --git a/tests/ipc/test_each_tpt.py b/tests/ipc/test_each_tpt.py index 7a7715f4..54fc91b6 100644 --- a/tests/ipc/test_each_tpt.py +++ b/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 - 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. + A cancelled `send_all()` may leave an arbitrary frame prefix on the + wire. Closing the actor-wide stream avoids decoder corruption but + also destroys unrelated contexts using that channel. The fake + stream publishes two header bytes and blocks, letting this test + cancel the sender inside frame publication. The sender must remain + blocked until the complete frame is written, then observe pending + 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() - await trio.sleep_forever() + if not self.wire: + 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(' 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, - tractor.msg.Start( - ns=__name__, - func='add_one', - kwargs={'n': 1}, - uid=('root', 'test'), - cid='partial-send', - ), + 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) diff --git a/tests/test_context_stream_semantics.py b/tests/test_context_stream_semantics.py index 7f89424c..e7535fd4 100644 --- a/tests/test_context_stream_semantics.py +++ b/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 diff --git a/tractor/_context.py b/tractor/_context.py index 8fd69743..214eea12 100644 --- a/tractor/_context.py +++ b/tractor/_context.py @@ -2021,9 +2021,16 @@ class Context: await chan.send(err_msg) return True - # XXX: local consumer has closed their side of - # the IPC so cancel the far end streaming task - except trio.BrokenResourceError: + # XXX: the local consumer may have closed its side of + # the IPC, in which case context/channel teardown owns + # 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' diff --git a/tractor/ipc/_transport.py b/tractor/ipc/_transport.py index c65c9d4a..c2fa9d9d 100644 --- a/tractor/ipc/_transport.py +++ b/tractor/ipc/_transport.py @@ -498,13 +498,30 @@ class MsgpackTransport(MsgTransport): # https://stackoverflow.com/a/54027962 size: bytes = struct.pack("