Fix swallowed `TransportClosed` on ctx scope exit

EOF-driven cancellation can be absorbed by the ctx nursery before
teardown sees a disconnected channel and skips its recorded error.
Route an unfinished transport fault through `wait_for_result()` so
it raises the original error without waiting on broken IPC or
manufacturing a cancellation requester.

Share `ContextStatus._outcome_is_unfinished()` between EOF
notification and ctx teardown. Both buffered `Return` msgs and
decoded falsey results remain completed outcomes; notification
requires no existing error while teardown raises the recorded one.

Deats,
- Reconcile the close example and advanced-fault matrix with prompt
  fault propagation, preserving break order and pre-`Stop` cases.
- Check exception leaves, socket causes, cancellation provenance,
  and post-close send/receive error precedence and identity. Keep
  synthetic Ctrl-C for the healthy no-break cases.
- Cover buffered/consumed `None`, `False`, `0`, and empty-string
  results across EOF, including final value and type preservation.
- Add `assert_child_reaped()` for retained process handles. Check
  both backend completion and PID disappearance without cleanup.

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

examples/advanced_faults/ipc_failure_during_stream.py

@@ -1,111 +1,78 @@
 '''
-Complex edge case where during real-time streaming the IPC tranport
+Detect a closed IPC transport during real-time streaming.
-channels are wiped out (purposely in this example though it could have
+
-been an outage) and we want to ensure that despite being in debug mode
+Detected closure must fail an unfinished context promptly, including
-(or not) the user can sent SIGINT once they notice the hang and the
+when the peer sent a stream stop first. Only the no-break example
-actor tree will eventually be cancelled without leaving any zombies.
+simulates a user pressing Ctrl-C to cancel a healthy actor tree.
 
 '''
 from contextlib import asynccontextmanager as acm
-from functools import partial
 
 from tractor import (
     open_nursery,
     context,
     Context,
-    ContextCancelled,
-    MsgStream,
     _testing,
-    trionics,
-    TransportClosed,
 )
 import trio
-import pytest
-
-
-async def break_ipc_then_error(
-    stream: MsgStream,
-    break_ipc_with: str|None = None,
-    pre_close: bool = False,
-):
-    await _testing.break_ipc(
-        stream=stream,
-        method=break_ipc_with,
-        pre_close=pre_close,
-    )
-    async for msg in stream:
-        await stream.send(msg)
-
-    assert 0
-
-
-async def iter_ipc_stream(
-    stream: MsgStream,
-    break_ipc_with: str|None = None,
-    pre_close: bool = False,
-):
-    async for msg in stream:
-        await stream.send(msg)
 
 
 @context
 async def recv_and_spawn_net_killers(
-
     ctx: Context,
     break_ipc_after: bool|int = False,
     pre_close: bool = False,
-
 ) -> None:
     '''
-    Receive stream msgs and spawn some IPC killers mid-stream.
+    Echo messages until a single, sequence-numbered IPC break.
+
+    Break before echoing the trigger so the caller is waiting for
+    that reply, rather than racing another send against peer close.
+    Keep this context unfinished: stream stop is not a final result.
 
     '''
-    broke_ipc: bool = False
     await ctx.started()
-    async with (
+    async with ctx.open_stream() as stream:
-        ctx.open_stream() as stream,
-        trionics.collapse_eg(),
-        trio.open_nursery() as tn,
-    ):
         async for i in stream:
-            print(f'child echoing {i}')
-            if not broke_ipc:
-                await stream.send(i)
-            else:
-                await trio.sleep(0.01)
-
             if (
                 break_ipc_after
                 and
                 i >= break_ipc_after
             ):
-                broke_ipc = True
+                if pre_close:
-                tn.start_soon(
+                    await stream.aclose()
-                    iter_ipc_stream,
+                    # Protocol Stop closes the stream, not its IPC.
-                    stream,
+                    assert stream.closed
-                )
+                    assert ctx.chan.connected()
-                tn.start_soon(
+                await _testing.break_ipc(
-                    partial(
+                    stream=stream,
-                        break_ipc_then_error,
+                    pre_close=False,
-                        stream=stream,
-                        pre_close=pre_close,
-                    )
                 )
+                await trio.sleep_forever()
+
+            print(f'child echoing {i}')
+            await stream.send(i)
 
 
 @acm
 async def stuff_hangin_ctlc(timeout: float = 1) -> None:
+    '''
+    Simulate Ctrl-C when a healthy stream makes no progress.
 
+    Detected-fault runs use a failing deadline instead, so a timeout
+    cannot masquerade as successful transport-failure propagation.
+
+    '''
     with trio.move_on_after(timeout) as cs:
         yield timeout
 
     if cs.cancelled_caught:
-        # pretend to be a user seeing no streaming action
+        # Pretend to be a user seeing no streaming action,
-        # thinking it's a hang, and then hitting ctl-c..
+        # thinking it's a hang, and then hitting Ctrl-C.
         print(
-            f"i'm a user on the PARENT side and thingz hangin "
+            f'PARENT streaming stalled after {timeout}s\n'
-            f'after timeout={timeout} ???\n\n'
+            f'\n'
-            'MASHING CTlR-C..!?\n'
+            f'MASHING Ctrl-C..!\n'
         )
         raise KeyboardInterrupt
 
@@ -115,153 +82,85 @@ async def main(
     start_method: str = 'trio',
     loglevel: str = 'cancel',
 
-    # by default we break the parent IPC first (if configured to break
+    # By default the parent breaks first, if configured to break
-    # at all), but this can be changed so the child does first (even if
+    # at all. Reverse the thresholds to make the child break first.
-    # both are set to break).
     break_parent_ipc_after: int|bool = False,
     break_child_ipc_after: int|bool = False,
     pre_close: bool = False,
     tpt_proto: str = 'tcp',
-
+    reg_addr: tuple[str, int|str]|None = None,
 ) -> None:
+    '''
+    Run a bounded close scenario or demonstrate Ctrl-C cancellation.
 
-    async with (
+    Tests pass their unique registrar address explicitly. Standalone
-        open_nursery(
+    invocation retains the runtime's default address selection.
-            start_method=start_method,
 
-            # NOTE: even debugger is used we shouldn't get
+    '''
-            # a hang since it never engages due to broken IPC
+    registry_kwargs = {}
-            debug_mode=debug_mode,
+    if reg_addr is not None:
-            loglevel=loglevel,
+        registry_kwargs = {
-            enable_transports=[tpt_proto],
+            'registry_addrs': [reg_addr],
+            'ensure_registry': True,
+        }
 
-        ) as an,
+    async with open_nursery(
-    ):
+        start_method=start_method,
-        sub_name: str = 'chitty_hijo'
+        # Even with debugging enabled, broken IPC must not hang
+        # trying to engage a debugger through the closed channel.
+        debug_mode=debug_mode,
+        loglevel=loglevel,
+        enable_transports=[tpt_proto],
+        **registry_kwargs,
+    ) as an:
         portal = await an.start_actor(
-            sub_name,
+            'chitty_hijo',
             enable_modules=[__name__],
         )
+        breaking = (
+            break_parent_ipc_after
+            or
+            break_child_ipc_after
+        )
+        async with portal.open_context(
+            recv_and_spawn_net_killers,
+            break_ipc_after=break_child_ipc_after,
+            pre_close=pre_close,
+        ) as (ctx, sent):
+            # A transport failure is not a remote cancellation RPC;
+            # do not require ContextCancelled or invent a canceller.
+            with trio.fail_after(3):
+                async with ctx.open_stream() as stream:
+                    for i in range(1000):
+                        if (
+                            break_parent_ipc_after
+                            and
+                            i >= break_parent_ipc_after
+                        ):
+                            # TODO: other methods? See break_ipc().
+                            # await ctx.chan.send(None)
+                            socket = ctx.chan.transport.stream
+                            # await socket.send_eof()
+                            await socket.aclose()
+                            assert not ctx.chan.connected()
+                            await trio.sleep_forever()
 
-        async with (
-            stuff_hangin_ctlc(timeout=2) as timeout,
-            _testing.expect_ctxc(
-                yay=(
-                    break_parent_ipc_after
-                    or
-                    break_child_ipc_after
-                ),
-                # TODO: we CAN'T remove this right?
-                # since we need the ctxc to bubble up from either
-                # the stream API after the `None` msg is sent
-                # (which actually implicitly cancels all remote
-                # tasks in the hijo) or from simluated
-                # KBI-mash-from-user
-                # or should we expect that a KBI triggers the ctxc
-                # and KBI in an eg?
-                reraise=True,
-            ),
-
-            portal.open_context(
-                recv_and_spawn_net_killers,
-                break_ipc_after=break_child_ipc_after,
-                pre_close=pre_close,
-            ) as (ctx, sent),
-        ):
-            rx_eoc: bool = False
-            ipc_break_sent: bool = False
-            async with ctx.open_stream() as stream:
-                for i in range(1000):
-
-                    if (
-                        break_parent_ipc_after
-                        and
-                        i > break_parent_ipc_after
-                        and
-                        not ipc_break_sent
-                    ):
-                        print(
-                            '#################################\n'
-                            'Simulating PARENT-side IPC BREAK!\n'
-                            '#################################\n'
-                        )
-
-                        # TODO: other methods? see break func above.
-                        # await stream._ctx.chan.send(None)
-                        # await stream._ctx.chan.transport.stream.send_eof()
-                        await stream._ctx.chan.transport.stream.aclose()
-                        ipc_break_sent = True
-
-                    # it actually breaks right here in the
-                    # mp_spawn/forkserver backends and thus the
-                    # zombie reaper never even kicks in?
-                    try:
-                        print(f'parent sending {i}')
                         await stream.send(i)
-                    except ContextCancelled as ctxc:
+                        if breaking:
-                        print(
-                            'parent received ctxc on `stream.send()`\n'
-                            f'{ctxc}\n'
-                        )
-                        assert 'root' in ctxc.canceller
-                        assert sub_name in ctx.canceller
-
-                        # TODO: is this needed or no?
-                        raise
-
-                    except (
-                        trio.ClosedResourceError,
-                        TransportClosed,
-                    ) as _tpt_err:
-                        # NOTE: don't send if we already broke the
-                        # connection to avoid raising a closed-error
-                        # such that we drop through to the ctl-c
-                        # mashing by user.
-                        with trio.CancelScope(shield=True):
-                            await trio.sleep(0.01)
-
-                    # timeout: int = 1
-                    # with trio.move_on_after(timeout) as cs:
-                    async with stuff_hangin_ctlc() as timeout:
-                        print(
-                            f'PARENT `stream.receive()` with timeout={timeout}\n'
-                        )
-                        # NOTE: in the parent side IPC failure case this
-                        # will raise an ``EndOfChannel`` after the child
-                        # is killed and sends a stop msg back to it's
-                        # caller/this-parent.
-                        try:
                             rx = await stream.receive()
-                            print(
+                        else:
-                                "I'm a happy PARENT user and echoed to me is\n"
+                            async with stuff_hangin_ctlc():
-                                f'{rx}\n'
+                                rx = await stream.receive()
-                            )
+                        assert rx == i
-                        except trio.EndOfChannel:
-                            rx_eoc: bool = True
-                            print('MsgStream got EoC for PARENT')
-                            raise
 
-            print(
+                if breaking:
-                'Streaming finished and we got Eoc.\n'
+                    # open_stream() absorbs a peer's stop/EoC, but
-                'Canceling `.open_context()` in root with\n'
+                    # open_context() still needs a result. EOF must
-                'CTlR-C..'
+                    # wake this wait without a later user SIGINT.
-            )
+                    await ctx.wait_for_result()
-            if rx_eoc:
+                    raise AssertionError('broken ctx returned?')
-                assert stream.closed
-                try:
-                    await stream.send(i)
-                    pytest.fail('stream not closed?')
-                except (
-                    TransportClosed,
-                    trio.ClosedResourceError,
-                    trio.EndOfChannel,
-                ) as send_err:
-                    if rx_eoc:
-                        assert send_err is stream._eoc
-                    else:
-                        assert send_err is stream._closed
 
-            raise KeyboardInterrupt
+                raise KeyboardInterrupt
 
 
 if __name__ == '__main__':

tests/test_advanced_faults.py

@@ -3,7 +3,7 @@ Sketchy network blackoutz, ugly byzantine gens, puedes eschuchar la
 cancelacion?..
 
 '''
-from functools import partial
+from contextlib import asynccontextmanager as acm
 from types import ModuleType
 
 import pytest
@@ -11,57 +11,184 @@ from _pytest.pathlib import import_path
 import trio
 import tractor
 from tractor import (
+    Portal,
     TransportClosed,
 )
+from tractor.runtime._supervise import ActorNursery
 from tractor._testing import (
     examples_dir,
     break_ipc,
+    assert_child_reaped,
 )
 
 
+@pytest.fixture
+def reaped_children(monkeypatch: pytest.MonkeyPatch):
+    '''
+    Retain spawned process handles and check synchronous reaping.
+
+    Snapshot each ActorNursery._children entry at publication, before
+    failure teardown removes it. Check both backend completion and
+    PID disappearance after trio.run(), without an external reaper.
+
+    '''
+    procs = []
+    start_actor = ActorNursery.start_actor
+
+    async def record_child(an, *args, **kwargs):
+        '''
+        Record only children spawned by this test's actor nursery.
+
+        '''
+        portal = await start_actor(an, *args, **kwargs)
+        procs.append(an._children[portal.channel.aid.uid][1])
+        return portal
+
+    monkeypatch.setattr(ActorNursery, 'start_actor', record_child)
+    yield procs
+    for proc in procs:
+        assert_child_reaped(proc)
+
+
+@pytest.fixture
+def fault_contexts(monkeypatch: pytest.MonkeyPatch):
+    '''
+    Observe fault state without altering cancellation or propagation.
+
+    Collect the entered `tractor.Context`s in a `list` delivered
+    to the consuming test. This allows introspecting a ctx's recorded
+    fault even if teardown incorrectly swallows that failure.
+
+    On teardown ensure transport-loss does not set
+    `ContextStatus._canceller`: connection loss alone supplies no
+    remote cancellation or task-error provenance. For the simulated
+    user-interrupt cases, ensure `Context.cancel_called` is True
+    because teardown requested cancellation of the remote task.
+
+    '''
+    contexts: list[tractor.Context] = []
+    open_context = Portal.open_context
+
+    @acm
+    async def record_context(portal, func, **kwargs):
+        '''
+        Hook into all `Portal.open_context() as pair` calls, stashing
+        their contexts in a `list` for consumption by dependent
+        tests for introspection.
+
+        '''
+        ctx: tractor.Context|None = None
+        try:
+            async with open_context(
+                portal,
+                func,
+                **kwargs,
+            ) as (ctx, sent):
+                if func.__name__ in {
+                    'recv_and_spawn_net_killers',
+                    'break_ipc_after_started',
+                }:
+                    contexts.append(ctx)
+                yield (ctx, sent)
+        finally:
+            if (
+                ctx in contexts
+                and
+                isinstance(ctx.maybe_error, TransportClosed)
+            ):
+                # MsgStream closure is distinct from transport loss.
+                # After teardown, send prioritizes the ctx error,
+                # while receive retains the stream's terminal marker.
+                stream = ctx._stream
+                assert stream.closed
+                assert not ctx.chan.connected()
+                with pytest.raises(TransportClosed) as err:
+                    await stream.send('after-close')
+                assert err.value is ctx.maybe_error
+                terminal = stream._eoc or stream._closed
+                assert isinstance(terminal, BaseException)
+                with pytest.raises(type(terminal)) as err:
+                    await stream.receive()
+                assert err.value is terminal
+
+    monkeypatch.setattr(Portal, 'open_context', record_context)
+    yield contexts
+    for ctx in contexts:
+        if isinstance(ctx.maybe_error, TransportClosed):
+            assert ctx.canceller is None
+        elif isinstance(ctx._status._local_error, KeyboardInterrupt):
+            assert ctx.cancel_called
+
+
+def assert_transport_closed(
+    exc: TransportClosed|BaseExceptionGroup,
+    local: bool,
+) -> tuple[TransportClosed, ...]:
+    '''
+    Validate every failure leaf, including its socket-level source.
+
+    Parent close interrupts receive with ClosedResourceError.
+    Peer close can report clean EOF or a reset, depending on buffered
+    traffic. No arbitrary exception or remote cancellation qualifies.
+
+    '''
+    if isinstance(exc, BaseExceptionGroup):
+        return tuple(
+            fault
+            for leaf in exc.exceptions
+            for fault in assert_transport_closed(leaf, local=local)
+        )
+
+    assert isinstance(exc, TransportClosed)
+    cause = exc.__cause__
+    assert exc.src_exc is cause
+    if local:
+        assert isinstance(cause, trio.ClosedResourceError)
+        assert 'closed locally' in str(exc)
+    elif cause is None:
+        assert 'gracefully closed' in str(exc)
+    else:
+        assert 'closed by peer' in str(exc)
+        assert isinstance(
+            cause,
+            (ValueError, trio.BrokenResourceError,
+             ConnectionResetError),
+        )
+        if isinstance(cause, ValueError):
+            assert 'unclean EOF' in str(cause)
+
+    # TODO, should we expect a certain exc-message (per tpt) too?
+    # e.g. cause.args[0] == 'another task closed this fd'
+    return (exc,)
+
+
 @pytest.mark.parametrize(
     'pre_aclose_msgstream',
-    [
+    [False, True],
-        False,
+    ids=['no_msgstream_aclose', 'pre_aclose_msgstream'],
-        True,
-    ],
-    ids=[
-        'no_msgstream_aclose',
-        'pre_aclose_msgstream',
-    ],
 )
 @pytest.mark.parametrize(
     'ipc_break',
     [
-        # no breaks
+        # No breaks: retain the healthy Ctrl-C cancellation contract.
-        {
+        dict(break_parent_ipc_after=False,
-            'break_parent_ipc_after': False,
+             break_child_ipc_after=False),
-            'break_child_ipc_after': False,
-        },
 
-        # only parent breaks
+        # Only parent breaks.
-        {
+        dict(break_parent_ipc_after=500,
-            'break_parent_ipc_after': 500,
+             break_child_ipc_after=False),
-            'break_child_ipc_after': False,
-        },
 
-        # only child breaks
+        # Only child breaks.
-        {
+        dict(break_parent_ipc_after=False,
-            'break_parent_ipc_after': False,
+             break_child_ipc_after=500),
-            'break_child_ipc_after': 500,
-        },
 
-        # both: break parent first
+        # Both configured: parent breaks first.
-        {
+        dict(break_parent_ipc_after=500,
-            'break_parent_ipc_after': 500,
+             break_child_ipc_after=800),
-            'break_child_ipc_after': 800,
-        },
-        # both: break child first
-        {
-            'break_parent_ipc_after': 800,
-            'break_child_ipc_after': 500,
-        },
 
+        # Both configured: child breaks first.
+        dict(break_parent_ipc_after=800,
+             break_child_ipc_after=500),
     ],
     ids=[
         'no_break',
@@ -75,238 +202,170 @@ def test_ipc_channel_break_during_stream(
     debug_mode: bool,
     loglevel: str,
     spawn_backend: str,
-    ipc_break: dict|None,
+    ipc_break: dict,
     pre_aclose_msgstream: bool,
     tpt_proto: str,
+    reg_addr: tuple[str, int|str],
+    reaped_children,
+    fault_contexts: list[tractor.Context],
 ):
     '''
-    Ensure we can have an IPC channel break its connection during
+    Detect local/peer closure without waiting for a later Ctrl-C.
-    streaming and it's still possible for the (simulated) user to kill
-    the actor tree using SIGINT.
 
-    We also verify the type of connection error expected in the parent
+    The old scenario required ContextCancelled and a hang even when
-    depending on which side if the IPC breaks first.
+    EOF had already notified the unfinished context. Sequence-number
+    thresholds choose the first breaker; the child closes before its
+    trigger reply, preventing the parent from advancing past it.
+
+    With pre-close, MsgStream absorbs the peer's `Stop`, but pending
+    Context.wait_for_result() must fail on transport loss. Check each
+    exception leaf and its source, bound teardown, and verify the
+    child is gone synchronously. No-break runs still raise synthetic
+    KeyboardInterrupt and reap the child through normal cancellation.
+
+    If Context._scope absorbs EOF-driven task cancellation, teardown
+    must raise the recorded fault despite a disconnected channel.
+    That path must not manufacture a cancel request or peer ack, and
+    must preserve the original TransportClosed object and its cause.
+
+    The fixture checks stream closure separately from transport
+    connectivity after unwinding, then retries send and receive to
+    verify their distinct error precedence and terminal identities.
 
     '''
-    if spawn_backend != 'trio':
+    if spawn_backend != 'trio' and debug_mode:
-        if debug_mode:
+        pytest.skip('debug_mode only supported on trio spawner')
-            pytest.skip('`debug_mode` only supported on `trio` spawner')
-
-        # non-`trio` spawners should never hit the hang condition that
-        # requires the user to do ctl-c to cancel the actor tree.
-        # expect_final_exc = trio.ClosedResourceError
-        expect_final_exc = TransportClosed
 
     mod: ModuleType = import_path(
-        examples_dir()
+        examples_dir() / 'advanced_faults'
-        / 'advanced_faults'
         / 'ipc_failure_during_stream.py',
         root=examples_dir(),
         consider_namespace_packages=False,
     )
-
+    parent = ipc_break['break_parent_ipc_after']
-    # by def we expect KBI from user after a simulated "hang
+    child = ipc_break['break_child_ipc_after']
-    # period" wherein the user eventually hits ctl-c to kill the
+    breaking = parent or child
-    # root-actor tree.
+    local = bool(
-    expect_final_exc: BaseException = KeyboardInterrupt
+        parent
-    expect_final_cause: BaseException|None = None
-
-    if (
-        # only expect EoC if trans is broken on the child side,
-        ipc_break['break_child_ipc_after'] is not False
         and
-        # AND we tell the child to call `MsgStream.aclose()`.
+        (
-        pre_aclose_msgstream
+            (not child)
-    ):
+            or
-        # expect_final_exc = trio.EndOfChannel
+            (parent < child)
-        # ^XXX NOPE! XXX^ since now `.open_stream()` absorbs this
-        # gracefully!
-        expect_final_exc = KeyboardInterrupt
-
-    # NOTE when ONLY the child breaks or it breaks BEFORE the
-    # parent we expect the parent to get a closed resource error
-    # on the next `MsgStream.receive()` and then fail out and
-    # cancel the child from there.
-    #
-    # ONLY CHILD breaks
-    if (
-        ipc_break['break_child_ipc_after']
-        and
-        ipc_break['break_parent_ipc_after'] is False
-    ):
-        # NOTE: we DO NOT expect this any more since
-        # the child side's channel will be broken silently
-        # and nothing on the parent side will indicate this!
-        # expect_final_exc = trio.ClosedResourceError
-
-        # NOTE: child will send a 'stop' msg before it breaks
-        # the transport channel BUT, that will be absorbed by the
-        # `ctx.open_stream()` block and thus the `.open_context()`
-        # should hang, after which the test script simulates
-        # a user sending ctl-c by raising a KBI.
-        if pre_aclose_msgstream:
-            expect_final_exc = KeyboardInterrupt
-
-            # XXX OLD XXX
-            # if child calls `MsgStream.aclose()` then expect EoC.
-            # ^ XXX not any more ^ since eoc is always absorbed
-            # gracefully and NOT bubbled to the `.open_context()`
-            # block!
-            # expect_final_exc = trio.EndOfChannel
-
-    # BOTH but, CHILD breaks FIRST
-    elif (
-        ipc_break['break_child_ipc_after'] is not False
-        and (
-            ipc_break['break_parent_ipc_after']
-            >
-            ipc_break['break_child_ipc_after']
         )
-    ):
+    )
-        if pre_aclose_msgstream:
-            expect_final_exc = KeyboardInterrupt
 
-    # NOTE when the parent IPC side dies (even if the child does as well
+    async def run():
-    # but the child fails BEFORE the parent) we always expect the
+        '''
-    # IPC layer to raise a closed-resource, NEVER do we expect
+        Bound the example, including actor nursery teardown.
-    # a stop msg since the parent-side ctx apis will error out
-    # IMMEDIATELY before the child ever sends any 'stop' msg.
-    #
-    # ONLY PARENT breaks
-    elif (
-        ipc_break['break_parent_ipc_after']
-        and
-        ipc_break['break_child_ipc_after'] is False
-    ):
-        expect_final_exc = tractor.TransportClosed
-        expect_final_cause = trio.ClosedResourceError
 
-    # BOTH but, PARENT breaks FIRST
+        '''
-    elif (
+        with trio.fail_after(10):
-        ipc_break['break_parent_ipc_after'] is not False
+            await mod.main(
-        and (
+                debug_mode=debug_mode,
-            ipc_break['break_child_ipc_after']
+                start_method=spawn_backend,
-            >
+                loglevel=loglevel,
-            ipc_break['break_parent_ipc_after']
+                pre_close=pre_aclose_msgstream,
-        )
+                tpt_proto=tpt_proto,
-    ):
+                reg_addr=reg_addr,
-        expect_final_exc = tractor.TransportClosed
+                **ipc_break,
-        expect_final_cause = trio.ClosedResourceError
-
-    with pytest.raises(
-        expected_exception=(
-            expect_final_exc,
-            ExceptionGroup,
-        ),
-    ) as excinfo:
-        try:
-            trio.run(
-                partial(
-                    mod.main,
-                    debug_mode=debug_mode,
-                    start_method=spawn_backend,
-                    loglevel=loglevel,
-                    pre_close=pre_aclose_msgstream,
-                    tpt_proto=tpt_proto,
-                    **ipc_break,
-                )
-            )
-        except KeyboardInterrupt as _kbi:
-            kbi = _kbi
-            if expect_final_exc is not KeyboardInterrupt:
-                pytest.fail(
-                    'Rxed unexpected KBI !?\n'
-                    f'{repr(kbi)}'
-                )
-
-            raise
-
-        except tractor.TransportClosed as _tc:
-            tc = _tc
-            if expect_final_exc is KeyboardInterrupt:
-                pytest.fail(
-                    'Unexpected transport failure !?\n'
-                    f'{repr(tc)}'
-                )
-            cause: Exception = tc.__cause__
-            assert (
-                # type(cause) is trio.ClosedResourceError
-                type(cause) is expect_final_cause
-
-                # TODO, should we expect a certain exc-message (per
-                # tpt) as well??
-                # and
-                # cause.args[0] == 'another task closed this fd'
             )
 
-            raise
+    if not breaking:
-
+        with pytest.raises(KeyboardInterrupt):
-    # get raw instance from pytest wrapper
+            trio.run(run)
-    value = excinfo.value
+    else:
-    if isinstance(value, ExceptionGroup):
+        with pytest.raises((TransportClosed, ExceptionGroup)) as err:
-        excs: tuple[Exception] = value.exceptions
+            trio.run(run)
-        assert (
+            ctx = fault_contexts[0]
-            len(excs) <= 2
+            pytest.fail(
-            and
+                'Transport failure was swallowed:\n'
-            all(
+                f'{ctx.repr_outcome()}\n'
-                isinstance(exc, TransportClosed)
+                f'error={ctx.maybe_error!r}'
-                for exc in excs
             )
-        )
+        faults = assert_transport_closed(err.value, local=local)
-        final_exc = excs[0]
+        ctx = fault_contexts[0]
-        assert isinstance(final_exc, expect_final_exc)
+        assert any(fault is ctx.maybe_error for fault in faults)
+        if ctx._scope.cancelled_caught:
+            error = ctx._status._remote_error
+            assert any(fault is error for fault in faults)
+            assert ctx._status._local_error is error
+            assert not ctx.cancel_called
+            assert not ctx.cancel_acked
+    assert len(reaped_children) == len(fault_contexts) == 1
 
 
 @tractor.context
 async def break_ipc_after_started(
     ctx: tractor.Context,
 ) -> None:
+    '''
+    Wait for the caller's stream barrier, then close unfinished IPC.
+
+    '''
     await ctx.started()
     async with ctx.open_stream() as stream:
+        assert await stream.receive() == 'ready'
 
         # TODO: make a test which verifies the error
         # for this, i.e. raises a `MsgTypeError`
         # await ctx.chan.send(None)
 
-        await break_ipc(
+        await break_ipc(stream=stream, pre_close=True)
-            stream=stream,
+        await trio.sleep_forever()
-            pre_close=True,
-        )
-        print('child broke IPC and terminating')
 
 
-def test_stream_closed_right_after_ipc_break_and_zombie_lord_engages():
+def test_stream_closed_right_after_ipc_break_and_zombie_lord_engages(
+    reg_addr: tuple[str, int|str],
+    tpt_proto: str,
+    spawn_backend: str,
+    loglevel: str,
+    reaped_children,
+    fault_contexts,
+):
     '''
-    Verify that is a subactor's IPC goes down just after bringing up
+    Reap a child after detected close without a later user SIGINT.
-    a stream the parent can trigger a SIGINT and the child will be
+
-    reaped out-of-IPC by the localhost process supervision machinery:
+    Previously the parent slept, timed out its context, then raised
-    aka "zombie lord".
+    KeyboardInterrupt to engage process supervision ("zombie lord").
+    EOF now notifies the unfinished context immediately. A ready
+    message ensures both streams exist before the child sends `Stop`
+    and closes IPC; the caller then waits for the missing result.
+
+    A bounded TransportClosed exit proves fault-driven teardown, and
+    reaped_children checks process completion and PID disappearance
+    synchronously. Timer-driven cancellation cannot pass this test.
 
     '''
     async def main():
-        with trio.fail_after(3):
+        '''
-            async with tractor.open_nursery() as an:
+        Await transport failure and bound the entire nursery exit.
+
+        '''
+        with trio.fail_after(10):
+            async with tractor.open_nursery(
+                registry_addrs=[reg_addr],
+                ensure_registry=True,
+                enable_transports=[tpt_proto],
+                start_method=spawn_backend,
+                loglevel=loglevel,
+            ) as an:
                 portal = await an.start_actor(
                     'ipc_breaker',
                     enable_modules=[__name__],
                 )
+                async with portal.open_context(
+                    break_ipc_after_started,
+                ) as (ctx, sent):
+                    async with ctx.open_stream() as stream:
+                        await stream.send('ready')
+                        await stream.receive()
+                    await ctx.wait_for_result()
+                    pytest.fail('broken context returned a result')
 
-                with trio.move_on_after(1):
+    with pytest.raises((TransportClosed, ExceptionGroup)) as err:
-                    async with (
-                        portal.open_context(
-                            break_ipc_after_started
-                        ) as (ctx, sent),
-                    ):
-                        async with ctx.open_stream():
-                            await trio.sleep(0.5)
-
-                        print('parent waiting on context')
-
-                print(
-                    'parent exited context\n'
-                    'parent raising KBI..\n'
-                )
-                raise KeyboardInterrupt
-
-    with pytest.raises(KeyboardInterrupt):
         trio.run(main)
+    faults = assert_transport_closed(err.value, local=False)
+    ctx = fault_contexts[0]
+    assert any(fault is ctx.maybe_error for fault in faults)
+    assert len(reaped_children) == len(fault_contexts) == 1

tests/test_ipc_eof.py

@@ -19,7 +19,8 @@ _streams: dict[str, tractor.Context] = {}
 @tractor.context
 async def stream_peer(
     ctx: tractor.Context,
-) -> None:
+    result_value: None|bool|int|str = None,
+) -> None|bool|int|str:
     '''
     Publish a live stream and echo until its caller finishes.
 
@@ -32,6 +33,7 @@ async def stream_peer(
             if value == 'finish':
                 break
             await stream.send(value)
+    return result_value
 
 
 async def send_peer_eof(cid: str) -> None:
@@ -52,6 +54,7 @@ async def consume_peer(
     owner: tractor.Context,
     wait_on: str,
     reg_addr,
+    result_value: None|bool|int|str,
 ) -> bool:
     '''
     Exercise a sibling connection outside child supervision.
@@ -114,7 +117,10 @@ async def consume_peer(
             ):
                 try:
                     async with (
-                        peer.open_context(stream_peer) as (ctx, _),
+                        peer.open_context(
+                            stream_peer,
+                            result_value=result_value,
+                        ) as (ctx, _),
                         ctx.open_stream() as stream,
                     ):
                         assert await stream.receive() == 'ready'
@@ -139,7 +145,7 @@ async def consume_peer(
                             if wait_on == 'returned':
                                 assert (
                                     await ctx.wait_for_result()
-                                ) is None
+                                ) == result_value
                         # Export the cid only after stream startup
                         # and any graceful Return delivery. The root
                         # uses it to select the faulted connection.
@@ -153,6 +159,11 @@ async def consume_peer(
                             await paused.wait()
                         if not graceful:
                             pytest.fail('resumed without EOF')
+                    if graceful:
+                        assert ctx._status._result == result_value
+                        assert type(ctx._status._result) is type(
+                            result_value,
+                        )
                 except TransportClosed as exc:
                     assert not graceful
                     assert exc is ctx._status._remote_error
@@ -212,28 +223,35 @@ async def consume_peer(
 
 
 @pytest.mark.parametrize(
-    'wait_on',
+    'wait_on, result_value',
     [
         # Unfinished context blocked in its receive API: wake it
         # without requiring an outbound write to discover EOF.
-        'receive',
+        ('receive', None),
 
         # Unfinished context blocked on an application Event:
         # exercise the proposed interruption of non-IPC awaits.
-        'event',
+        ('event', None),
 
         # Final Return already consumed: preserve success when the
         # connection closes after its remote task has completed.
-        'returned',
+        ('returned', None),
+        ('returned', False),
+        ('returned', 0),
+        ('returned', ''),
 
         # Final Return received but not drained: preserve the same
         # success even though the owner has not requested its result.
-        'buffered',
+        ('buffered', None),
+        ('buffered', False),
+        ('buffered', 0),
+        ('buffered', ''),
     ],
 )
 def test_discovered_peer_eof(
     reg_addr,
     wait_on: str,
+    result_value: None|bool|int|str,
 ) -> None:
     '''
     Exercise the proposed EOF-driven cancellation of context owners.
@@ -257,6 +275,10 @@ def test_discovered_peer_eof(
     value or leaving it buffered while the owner pauses. Observing
     delivery and the EOF report synchronizes those races without a
     timing sleep. Neither completed owner may be cancelled by EOF.
+    Falsey payloads must remain completed outcomes in both states;
+    only the `Unresolved` sentinel permits EOF notification. Check
+    the final value and type after teardown so False and 0 cannot
+    accidentally pass as each other.
 
     '''
     async def main():
@@ -285,6 +307,7 @@ def test_discovered_peer_eof(
                 consume_peer,
                 wait_on=wait_on,
                 reg_addr=reg_addr,
+                result_value=result_value,
             ) as (ctx, cid):
                 # consume_peer exports cid after both owners are
                 # blocked; only then break their shared connection.

tractor/_context.py

@@ -223,6 +223,23 @@ class ContextStatus(pretty_struct.Struct):
         '''
         return self._result is not Unresolved
 
+    def _outcome_is_unfinished(self) -> bool:
+        '''
+        Neither a received outcome nor a decoded result is available.
+
+        Shared by `process_messages()` on EOF notification and
+        `open_context_from_portal()` after absorbed scope
+        cancellation.
+        A buffered `Return` precedes decoding its payload; falsey
+        decoded values are still results, not `Unresolved`.
+
+        '''
+        return (
+            self._outcome_msg is Unresolved
+            and
+            self._result is Unresolved
+        )
+
     def pformat(
         self,
         field_indent: int = 2,
@@ -2623,24 +2640,46 @@ async def open_context_from_portal(
 
         raise  # duh
 
-    # no local scope error, the "clean exit with a result" case.
+    # No escaped scope error: either a result or absorbed task cancel.
     else:
-        if ctx.chan.connected():
+        if (
+            ctx.chan.connected()
+            or
+            (
+                # EOF cancels Context._scope, whose nursery absorbs
+                # its own trio.Cancelled. A disconnected channel must
+                # not bypass the recorded fault on that exit path.
+                # Share ContextStatus._outcome_is_unfinished() with
+                # process_messages(): preserve buffered Return and
+                # self-cancel ack state. Unlike that notification
+                # site, teardown checks for an already recorded
+                # TransportClosed rather than an absent error.
+                isinstance(
+                    ctx._status._remote_error,
+                    TransportClosed,
+                )
+                and
+                ctx._status._outcome_is_unfinished()
+            )
+        ):
             log.runtime(
                 'Waiting on final context result for\n'
                 f'peer: {uid}\n'
                 f'|_{ctx._task}\n'
             )
             # XXX NOTE XXX: the below call to
-            # `Context.wait_for_result()` will ALWAYS raise
+            # `Context.wait_for_result()` raises a recorded error
-            # a `ContextCancelled` (via an embedded call to
+            # (via an embedded call to
-            # `Context._maybe_raise_remote_err()`) IFF
+            # `Context._maybe_raise_remote_err()`) when
             # a `ContextStatus._remote_error` was set by the runtime
             # via a call to
             # `Context._maybe_cancel_and_set_remote_error()`.
-            # As per `Context._deliver_msg()`, that error IS
+            # As per `Context._deliver_msg()`, that error is set
-            # ALWAYS SET any time "child" side fails and causes
+            # when the "child" side fails and causes "parent side"
-            # "parent side" cancellation via a `ContextCancelled` here.
+            # cancellation. `process_messages()` also sets it for
+            # local transport loss: then we raise `TransportClosed`,
+            # not `ContextCancelled`, without waiting on IPC.
+            # The original cause and self-cancel ack policy survive.
             try:
                 result_or_err: Exception|Any = await ctx.wait_for_result()
             except BaseException as berr:

tractor/_testing/__init__.py

@@ -24,6 +24,10 @@ from contextlib import (
 )
 import os
 import pathlib
+from multiprocessing.process import BaseProcess
+
+import psutil
+import trio
 
 import tractor
 from tractor.devx.debug import (
@@ -37,6 +41,24 @@ from .fault_simulation import (
 )
 
 
+def assert_child_reaped(
+    proc: trio.Process|BaseProcess,
+) -> None:
+    '''
+    Check a test-owned child's completion without doing any cleanup.
+
+    The caller retains the handle from its own actor nursery. A
+    return code alone does not prove PID disappearance; check both.
+    Unlike `_reap.reap()`, this never discovers or signals processes.
+
+    '''
+    if isinstance(proc, trio.Process):
+        assert proc.returncode is not None
+    else:
+        assert proc.exitcode is not None
+    assert not psutil.pid_exists(proc.pid)
+
+
 # TODO, use dulwhich for this instead?
 # -> we're going to likely need it (or something similar)
 #   for supporting hot-coad reload feats eventually anyway!

tractor/runtime/_rpc.py

@@ -46,7 +46,6 @@ from trio import (
 from ..ipc import Channel
 from .._context import (
     Context,
-    Unresolved,
 )
 from .._exceptions import (
     ContextCancelled,
@@ -1369,13 +1368,13 @@ async def process_messages(
             )
 
     except TransportClosed as tc:
-        # MsgpackTransport._iter_pkts() raises TransportClosed for
+        # MsgpackTransport._iter_packets() raises TransportClosed for
         # a socket EOF (including an empty frame header), or wraps
         # a reset/read error. This is not MsgStream's EndOfChannel
         # after a protocol Stop, nor a Context's final Return.
         #
-        # Proposed policy, pending the transport-fault review:
+        # Once this channel is terminally closed, an unfinished
-        # an unfinished parent Context has lost its connection,
+        # parent Context has lost its connection,
         # NOT necessarily its remote task. The call below cancels
         # Context._scope even when its owner is blocked in unrelated
         # application code. That checkpoint sees trio.Cancelled;
@@ -1385,16 +1384,17 @@ async def process_messages(
         #
         # Actor._contexts can include multiple connections to one
         # peer. Match Context.chan by identity to leave other channels
-        # alone. Preserve received outcomes, including a Return which
+        # alone. ContextStatus._outcome_is_unfinished() is shared with
-        # its owner has not drained, instead of replacing them by EOF.
+        # context teardown to preserve received outcomes, including
+        # a Return its owner has not drained. Here only notify if no
+        # error is recorded yet; teardown instead raises that error.
         for ctx in tuple(actor._contexts.values()):
             if (
                 ctx.chan is chan
                 and ctx.side == 'parent'
                 and ctx._scope is not None
                 and ctx._status._remote_error is None
-                and ctx._status._outcome_msg is Unresolved
+                and ctx._status._outcome_is_unfinished()
-                and not ctx._status._final_result_is_set()
             ):
                 ctx._maybe_cancel_and_set_remote_error(tc)
 
@@ -1407,7 +1407,7 @@ async def process_messages(
         # TODO: maybe add a teardown handshake? and,
         # -[x] don't show this msg if it's an ephemeral discovery ep call?
         #  |_ see the below `.report_n_maybe_raise()` impl as well as
-        #     tc-exc input details in `MsgpackTCPStream._iter_pkts()`
+        #     tc-exc input details in `MsgpackTCPStream._iter_packets()`
         #     for different read-failure cases.
         # -[ ] figure out how this will break with other transports?
         tc.report_n_maybe_raise(