Fix IB `MethodProxy` response routing

`open_client_proxy()` broadcast its one-slot linked channel to an
always-running status relay and an idle method receiver. Status
traffic advanced only the relay, so the first lazy contract lookup
for an uncached order raised `Lagged` before receiving its
response.

Give `relay_client_proxy_messages()` sole receive ownership. Route
tagged responses from concurrent caller tasks into their per-`mid`
slots while each proxy context retains its own channel and relay.

Deats,
- log stale responses with their original method name;
- warn when unresolved-request metadata exceeds its bounded limit;
- drop late responses after caller cancellation;
- preserve the reconnect design note and message-case rationale;
- accept clean channel EOF during proxy teardown;
- cover idle status, concurrency, cancellation and EOF paths.

Prompt-IO: ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md

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

.claude/skills/run-tests/test-harness-reference.md

@@ -108,6 +108,7 @@ Deterministic or local first-pass targets:
 - `tests/test_store_cli.py`
 - `tests/test_backfill_audit_snippet.py`
 - `tests/test_ib_history.py`
+- `tests/test_ib_method_proxy.py`
 - `tests/test_history_backfill.py`
 - `tests/test_ldshm.py`
 - `tests/test_accounting.py::test_account_file_default_empty`
@@ -193,6 +194,7 @@ tests/
   test_ems.py               actor, EMS, and paper-position behavior
   test_feeds.py             live Binance/Kraken feeds and shared memory
   test_ib_history.py        deterministic IB history request formatting
+  test_ib_method_proxy.py   deterministic IB asyncio proxy routing
   test_history_backfill.py  deterministic history/SHM orchestration
   test_ldshm.py             SHM unpublished-slot guard
   test_questrade.py         obsolete credentialed tests; skipped
@@ -213,6 +215,7 @@ tests/
 | `piker/storage/cli.py` command UX | `tests/test_store_cli.py` | fake SHM/runtime, no mutation |
 | `snippets/nativedb_backfill_audit.xsh` | `tests/test_backfill_audit_snippet.py` | disposable paths only |
 | `piker/brokers/ib/api.py`, `feed.py` history | `tests/test_ib_history.py` | fake client, no network |
+| `piker/brokers/ib/api.py` method proxy | `tests/test_ib_method_proxy.py` | fake channel, no network |
 | `piker/tsp/_history.py` | `tests/test_history_backfill.py` | fake provider/storage/SHM |
 | `piker/storage/cli.py` SHM null-slot guard | `tests/test_ldshm.py` | synthetic timestamps, no SHM mutation |
 | `piker/config.py` | `test_account_file_default_empty` | root-network test has a known mismatch |

ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md

@@ -0,0 +1,36 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: ses_0799212ebffe42arY96czXn89F
+timestamp: 2026-08-11T20:58:26Z
+git_ref: 06d5ea5d
+scope: code
+substantive: true
+raw_file: 20260811T205826Z_06d5ea5d_prompt_io.raw.md
+---
+
+## Prompt
+
+Diagnose and fix an IB live-order failure where lazy NVDA contract
+qualification raised `tractor.trionics.Lagged` after the order reached
+`brokerd.ib`. Determine whether recent history changes or cancelled symbol
+searches caused the stale state.
+
+## Response summary
+
+Traced the failure to competing consumers of a one-slot Trio broadcast, not
+the recent SHM change or stale asyncio client state. Replaced broadcast
+fan-out with one response/status demultiplexer, retained request-ID
+correlation, handled cancellation and teardown, documented the task/channel
+ownership model, warned on diagnostic metadata eviction, and added
+deterministic proxy regressions.
+
+## Files changed
+
+- `piker/brokers/ib/api.py` - sole-reader IB proxy response routing.
+- `tests/test_ib_method_proxy.py` - status, concurrency, cancellation and EOF regressions.
+- `.claude/skills/run-tests/test-harness-reference.md` - proxy test mapping.
+
+## Human edits
+
+None - generated output follows the user's live failure diagnosis.

ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.raw.md

@@ -0,0 +1,66 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-11T20:58:26Z
+git_ref: 06d5ea5d
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+The user reported that certain IB order entries might have broken and
+provided a live failure for a first NVDA sell order. EMS delivered the
+request to `brokerd.ib`, but lazy `get_mkt_info('nvda.nasdaq')` contract
+qualification failed in `MethodProxy._run_method()` with:
+
+```text
+tractor.trionics._broadcast.Lagged:
+Task `piker.brokers.ib.broker.handle_order_requests` overrun and
+dropped `0` values
+```
+
+The user suspected a cancelled symbol-search task might have left stale
+asyncio state. Cancellation can leave an orphaned method response, but the
+observed failure was stale Trio broadcast state. `open_client_proxy()`
+broadcast a one-slot `LinkedTaskChannel` to an always-running IB status relay
+and an idle method receiver. Status traffic advanced only the relay, so the
+first method call for an uncached symbol entered an already-lagged receiver.
+Orders for eagerly cached contracts bypassed this path.
+
+The immediately preceding sparse-history commit did not modify IB order,
+symbol, or proxy code; it exposed no direct dependency on this failure.
+
+> `git diff HEAD~1..HEAD -- piker/brokers/ib/api.py`
+
+Generated a sole-reader message demultiplexer for each IB client proxy.
+`relay_client_proxy_messages()` owns every asyncio-to-Trio receive, handles
+status/error traffic, and dispatches `mid`-tagged responses into per-call
+`MethodProxy` wait slots. Cancelled calls remove their pending slot, causing
+late responses to be logged and dropped without skewing later calls. Clean
+channel EOF is accepted during normal proxy teardown.
+
+Follow-up review restored the masked reconnect TODO and per-message case
+rationale, documented the one-reader/many-writer task hierarchy and separate
+per-proxy channels, and added a warning with request method metadata when the
+bounded unresolved-request diagnostic table evicts its oldest entry.
+
+> `git diff HEAD~1..HEAD -- tests/test_ib_method_proxy.py`
+
+Generated deterministic regressions for idle status traffic before first
+qualification, two concurrent calls with reverse-order responses, cancelled
+calls with late responses followed by a successful call, and graceful
+channel EOF. Every synchronization wait is timeout-bounded.
+
+> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
+
+Registered the deterministic IB method-proxy target and source mapping.
+
+Verification output:
+
+```text
+........................                                                 [100%]
+25 passed, 4 warnings in 3.44s
+```
+
+The warnings are existing Tractor/Trio deprecations from the EMS test.
+Adversarial review drove a clean-EOF guard and bounded polling. Final review
+found no actionable issues. Live IB order validation remains pending and
+requires restarting `brokerd.ib`.

piker/brokers/ib/api.py

@@ -1643,6 +1643,61 @@ class MethodProxy:
         # to the next caller causing an off-by-one skew of
         # every result thereafter!
         self._mids = itertools.count()
+        self._pending: dict[
+            int,
+            tuple[trio.Event, list[dict]],
+        ] = {}
+        self._request_methods: dict[int, str|None] = {}
+
+    def _deliver_method_response(
+        self,
+        msg: dict,
+
+    ) -> bool:
+        '''
+        Deliver one correlated response to its waiting caller.
+
+        '''
+        mid: int = msg['mid']
+        meth: str|None = self._request_methods.pop(mid, None)
+        pending = self._pending.get(mid)
+        if pending is None:
+            log.warning(
+                f'Dropping stale method-resp,\n'
+                f'mid: {mid}\n'
+                f'meth: {meth!r}\n'
+                f'(caller prolly got cancelled '
+                f'before its resp?)\n'
+            )
+            return False
+
+        event, slot = pending
+        slot.append(msg)
+        event.set()
+        return True
+
+    def _track_request_method(
+        self,
+        mid: int,
+        meth: str|None,
+
+    ) -> None:
+        '''
+        Retain bounded request metadata for stale-response logs.
+
+        '''
+        self._request_methods[mid] = meth
+        if len(self._request_methods) > 256:
+            oldest_mid: int = next(iter(self._request_methods))
+            oldest_meth: str|None = self._request_methods.pop(
+                oldest_mid
+            )
+            log.warning(
+                f'Evicting unresolved method-request metadata,\n'
+                f'mid: {oldest_mid}\n'
+                f'meth: {oldest_meth!r}\n'
+                f'pending metadata limit: 256\n'
+            )
 
     async def _run_method(
         self,
@@ -1656,65 +1711,30 @@ class MethodProxy:
         ``tractor.to_asyncio`` layer.
 
         '''
-        chan = self.chan
         mid: int = next(self._mids)
-        await chan.send((meth, kwargs, mid))
+        event = trio.Event()
+        slot: list[dict] = []
+        self._pending[mid] = event, slot
+        self._track_request_method(mid, meth)
+        try:
+            await self.chan.send((meth, kwargs, mid))
+            await event.wait()
+        finally:
+            self._pending.pop(mid, None)
 
-        while not chan.closed():
+        msg: dict = slot[0]
-            msg = await chan.receive()
+        match msg:
+            # OUR method-call response B)
+            case {'result': res}:
+                return res
 
-            # TODO: implement reconnect functionality like
+            case {'exception': err}:
-            # in our `.data._web_bs.NoBsWs`
+                raise err
-            # try:
-            #     msg = await chan.receive()
-            # except ConnectionError:
-            #     self.reset()
 
-            match msg:
+            case _:
-                # OUR method-call response B)
+                raise RuntimeError(
-                case {'mid': resp_mid, 'result': res} if (
+                    f'Invalid method response for mid={mid}: {msg!r}'
-                    resp_mid == mid
+                )
-                ):
-                    return res
-
-                case {'mid': resp_mid, 'exception': err} if (
-                    resp_mid == mid
-                ):
-                    raise err
-
-                # an "orphaned" response to some prior
-                # (cancelled) caller's request; drop it and
-                # keep waiting for ours.
-                case {'mid': resp_mid}:
-                    log.warning(
-                        f'Dropping stale method-resp,\n'
-                        f'mid: {resp_mid} (ours: {mid})\n'
-                        f'(a prior caller prolly got '
-                        f'cancelled before its resp?)\n'
-                    )
-                    continue
-
-                # out-of-band (inline) client error: raise
-                # to the current caller as before.
-                case {'exception': err}:
-                    raise err
-
-                case ('error', emsg):
-                    log.warning(f'IB error relay: {emsg}')
-                    continue
-
-                # routine (api-farm conn) status events
-                # relayed inline by `Client.inline_errors()`;
-                # not a response to our method call so just
-                # log at info and keep waiting.
-                case ('event', emsg):
-                    log.info(
-                        f'IB status event relay: {emsg}'
-                    )
-                    continue
-
-                case _:
-                    log.warning(f'UNKNOWN IB MSG: {msg}')
 
     def status_event(
         self,
@@ -1800,6 +1820,73 @@ async def open_aio_client_method_relay(
                 raise ValueError(f'Unhandled msg {msg}')
 
 
+async def relay_client_proxy_messages(
+    chan: tractor.to_asyncio.LinkedTaskChannel,
+    proxy: MethodProxy,
+
+) -> None:
+    '''
+    Route one proxy's asyncio-client msgs from a single reader.
+
+    Each `open_client_proxy()` allocates its own linked channel and
+    starts this task in its relay nursery. Any number of Trio caller
+    tasks may concurrently write tagged requests through
+    `MethodProxy._run_method()` and wait on their per-`mid` slots.
+    This task alone reads the channel and dispatches each response to
+    its waiting caller.
+
+    Separate proxy contexts own separate channels and reader tasks;
+    no cross-proxy response broadcast is intended. IB status events
+    still reach each proxy through its own `Client.inline_errors()`
+    handler.
+
+    '''
+    try:
+        while not chan.closed():
+            # TODO: implement reconnect functionality like
+            # in our `.data._web_bs.NoBsWs`
+            # try:
+            msg = await chan.receive()
+            # except ConnectionError:
+            #     proxy.reset()
+
+            match msg:
+                # Correlated response from a method call.
+                case {'mid': _}:
+                    proxy._deliver_method_response(msg)
+
+                # routine (api-farm conn) status events
+                # relayed inline by `Client.inline_errors()`;
+                # not a response to a method call so just
+                # log at info and keep waiting.
+                case ('event', status_msg):
+                    reason = status_msg['reason']
+                    event = proxy.event_table.pop(reason, None)
+                    if (
+                        event
+                        and
+                        event.statistics().tasks_waiting
+                    ):
+                        log.info(f'Relaying ib status message: {msg}')
+                        event.set()
+
+                # Inline API error with no request correlation.
+                case ('error', emsg):
+                    log.warning(f'IB error relay: {emsg}')
+
+                # Out-of-band client error with no method `mid`.
+                case {'exception': err}:
+                    log.error(
+                        f'Uncorrelated IB client error: {err!r}'
+                    )
+
+                # Preserve visibility into unexpected relay traffic.
+                case _:
+                    log.warning(f'UNKNOWN IB MSG: {msg}')
+    except trio.EndOfChannel:
+        log.info('IB client proxy relay closed')
+
+
 @acm
 async def open_client_proxy(
     client: Client,
@@ -1836,28 +1923,11 @@ async def open_client_proxy(
                 continue
             setattr(proxy, name, partial(proxy._run_method, meth=name))
 
-        async def relay_events():
+        relay_tn.start_soon(
-
+            relay_client_proxy_messages,
-            async with chan.subscribe() as msg_stream:
+            chan,
-
+            proxy,
-                async for msg in msg_stream:
+        )
-                    if 'event' not in msg:
-                        continue
-
-                    # if 'event' in msg:
-                    # wake up any system event waiters.
-                    etype, status_msg = msg
-                    reason = status_msg['reason']
-
-                    ev = proxy.event_table.pop(reason, None)
-
-                    if ev and ev.statistics().tasks_waiting:
-                        log.info(f'Relaying ib status message: {msg}')
-                        ev.set()
-
-                    continue
-
-        relay_tn.start_soon(relay_events)
 
         yield proxy
 

tests/test_ib_method_proxy.py

@@ -0,0 +1,271 @@
+'''
+IB asyncio method-proxy regressions.
+
+'''
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+import trio
+
+from piker.brokers.ib import api as ib_api
+from piker.brokers.ib.api import (
+    MethodProxy,
+    relay_client_proxy_messages,
+)
+
+
+class FakeChannel:
+    '''
+    Model the Trio side of an asyncio linked-task channel.
+
+    '''
+    def __init__(self) -> None:
+        self.requests: list[tuple] = []
+        (
+            self._tx,
+            self._rx,
+        ) = trio.open_memory_channel[Any](10)
+
+    async def send(self, msg: tuple) -> None:
+        '''
+        Capture one Trio-to-asyncio method request.
+
+        '''
+        self.requests.append(msg)
+
+    async def receive(self) -> Any:
+        '''
+        Receive one simulated asyncio-to-Trio response.
+
+        '''
+        return await self._rx.receive()
+
+    def closed(self) -> bool:
+        '''
+        Report an open channel for the lifetime of each test.
+
+        '''
+        return False
+
+    async def respond(self, msg: Any) -> None:
+        '''
+        Send one response or status event to the proxy relay.
+
+        '''
+        await self._tx.send(msg)
+
+    async def close_responses(self) -> None:
+        '''
+        Close the simulated asyncio-to-Trio response channel.
+
+        '''
+        await self._tx.aclose()
+
+
+def test_method_proxy_routes_after_idle_status_events() -> None:
+    '''
+    Idle IB status traffic must not break first-symbol qualification.
+
+    ``open_client_proxy()`` previously broadcast a one-slot channel
+    to an always-running status relay and an idle method receiver.
+    Status events advanced only the relay, so a first order for an
+    uncached symbol raised ``Lagged`` before receiving
+    ``get_sym_details()``. Feed several idle events through the
+    sole-reader relay, then start two method calls and return their
+    responses in reverse order. Both exact results prove status
+    traffic cannot overrun a dormant caller and concurrent responses
+    remain correlated by request ID.
+
+    '''
+    async def main() -> None:
+        chan = FakeChannel()
+        proxy = MethodProxy(
+            chan,
+            event_table={},
+            asyncio_ns=SimpleNamespace(),
+        )
+        results: dict[str, str] = {}
+
+        async def call(name: str) -> None:
+            results[name] = await proxy._run_method(meth=name)
+
+        async with trio.open_nursery() as nursery:
+            nursery.start_soon(
+                relay_client_proxy_messages,
+                chan,
+                proxy,
+            )
+            for i in range(3):
+                await chan.respond((
+                    'event',
+                    {'reason': f'farm-status-{i}'},
+                ))
+
+            nursery.start_soon(call, 'first')
+            nursery.start_soon(call, 'second')
+            with trio.fail_after(0.1):
+                while len(chan.requests) < 2:
+                    await trio.lowlevel.checkpoint()
+
+            mids: dict[str, int] = {
+                meth: mid
+                for meth, _kwargs, mid in chan.requests
+            }
+            await chan.respond({
+                'mid': mids['second'],
+                'result': 'second-result',
+            })
+            await chan.respond({
+                'mid': mids['first'],
+                'result': 'first-result',
+            })
+            with trio.fail_after(0.1):
+                while len(results) < 2:
+                    await trio.lowlevel.checkpoint()
+            nursery.cancel_scope.cancel()
+
+        assert results == {
+            'first': 'first-result',
+            'second': 'second-result',
+        }
+
+    trio.run(main)
+
+
+def test_method_proxy_relay_accepts_clean_eof() -> None:
+    '''
+    Normal asyncio relay completion must not fail proxy teardown.
+
+    The sole-reader relay blocks in ``LinkedTaskChannel.receive()``
+    until the asyncio task exits and closes its Trio send channel.
+    Close the fake response sender before entering the relay and
+    prove the resulting ``EndOfChannel`` is graceful completion.
+
+    '''
+    async def main() -> None:
+        chan = FakeChannel()
+        proxy = MethodProxy(
+            chan,
+            event_table={},
+            asyncio_ns=SimpleNamespace(),
+        )
+        await chan.close_responses()
+        with trio.fail_after(0.1):
+            await relay_client_proxy_messages(chan, proxy)
+
+    trio.run(main)
+
+
+def test_method_proxy_drops_cancelled_call_response() -> None:
+    '''
+    A cancelled symbol search must not skew the next method response.
+
+    Search timeouts can cancel a Trio caller after its request
+    reaches the asyncio client. Wait until one request is sent,
+    cancel its caller, and then deliver the orphaned response. The
+    sole-reader relay must drop that stale ``mid`` without failing; a
+    subsequent call must still receive its own exact response.
+
+    '''
+    async def main() -> None:
+        chan = FakeChannel()
+        proxy = MethodProxy(
+            chan,
+            event_table={},
+            asyncio_ns=SimpleNamespace(),
+        )
+        call_scope = trio.CancelScope()
+        cancelled = trio.Event()
+        result: list[str] = []
+
+        async def cancelled_call() -> None:
+            with call_scope:
+                try:
+                    await proxy._run_method(meth='cancelled')
+                finally:
+                    cancelled.set()
+
+        async def next_call() -> None:
+            value = await proxy._run_method(meth='next')
+            result.append(value)
+
+        async with trio.open_nursery() as nursery:
+            nursery.start_soon(
+                relay_client_proxy_messages,
+                chan,
+                proxy,
+            )
+            nursery.start_soon(cancelled_call)
+            with trio.fail_after(0.1):
+                while len(chan.requests) < 1:
+                    await trio.lowlevel.checkpoint()
+            cancelled_mid: int = chan.requests[0][2]
+            call_scope.cancel()
+            with trio.fail_after(0.1):
+                await cancelled.wait()
+            assert proxy._pending == {}
+            assert (
+                proxy._request_methods[cancelled_mid]
+                ==
+                'cancelled'
+            )
+
+            await chan.respond({
+                'mid': cancelled_mid,
+                'result': 'stale',
+            })
+            with trio.fail_after(0.1):
+                while cancelled_mid in proxy._request_methods:
+                    await trio.lowlevel.checkpoint()
+            nursery.start_soon(next_call)
+            with trio.fail_after(0.1):
+                while len(chan.requests) < 2:
+                    await trio.lowlevel.checkpoint()
+            next_mid: int = chan.requests[1][2]
+            await chan.respond({
+                'mid': next_mid,
+                'result': 'fresh',
+            })
+            with trio.fail_after(0.1):
+                while not result:
+                    await trio.lowlevel.checkpoint()
+            nursery.cancel_scope.cancel()
+
+        assert result == ['fresh']
+
+    trio.run(main)
+
+
+def test_method_proxy_warns_when_metadata_is_evicted(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Unresolved metadata eviction must remain operator-visible.
+
+    Cancelled calls whose asyncio responses never arrive retain a
+    bounded method name solely for later stale-response diagnostics.
+    Fill that table beyond its limit and prove the oldest entry is
+    evicted with its request ID, method name, and configured bound in
+    the warning.
+
+    '''
+    proxy = MethodProxy(
+        FakeChannel(),
+        event_table={},
+        asyncio_ns=SimpleNamespace(),
+    )
+    warnings: list[str] = []
+    monkeypatch.setattr(ib_api.log, 'warning', warnings.append)
+
+    for mid in range(257):
+        proxy._track_request_method(mid, f'method-{mid}')
+
+    assert len(proxy._request_methods) == 256
+    assert 0 not in proxy._request_methods
+    assert warnings == [
+        'Evicting unresolved method-request metadata,\n'
+        'mid: 0\n'
+        "meth: 'method-0'\n"
+        'pending metadata limit: 256\n'
+    ]