Notify unfinished `Context` owners on IPC EOF

Save the proposed transport-fault handling for discovered peers.
`process_messages()` reports EOF then exits, leaving parent ctx
owners blocked until another send detects the loss. Wake unfinished
owners on that exact channel with `TransportClosed`, including
owners paused at application checkpoints unrelated to IPC.

This scope-cancellation policy remains provisional pending the
transport contract discussion. Saving the impl and regression cases
on this branch does not settle that policy or imply acceptance.

Deats,
- Match `Context.chan` by identity and notify all unfinished parent
  ctxs on the failed connection, leaving other channels alone.
- Preserve received outcomes and record buffered `Return` receipt
  using the `Unresolved` sentinel before EOF can replace success.
- Keep local transport loss distinct from a remote cancellation
  request: do not invent `Context._canceller` provenance.
- Preserve the original disconnect cause through owner cancellation
  and unwinding in `Context._maybe_raise_remote_err()`.

Also,
- Exercise two owners sharing a discovered sibling channel, blocked
  in stream receive or an application `Event`, plus a healthy peer
  that continues to echo after fault handling.
- Synchronize EOF injection after startup and task quiescence.
  Cover drained and buffered final results without timing sleeps.
- Regress cause preservation directly with a connection-reset cause
  and a distinct owner-unwinding exception.
- Carry the staged structured session provenance unchanged. Its
  broader cancellation observations do not expand this boundary;
  the root-owned `cancel_soon()` fix remains outside this save.

Exact-tree project tests remain pending in the commit plan. Earlier
session results in the provenance log are not passes for this tree.

Prompt-IO: ai/prompt-io/opencode/20260916_ib_disconnect_cleanup.md

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

ai/prompt-io/opencode/20260916_ib_disconnect_cleanup.md

@@ -0,0 +1,73 @@
+---
+model: openai/gpt-6-astra
+service: opencode
+scope: code
+substantive: true
+git_ref: 0d95a9e6
+raw_file: 20260916_ib_disconnect_cleanup.raw.md
+---
+
+## Prompt
+
+Fix runtime failures exposed when stopping IB Gateway: idle chart
+consumers miss disconnect notification, infected actors retain closed
+service nurseries, late SIGINT raises in cancel_soon, and creator shm
+segments survive actor exit. Develop on piker_pin in isolation and do
+not change the live piker setup.
+
+## Response summary
+
+Notify unfinished parent contexts on the exact failed IPC channel,
+preserve delivered outcomes and original transport causes, and record
+buffered Return outcomes before subsequent EOF. Schedule synchronous
+actor cancellation in the root nursery rather than a possibly closed
+service nursery, deduplicating repeated requests.
+
+## Files changed
+
+- tractor/runtime/_rpc.py: propagate transport closure to ctx owners.
+- tractor/_context.py: preserve transport cause and buffered outcomes.
+- tractor/runtime/_runtime.py: root-owned cancellation task.
+- tests/test_ipc_eof.py: active/idle peers and graceful EOF regressions.
+- tests/test_cancel_soon.py: closed-service cancellation and actual
+  creator-owned shm unlinking under native/infected runtimes.
+
+## Verification
+
+These are session observations, not authenticated passes for the
+individual commit-plan boundary trees. The shared provenance is carried
+by the final cancellation boundary and describes both runtime fixes.
+Exact-tree checks remain pending in the commit plan.
+
+Baseline EOF cases: three failures and two passes. Fixed TCP selection:
+34 passes; UDS EOF cases: five passes. Combined cancellation and EOF
+cases: nine passes. Extended actual-SHM cancellation test: four passes;
+original method injected into fresh actors: four expected failures.
+No user processes or shm names were touched; no installations or
+commits were made. Full Gateway/SIGINT/debugger sequence remains for
+integration validation; automatic IB reconnect is not implemented.
+
+## Human contributions
+
+The human supplied incident traces, selected piker_pin, identified
+dependent piker compatibility branches, and explicitly required worktree
+isolation and preserving the live setup. Acceptance remains pending.
+
+## Local review adjustment
+
+The human challenged treating a locally observed transport loss like
+a remote task failure/cancellation and requested readability changes.
+Removed invented peer attribution from Context._canceller, added
+assertions for absent cancellation provenance, restored the Return
+warning, expanded predicates, and documented test ordering and fault
+semantics. The proposed EOF-driven scope cancellation remains in place
+only pending a separate contract decision; test success is not policy
+acceptance. Piker-level explicit disconnect supervision is an alternative
+to changing the runtime default. Issues #400, #368, and #22 were cited
+by the human for the design discussion, not fetched in this pass.
+
+Review verification: five EOF tests and two service-nursery cases passed.
+Seven inter-peer cancellation cases could not start because disabling
+automatic reaping removes their track_orphaned_uds_per_test fixture.
+That coverage remains pending; no reaper was enabled against live state.
+The existing advanced-fault expectations also remain to be reconciled.

tests/test_ipc_eof.py

@@ -0,0 +1,322 @@
+'''
+Exercise the proposed discovered-peer EOF notification policy.
+
+'''
+from unittest.mock import Mock
+
+import pytest
+import trio
+from trio.testing import wait_all_tasks_blocked
+
+import tractor
+from tractor._exceptions import TransportClosed
+from tractor.msg.types import Return
+
+
+_streams: dict[str, tractor.Context] = {}
+
+
+@tractor.context
+async def stream_peer(
+    ctx: tractor.Context,
+) -> None:
+    '''
+    Publish a live stream and echo until its caller finishes.
+
+    '''
+    _streams[ctx.cid] = ctx
+    await ctx.started()
+    async with ctx.open_stream() as stream:
+        await stream.send('ready')
+        async for value in stream:
+            if value == 'finish':
+                break
+            await stream.send(value)
+
+
+async def send_peer_eof(cid: str) -> None:
+    '''
+    Close only this test channel's outgoing transport half.
+
+    This helper sends no Stop, Return, or Error. Keeping the read
+    half open avoids a local fd-close error crashing the peer actor
+    and confusing transport propagation with supervision.
+
+    '''
+    ctx = _streams[cid]
+    await ctx.chan._transport.stream.send_eof()
+
+
+@tractor.context
+async def consume_peer(
+    owner: tractor.Context,
+    wait_on: str,
+    reg_addr,
+) -> bool:
+    '''
+    Exercise a sibling connection outside child supervision.
+
+    '''
+    with trio.fail_after(10):
+        async with (
+            # Discovery creates a sibling channel, independent of
+            # the root actor's control connection to eof_peer.
+            tractor.find_actor(
+                'eof_peer',
+                registry_addrs=[reg_addr],
+            ) as peer,
+            # This second peer is a control: the fault on peer's
+            # channel must not terminate unrelated communications.
+            tractor.find_actor(
+                'healthy_peer',
+                registry_addrs=[reg_addr],
+            ) as healthy,
+            healthy.open_context(stream_peer) as (good, _),
+            good.open_stream() as good_stream,
+            trio.open_nursery() as tn,
+        ):
+            # owner.chan carries the root's orchestration messages;
+            # losing peer.channel must not lose the result channel.
+            assert peer.channel is not owner.chan
+            # Confirm the healthy context is live before injecting
+            # the fault, not merely connect to it after teardown.
+            assert await good_stream.receive() == 'ready'
+            errors: list[TransportClosed] = []
+
+            # Set only once BOTH affected context owners observe
+            # TransportClosed; one notified owner is insufficient.
+            finished = trio.Event()
+
+            # Model an application checkpoint unrelated to IPC.
+            # Never release it for fault cases: only the proposed
+            # context-scope cancellation may interrupt that wait.
+            # Completed contexts are explicitly released after EOF.
+            paused = trio.Event()
+
+            # A local report hook confirms process_messages handled
+            # EOF before we inspect a completed context's outcome.
+            eof_seen = trio.Event()
+            graceful = wait_on in {'returned', 'buffered'}
+            report = TransportClosed.report_n_maybe_raise
+
+            def report_eof(exc, *args, **kwargs):
+                # During this gate only the injected peer channel
+                # closes; healthy/control teardown happens later.
+                eof_seen.set()
+                return report(exc, *args, **kwargs)
+
+            # This sibling is a fresh test actor; observe the real
+            # handler without changing its exception policy.
+            TransportClosed.report_n_maybe_raise = report_eof
+
+            async def consume(
+                task_status=trio.TASK_STATUS_IGNORED,
+            ):
+                try:
+                    async with (
+                        peer.open_context(stream_peer) as (ctx, _),
+                        ctx.open_stream() as stream,
+                    ):
+                        assert await stream.receive() == 'ready'
+                        if graceful:
+                            # Observe Context._deliver_msg receipt,
+                            # without requiring the owner to drain
+                            # the final result from its feeder.
+                            returned = trio.Event()
+                            deliver = ctx._deliver_msg
+
+                            async def record_return(msg):
+                                result = await deliver(msg)
+                                if isinstance(msg, Return):
+                                    returned.set()
+                                return result
+
+                            ctx._deliver_msg = record_return
+                            # Ask the remote context to finish while
+                            # leaving its shared actor channel open.
+                            await stream.send('finish')
+                            await returned.wait()
+                            if wait_on == 'returned':
+                                assert (
+                                    await ctx.wait_for_result()
+                                ) is None
+                        # Export the cid only after stream startup
+                        # and any graceful Return delivery. The root
+                        # uses it to select the faulted connection.
+                        task_status.started(ctx)
+                        if graceful:
+                            await paused.wait()
+                            assert ctx._remote_error is None
+                        elif wait_on == 'receive':
+                            await stream.receive()
+                        else:
+                            await paused.wait()
+                        if not graceful:
+                            pytest.fail('resumed without EOF')
+                except TransportClosed as exc:
+                    assert not graceful
+                    assert exc is ctx._remote_error
+                    assert exc.__cause__ is exc._src_exc
+                    assert 'closed' in exc.message
+                    # This is a local transport observation, not
+                    # evidence that a peer requested cancellation.
+                    assert ctx.canceller is None
+                    assert not ctx.cancel_called
+                    assert not ctx.cancel_acked
+                    errors.append(exc)
+                    if len(errors) == 2:
+                        finished.set()
+
+            # Both contexts share one discovered channel but own
+            # separate local scopes. EOF must notify both owners.
+            first = await tn.start(consume)
+            second = await tn.start(consume)
+            assert first.chan is second.chan
+            # Neither readiness handshake alone proves the owner is
+            # blocked at the intended checkpoint. Quiescence does.
+            await wait_all_tasks_blocked()
+            original_send = peer.channel.send
+            checking_eof = True
+
+            async def send_after_notification(*args, **kwargs):
+                # Cleanup may send, but must not discover the EOF.
+                if checking_eof:
+                    assert first._remote_error is not None
+                    assert second._remote_error is not None
+                return await original_send(*args, **kwargs)
+
+            if not graceful:
+                peer.channel.send = send_after_notification
+            # Release the root's fault injection only after both
+            # owners are waiting and the no-send-detection guard is
+            # installed. The root uses a separate control channel.
+            await owner.started(first.cid)
+            try:
+                with trio.fail_after(2):
+                    await eof_seen.wait()
+                    if graceful:
+                        paused.set()
+                    else:
+                        await finished.wait()
+            finally:
+                checking_eof = False
+            assert len(errors) == (0 if graceful else 2)
+            assert paused.is_set() is graceful
+            assert good._remote_error is None
+            # Prove the unrelated peer still serves application
+            # traffic after EOF handling, not just that it has no
+            # recorded error yet.
+            await good_stream.send('still alive')
+            assert await good_stream.receive() == 'still alive'
+    return True
+
+
+@pytest.mark.parametrize(
+    'wait_on',
+    [
+        # Unfinished context blocked in its receive API: wake it
+        # without requiring an outbound write to discover EOF.
+        'receive',
+
+        # Unfinished context blocked on an application Event:
+        # exercise the proposed interruption of non-IPC awaits.
+        'event',
+
+        # Final Return already consumed: preserve success when the
+        # connection closes after its remote task has completed.
+        'returned',
+
+        # Final Return received but not drained: preserve the same
+        # success even though the owner has not requested its result.
+        'buffered',
+    ],
+)
+def test_discovered_peer_eof(
+    reg_addr,
+    wait_on: str,
+) -> None:
+    '''
+    Exercise the proposed EOF-driven cancellation of context owners.
+
+    Previously process_messages logged TransportClosed and returned,
+    leaving parent Context scopes alive until a later outbound send.
+    Two owners share a discovered channel, independently of the
+    ActorNursery's child-control channel. Both receive an initial
+    stream item, then block in receive() or on an unset Event.
+    Nursery.start() and wait_all_tasks_blocked() establish that
+    ordering before a control RPC half-closes the peer transport.
+
+    Under the proposed policy both owners must raise TransportClosed
+    before the deadline without sending to discover the failure.
+    The Event case deliberately asserts interruption of a non-IPC
+    checkpoint; whether to adopt that contract remains under review.
+    An unrelated peer stream must still echo afterward and close
+    gracefully. The original EOF cause must survive unwinding.
+
+    Graceful variants deliver Return before EOF, either draining its
+    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.
+
+    '''
+    async def main():
+        async with (
+            tractor.open_nursery(
+                registry_addrs=[reg_addr],
+            ) as an,
+        ):
+            # Retain a root-to-peer portal exclusively for injecting
+            # EOF. Its transport is not the sibling channel tested.
+            control = await an.start_actor(
+                'eof_peer',
+                enable_modules=[__name__],
+            )
+            await an.start_actor(
+                'healthy_peer',
+                enable_modules=[__name__],
+            )
+            # A separate consumer ensures failure cannot propagate
+            # merely through the root's child-supervision nursery.
+            consumer = await an.start_actor(
+                'consumer',
+                enable_modules=[__name__],
+            )
+            async with consumer.open_context(
+                consume_peer,
+                wait_on=wait_on,
+                reg_addr=reg_addr,
+            ) as (ctx, cid):
+                # consume_peer exports cid after both owners are
+                # blocked; only then break their shared connection.
+                await control.run(send_peer_eof, cid=cid)
+                assert await ctx.wait_for_result() is True
+
+            await an.cancel()
+
+    trio.run(main)
+
+
+def test_disconnect_cause_survives_unwind() -> None:
+    '''
+    Preserve the transport cause when cancelling a context owner.
+
+    Context result/stream receivers can supply their cancellation as
+    from_src_exc while unwinding. That must not replace the original
+    socket error in TransportClosed.__cause__. Exercise the actual
+    raise path with a reset cause and a distinct unwinding exception;
+    both exception identity and the original cause must survive.
+
+    '''
+    ctx = Mock()
+    ctx._is_self_cancelled.return_value = False
+    cause = ConnectionResetError('test peer reset')
+    error = TransportClosed('test disconnect', src_exc=cause)
+    with pytest.raises(TransportClosed) as raised:
+        tractor.Context._maybe_raise_remote_err(
+            ctx,
+            error,
+            from_src_exc=RuntimeError('owner unwinding'),
+        )
+    assert raised.value is error
+    assert error.__cause__ is cause
+    assert error.src_exc is cause

tractor/_context.py

@@ -828,6 +828,17 @@ class Context:
                 f'{error}'
             )
 
+        elif isinstance(error, TransportClosed):
+            # TransportClosed is observed locally, not an Error msg
+            # identifying a cancellation requester or failed task.
+            # Preserve Context._canceller rather than inventing a
+            # peer-originated cancellation from connection loss.
+            cid: str = self.cid
+            log.warning(
+                f'Locally observed transport fault for ctx {cid!r}\n'
+                f'{error!r}'
+            )
+
         elif isinstance(error, MsgTypeError):
             msgerr = True
             self._canceller = error.src_uid
@@ -855,7 +866,11 @@ class Context:
                 f'{error!r}'
             )
 
-        if self._canceller is None:
+        if (
+            self._canceller is None
+            and
+            not isinstance(error, TransportClosed)
+        ):
             log.error('Ctx has no canceller set!?')
 
         cs: trio.CancelScope = self._scope
@@ -878,7 +893,8 @@ class Context:
         # `.open_context()`  block.
         if (
             cs
-            and not cs.cancel_called
+            and
+            not cs.cancel_called
 
             # XXX this is an expected cancel request response
             # message and we **don't need to raise it** in the
@@ -1310,7 +1326,13 @@ class Context:
         #       runtime frames from the tb explicitly?
         # https://docs.python.org/3/reference/simple_stmts.html#the-raise-statement
         # https://stackoverflow.com/a/24752607
-        if from_src_exc is not False:
+        # A transport failure already carries the disconnect cause.
+        # Do not replace it with cancellation used to wake the owner.
+        if (
+            from_src_exc is not False
+            and
+            not isinstance(remote_error, TransportClosed)
+        ):
             raise remote_error from from_src_exc
 
         raise remote_error
@@ -1924,13 +1946,16 @@ class Context:
                         stream._stop_msg = msg
 
                 case Return():
-                    if not self._outcome_msg:
+                    if self._outcome_msg is Unresolved:
                         log.warning(
                             f'Setting final outcome msg AFTER '
-                            f'`._rx_chan.send()`??\n'
+                            f'`._send_chan.send_nowait()`??\n'
                             f'\n'
                             f'{msg}'
                         )
+                        # Record receipt even while the owner is
+                        # idle and has not drained the result yet.
+                        # A subsequent EOF must not replace Return.
                         self._outcome_msg = msg
 
             return True

tractor/runtime/_rpc.py

@@ -46,6 +46,7 @@ from trio import (
 from ..ipc import Channel
 from .._context import (
     Context,
+    Unresolved,
 )
 from .._exceptions import (
     ContextCancelled,
@@ -1368,6 +1369,35 @@ async def process_messages(
             )
 
     except TransportClosed as tc:
+        # MsgpackTransport._iter_pkts() 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:
+        # an unfinished 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;
+        # open_context() subsequently raises TransportClosed while
+        # unwinding. This intentionally remains distinct from a
+        # peer's explicit cancellation request/remote Error msg.
+        #
+        # 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
+        # its owner has not drained, instead of replacing them by EOF.
+        for ctx in tuple(actor._contexts.values()):
+            if (
+                ctx.chan is chan
+                and ctx.side == 'parent'
+                and ctx._scope is not None
+                and ctx._remote_error is None
+                and ctx._outcome_msg is Unresolved
+                and not ctx._final_result_is_set()
+            ):
+                ctx._maybe_cancel_and_set_remote_error(tc)
+
         # channels "breaking" (for TCP streams by EOF or 104
         # connection-reset) is ok since we don't have a teardown
         # handshake for them (yet) and instead we simply bail out of