Reproduce `@context` final-`Return` disconnect cascade

A chart disconnect can leave its context endpoint returning over
a closed channel. Exercise the real `_rpc._invoke()` dispatcher
with loopback TCP and a sibling service to isolate the resulting
`TransportClosed` nursery cancellation.

Deats,
- Order caller close, callee EOF and endpoint completion with
  events instead of timing sleeps or injected transport errors.
- Keep an ordinary-RPC control for existing response hardening.
- Capture the collateral task, actor/PID, nursery, cancel scope
  and cancellation source in a test-generated exception note.
- Record the live IB diagnosis and generated-test provenance.

The ordinary-RPC control passes; the context case remains red
with `ClosedResourceError` wrapped as `TransportClosed`. This
commit establishes the regression before runtime remediation.

Prompt-IO: ai/prompt-io/20261019T140536Z_9a0f220f_final_return_prompt_io.md

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

ai/prompt-io/20261019T140536Z_9a0f220f_final_return_prompt_io.md

@@ -0,0 +1,71 @@
+---
+created_at: 2026-10-19T14:05:36Z
+generated_by:
+  harness: opencode
+  provider: openai
+  model: openai/gpt-6.1-sol
+git_ref: 9a0f220f
+scope: tests
+substantive: true
+raw_file: 20261019T140536Z_9a0f220f_final_return_prompt_io.raw.md
+---
+
+## Prompt
+
+The human reported chart Ctrl-C causing search's final Return delivery
+to raise TransportClosed in datad.ib, cancelling data services and then
+pikerd's trading/sampler services. They asked whether the Piker branch
+ib_connectivity_harden already remediates it and authorized a minimal
+Tractor sibling-survival reproducer if no existing remediation applies:
+
+> ok let's do it if we find no other redmediations per above.
+
+The human explicitly selected capture of this test-only provenance.
+
+## Response summary
+
+Compared ib_connectivity_harden with the committed evaluation baseline:
+efa72980's IB socket-disconnect propagation is already present; the
+missing d126302f/c0dae89a changes concern idle arithmetic and sampler
+cancellation-state access. Tractor ebf2258b hardens ordinary RPC response
+sends, but the context final Return remains exposed. Inspected the
+additional ipc_fault_regressions branch: its EOF/context-status changes
+do not guard this final-result send.
+
+Added a focused real-TCP, same-process dispatcher regression with an
+ordinary RPC control and a context case. Explicit peer-close/EOF events
+force the failure ordering. A sibling request/response checks service
+survival; the failure path confirms premature sibling cancellation.
+
+## Files changed
+
+- tests/test_context_final_return_disconnect.py: reproducer and control.
+- This provenance pair.
+
+## Verification
+
+- Ordinary RPC control passed.
+- Context case failed with the reported TransportClosed caused by
+  ClosedResourceError, confirming collateral nursery cancellation.
+- Final two-case run: 1 failed, 1 passed in 0.07s.
+- Ruff passed; git diff --check passed.
+- Test uses isolated local TCP, no broker, feed, Qt or asyncio bridge.
+- Root actor and channels exit through their context managers.
+- No production fix, staging, commit or ref changes.
+
+## Human edits
+
+None observed in the generated test at capture time.
+
+## Follow-up: identify the collateral sibling
+
+The human requested the task repr and runtime metadata behind the
+test-generated exception note. The cancellation handler now captures
+the sibling task, Actor.aid (including PID), parent nursery, nursery
+cancel scope and original trio.Cancelled repr while still unwinding.
+The test attaches that snapshot to the escaping TransportClosed;
+Tractor production reporting and exception handling are unchanged.
+
+The updated two-case reproducer again yielded 1 passed, 1 failed in
+0.07s, with the sibling identity and source invoke task in the note.
+Ruff and git diff --check passed after the diagnostic change.

ai/prompt-io/20261019T140536Z_9a0f220f_final_return_prompt_io.raw.md

@@ -0,0 +1,66 @@
+---
+created_at: 2026-10-19T14:05:36Z
+generated_by:
+  harness: opencode
+  provider: openai
+  model: openai/gpt-6.1-sol
+git_ref: 9a0f220f
+diff_cmd: git diff --no-index /dev/null tests/test_context_final_return_disconnect.py
+---
+
+## Captured output
+
+`ib_connectivity_harden` has no fix for this path. Its IB socket
+disconnect handling is already in our evaluation; the two missing
+commits fix idle-timer arithmetic and sampler cancellation-state
+access.
+
+I also checked Tractor’s related fault work. The existing
+response-send fix covers ordinary RPCs, but leaves the `@context`
+final-result send exposed. I’m building the reproducer in the
+evaluated Tractor worktree, with isolated local IPC.
+
+The reproducer hits the same failure with real loopback TCP:
+**ordinary RPC passes; `@context` fails**, and the unrelated
+sibling is confirmed cancelled before it can answer a request.
+The two cases finish in 0.07 seconds, without IB, Qt or asyncio.
+
+## Generated test
+
+> `git diff --no-index /dev/null tests/test_context_final_return_disconnect.py`
+
+The file is currently untracked. The pointer above displays the entire
+generated artifact without duplicating it in this record.
+
+finish_after_disconnect() is a real Tractor context endpoint that
+publishes readiness and returns after an event. ordinary_result()
+provides a non-context RPC control. The parameterized test invokes the
+actual dispatcher with a real root actor and real loopback TCP channels.
+It drains startup messages, closes the caller, observes callee EOF,
+closes the callee channel, then releases the endpoint. A nursery sibling
+must answer a subsequent request. On the observed TransportClosed it
+also verifies collateral sibling cancellation and re-raises the fault.
+
+The explicit event ordering isolates final-result delivery, without
+mocked exceptions, sleeps, provider connections or process supervision.
+
+## Final verification
+
+Command environment: evaluation Piker py313/bin/python (Python 3.13.14,
+pytest 9.0.2), importing Tractor from address_logging_levels at 9a0f220f.
+VIRTUAL_ENV was removed; TRACTOR_SPAWN_METHOD and TRACTOR_LOGLEVEL unset.
+LD_LIBRARY_PATH included the existing Nix GCC runtime library.
+
+`python -m pytest tests/test_context_final_return_disconnect.py --tpt-proto=tcp --tb=short`
+
+Result: 1 failed, 1 passed in 0.07s. Diagnostic --tb=short was used
+after the ordinary full-trace run confirmed the dispatcher failure.
+The context case raises TransportClosed caused by ClosedResourceError
+at MsgpackTransport.send(), with the note:
+The unrelated sibling was cancelled before it could answer the post-disconnect request.
+
+`ruff check tests/test_context_final_return_disconnect.py`
+
+All checks passed!
+
+`git diff --check` passed. No production runtime edits were made.

tests/test_context_final_return_disconnect.py

@@ -0,0 +1,227 @@
+'''
+Isolate final-result delivery from sibling service lifetimes.
+
+'''
+from uuid import uuid4
+
+import msgspec
+import pytest
+import tractor
+import trio
+
+from tractor.ipc import Channel
+from tractor.msg import Aid, StartAck, Started
+from tractor.runtime import _rpc
+
+
+@tractor.context
+async def finish_after_disconnect(
+    ctx: tractor.Context,
+    peer_closed: trio.Event,
+    endpoint_finished: trio.Event,
+
+) -> str:
+    '''
+    Finish successfully only after the caller connection is closed.
+
+    '''
+    await ctx.started('ready')
+    await peer_closed.wait()
+    endpoint_finished.set()
+    return 'finished'
+
+
+async def ordinary_result(
+    peer_closed: trio.Event,
+    endpoint_finished: trio.Event,
+
+) -> str:
+    '''
+    Exercise the already-hardened ordinary RPC final-result path.
+
+    '''
+    await peer_closed.wait()
+    endpoint_finished.set()
+    return 'finished'
+
+
+@pytest.mark.parametrize(
+    'context_endpoint',
+    [False, True],
+    ids=['ordinary-rpc', 'context'],
+)
+def test_final_return_disconnect_preserves_sibling(
+    reg_addr: tuple,
+    context_endpoint: bool,
+
+) -> None:
+    '''
+    A departing caller must not cancel an unrelated service task.
+
+    Chart Ctrl-C made open_symbol_search finish before Tractor sent
+    its final Return. TransportClosed from that send escaped _invoke
+    and cancelled the shared service nursery, including IB feeds.
+
+    Use real loopback TCP channels and a real Actor/Context, invoking
+    _invoke directly to control the precise response-send ordering.
+    The caller consumes startup messages, closes its socket, and the
+    callee observes EOF and closes its channel before releasing the
+    successful endpoint. No transport exception is mocked or injected.
+
+    After _invoke completes, a sibling must still answer a request.
+    The ordinary-RPC case is a control for the existing response-send
+    hardening. The context case exposes the uncovered final Return.
+    Capture the sibling's task, actor, nursery and cancellation reprs
+    while unwinding so the failure identifies the collateral victim
+    and the cancellation source even after the runtime has exited.
+    This isolates response delivery; it does not exercise discovery,
+    process supervision, the asyncio bridge, or Qt cancellation.
+
+    '''
+    collateral_cancellations: list[tuple[trio.Cancelled, str]] = []
+
+    async def main() -> None:
+        '''
+        Order peer EOF before endpoint completion without sleeps.
+
+        '''
+        peer_closed = trio.Event()
+        endpoint_finished = trio.Event()
+        invoke_finished = trio.Event()
+        ping = trio.Event()
+        pong = trio.Event()
+        sibling_stopped = trio.Event()
+
+        async def sibling(
+            task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED,
+        ) -> None:
+            '''
+            Answer a post-disconnect request until explicitly stopped.
+
+            '''
+            task_status.started()
+            try:
+                await ping.wait()
+                pong.set()
+                await trio.sleep_forever()
+            except trio.Cancelled as exc:
+                if not pong.is_set():
+                    assert endpoint_finished.is_set()
+                    task: trio.lowlevel.Task = (
+                        trio.lowlevel.current_task()
+                    )
+                    aid: Aid = actor.aid
+                    nursery: trio.Nursery = task.parent_nursery
+                    cancel_scope: trio.CancelScope = (
+                        nursery.cancel_scope
+                    )
+                    runtime_info: str = (
+                        f'\nSibling task: {task!r}\n'
+                        f'Actor identity: {aid!r}\n'
+                        f'Parent nursery: {nursery!r}\n'
+                        f'Parent cancel scope: {cancel_scope!r}\n'
+                        f'Cancellation: {exc!r}'
+                    )
+                    collateral_cancellations.append(
+                        (exc, runtime_info),
+                    )
+                raise
+            finally:
+                sibling_stopped.set()
+
+        with trio.fail_after(10):
+            async with tractor.open_root_actor(
+                registry_addrs=[reg_addr],
+                enable_transports=['tcp'],
+                loglevel='error',
+            ) as actor:
+                listeners = await trio.open_tcp_listeners(
+                    0,
+                    host='127.0.0.1',
+                )
+                [listener] = listeners
+                async with listener:
+                    port: int = listener.socket.getsockname()[1]
+                    caller_stream = await trio.open_tcp_stream(
+                        '127.0.0.1',
+                        port,
+                    )
+                    callee_stream = await listener.accept()
+
+                async with caller_stream, callee_stream:
+                    caller = Channel.from_stream(caller_stream)
+                    callee = Channel.from_stream(callee_stream)
+                    caller._aid = actor.aid
+                    callee._aid = Aid(
+                        name='departing-chart',
+                        uuid=str(uuid4()),
+                    )
+                    cid: str = str(uuid4())
+                    endpoint = (
+                        finish_after_disconnect
+                        if context_endpoint else ordinary_result
+                    )
+                    kwargs = {
+                        'peer_closed': peer_closed,
+                        'endpoint_finished': endpoint_finished,
+                    }
+
+                    async def invoke(
+                        task_status: trio.TaskStatus,
+                    ) -> None:
+                        '''
+                        Run the actual RPC dispatcher to completion.
+
+                        '''
+                        await _rpc._invoke(
+                            actor,
+                            cid,
+                            callee,
+                            endpoint,
+                            kwargs,
+                            task_status=task_status,
+                        )
+                        invoke_finished.set()
+
+                    async with trio.open_nursery() as services:
+                        await services.start(sibling)
+                        ctx = await services.start(invoke)
+                        ack = await caller.recv()
+                        assert isinstance(ack, StartAck)
+                        assert ack.cid == cid
+                        if context_endpoint:
+                            started = await caller.recv()
+                            assert isinstance(started, Started)
+                            assert started.cid == cid
+                            assert msgspec.msgpack.decode(
+                                started.pld,
+                            ) == 'ready'
+
+                        await caller.aclose()
+                        with pytest.raises(tractor.TransportClosed):
+                            await callee.recv()
+                        await callee.aclose()
+                        peer_closed.set()
+
+                        await invoke_finished.wait()
+                        assert endpoint_finished.is_set()
+                        assert not sibling_stopped.is_set()
+                        assert (callee, cid) not in actor._rpc_tasks
+                        ping.set()
+                        await pong.wait()
+                        if context_endpoint:
+                            assert ctx._result == 'finished'
+                        services.cancel_scope.cancel()
+
+    try:
+        trio.run(main)
+    except tractor.TransportClosed as exc:
+        assert collateral_cancellations
+        cancelled, runtime_info = collateral_cancellations[0]
+        assert 'TransportClosed' in str(cancelled)
+        exc.add_note(
+            'The unrelated sibling was cancelled before it could '
+            'answer the post-disconnect request.'
+            + runtime_info
+        )
+        raise