.ib.api: req-id correlate `MethodProxy` calls

The (single, shared) relay chan has NO ordering guarantee
when a caller task is cancelled (eg. by a search-req
timeout) after sending its request but before consuming the
response: the "orphaned" response gets mis-delivered to the
next caller, off-by-one skewing every result thereafter!

Deats,
- tag each request with a `mid` from an `itertools.count()`
  on the proxy and echo it back in both the result and
  exception resps from `open_aio_client_method_relay()`.
- drop (w/ a warning) any stale resp whose `mid` doesn't
  match the current caller's and keep waiting for ours.
- rewrite the resp-wait loop as a `match:` (resolving the
  old "py3.10 syntax" TODO) incl. the prior inline
  `('error', ...)`/`('event', ...)` relay cases.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

ai/prompt-io/claude/20260610T213549Z_f084e899_prompt_io.md

@@ -0,0 +1,60 @@
+---
+model: claude-fable-5[1m]
+service: claude
+session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
+timestamp: 2026-06-10T21:35:49Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T213549Z_f084e899_prompt_io.md
+scope: code
+substantive: true
+raw_file: 20260610T213549Z_f084e899_prompt_io.raw.md
+---
+
+## Prompt
+
+User bug report from live `ib` testing of the
+(datad|brokerd)-split branch:
+
+> ok doing some testing and noticing live orders do not
+> work otb, pikerd show this on live submission,
+> [pdb traceback: `RuntimeError("Can not order
+> {symbol}, no live feed?")` at `ib/api.py:1152`
+> `submit_limit()` with `self._contracts == {}` in the
+> `brokerd.ib` actor + a `TrioTaskExited` teardown
+> cascade]
+
+Follow-up user direction fixing the design intent:
+
+> i think the main thing here is that [brokerd] should
+> be able to always submit orders without a live feed
+> being up and registered in the same subactor right?
+
+## Response summary
+
+Root cause: pre-split the feed code populated the
+actor-local `Client._contracts` cache (via
+`get_mkt_info()` writing `mkt.bs_fqme` keys) in the
+SAME process as order submission; post-split the
+trading actor's client is never warmed. Fix: lazily
+qualify + cache the contract per order request in
+`handle_order_requests()` by running the same
+`get_mkt_info(fqme, proxy=...)` ep the feed side uses,
+plus per-order error relay (`BrokerdError`) so one bad
+submission can't crash the whole trades dialog (the
+`TrioTaskExited` storm was teardown cascade from the
+original raise).
+
+## Files changed
+
+- `piker/brokers/ib/broker.py` — thread `proxies` into
+  `handle_order_requests()`; lazy contract qualify on
+  cache-miss; guard `submit_limit()` w/ `BrokerdError`
+  relay; uncomment the (anticipatory) `get_mkt_info`
+  import
+- `piker/brokers/ib/api.py` — fix the non-f-string
+  raise msg + document the new qualification contract
+
+## Human edits
+
+None — committed as generated. Live `ib` order retest
+performed by the human post-commit.

ai/prompt-io/claude/20260610T213549Z_f084e899_prompt_io.raw.md

@@ -0,0 +1,57 @@
+---
+model: claude-fable-5[1m]
+service: claude
+timestamp: 2026-06-10T21:35:49Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T213549Z_f084e899_prompt_io.md
+---
+
+NOTE: diff-ref mode entry (code committed in the same
+commit as this log); recorded from the live debug
+session per the `/prompt-io` skill rules.
+
+> `git log -1 -p --follow -- piker/brokers/ib/broker.py`
+> `git log -1 -p --follow -- piker/brokers/ib/api.py`
+
+Key diagnostic chain (from session):
+
+- pdb showed `Client._contracts == {}` inside
+  `brokerd.ib`'s `submit_limit()`; the cache has
+  exactly TWO write sites: `Client.find_contracts()`
+  (api.py, keys `f'{sym}.{exch}.ib'`) and
+  `symbols.get_mkt_info()` (key `mkt.bs_fqme`, eg.
+  'nvda.nasdaq' — NO `.ib` suffix).
+- `BrokerdOrder.symbol` arrives as the bs_fqme form
+  ('nvda.nasdaq') so ONLY the `get_mkt_info()` write
+  site produces the key `submit_limit()` reads —
+  ie. pre-split it was the feed's in-proc
+  `get_mkt_info(sym, proxy=proxy)` call keeping orders
+  working, NOT `find_contracts()`.
+- the existing TODO at `symbols.py:642-644` literally
+  predicted this: "this is going to be problematic
+  if/when we split out the datad vs. brokerd actors
+  since the mktmap lookup table will now be
+  inaccessible.."
+- instance identity verified: `proxy._aio_ns` IS the
+  same `Client` obj as `_accounts2clients[account]`
+  (both sourced from the `load_aio_clients()` cache via
+  `open_client_proxies()`), so a brokerd-side
+  `get_mkt_info(fqme, proxy=proxies[account])` warms
+  exactly the dict `submit_limit()` reads. It also
+  populates `client._cons2mkts` which the
+  position-audit path (`broker.py` backup-table code)
+  needs in this actor anyway.
+- the `TrioTaskExited` storm in the user's log
+  (`recv_trade_updates`, `open_aio_client_method_relay`
+  aio tasks) is teardown cascade: the raise crashed
+  `handle_order_requests` -> nursery teardown ripped
+  the trio sides of still-running aio relay tasks.
+  Hence the added per-order try/except ->
+  `BrokerdError` relay hardening so a single bad
+  submission degrades to an EMS error msg instead of
+  killing the backend's entire order-ctl dialog.
+
+Verification: `tests/test_services.py` (5 passed) +
+`tests/test_ems.py` (6 passed) regression-green; live
+`ib` submission retest delegated to the human (needs a
+running TWS/gw).

piker/brokers/ib/api.py

@@ -967,8 +967,26 @@ class Client:
 
                 else:
                     raise
+            # XXX: ambiguous/unqualifiable contracts are
+            # returned as `None` entries by
+            # `qualifyContractsAsync()` (which also logs an
+            # "Ambiguous contract" warning listing the
+            # possible matches) so filter them BEFORE any
+            # attr access B)
+            contracts: list[Contract] = [
+                tract
+                for tract in contracts
+                if tract is not None
+            ]
             if not contracts:
-                raise ValueError(f"No contract could be found {con}")
+                raise ValueError(
+                    f'No (unambiguous) contract could be '
+                    f'qualified for {con!r}\n'
+                    f'\n'
+                    f'If a stonk, you likely need a (more) '
+                    f'venue-qualified fqme,\n'
+                    f"eg. 'gld.arca.ib' instead of 'gld.ib'\n"
+                )
 
         # pack all contracts into cache
         for tract in contracts:
@@ -1146,10 +1164,16 @@ class Client:
         try:
             con: Contract = self._contracts[symbol]
         except KeyError:
-            # require that the symbol has been previously cached by
+            # require that the mkt's contract has been previously
-            # a data feed request - ensure we aren't making orders
+            # qualified and cached (see `.symbols.get_mkt_info()`
-            # against non-known prices.
+            # which is run for any feed-init OR lazily by the
-            raise RuntimeError("Can not order {symbol}, no live feed?")
+            # order-request handler in `.broker`) - ensures we
+            # aren't making orders against unknown contracts.
+            raise RuntimeError(
+                f'Can not order {symbol!r}, '
+                f'no qualified contract cached!?\n'
+                f'_contracts: {list(self._contracts)!r}\n'
+            )
 
         try:
             trade = self.ib.placeOrder(
@@ -1608,6 +1632,17 @@ class MethodProxy:
         self.event_table = event_table
         self._aio_ns = asyncio_ns
 
+        # request-id counter for correlating each method
+        # call to its (eventual) response; necessary since
+        # the (single, shared) chan has NO other ordering
+        # guarantee whenever a caller task is CANCELLED
+        # (eg. by a search-req timeout) after sending its
+        # request but before consuming the response: the
+        # "orphaned" response otherwise gets mis-delivered
+        # to the next caller causing an off-by-one skew of
+        # every result thereafter!
+        self._mids = itertools.count()
+
     async def _run_method(
         self,
         *,
@@ -1621,10 +1656,10 @@ class MethodProxy:
 
         '''
         chan = self.chan
-        await chan.send((meth, kwargs))
+        mid: int = next(self._mids)
+        await chan.send((meth, kwargs, mid))
 
         while not chan.closed():
-            # send through method + ``kwargs: dict`` as pair
             msg = await chan.receive()
 
             # TODO: implement reconnect functionality like
@@ -1634,23 +1669,50 @@ class MethodProxy:
             # except ConnectionError:
             #     self.reset()
 
-            # print(f'NEXT MSG: {msg}')
+            match msg:
-
+                # OUR method-call response B)
-            # TODO: py3.10 ``match:`` syntax B)
+                case {'mid': resp_mid, 'result': res} if (
-            if 'result' in msg:
+                    resp_mid == mid
-                res = msg.get('result')
+                ):
                     return res
 
-            elif 'exception' in msg:
+                case {'mid': resp_mid, 'exception': err} if (
-                err = msg.get('exception')
+                    resp_mid == mid
+                ):
                     raise err
 
-            elif 'error' in msg:
+                # an "orphaned" response to some prior
-                etype, emsg = msg
+                # (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
 
-            else:
+                # 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(
@@ -1704,22 +1766,31 @@ async def open_aio_client_method_relay(
                 log.info('asyncio `Client` method-proxy SHUTDOWN!')
                 break
 
-            case (meth_name, kwargs):
+            case (meth_name, kwargs, mid):
-                meth_name, kwargs = msg
                 meth = getattr(client, meth_name)
 
                 try:
                     resp = await meth(**kwargs)
-                    # echo the msg back
+                    # echo the msg back tagged with the
-                    chan.send_nowait({'result': resp})
+                    # req-id for caller-side correlation.
+                    chan.send_nowait({
+                        'mid': mid,
+                        'result': resp,
+                    })
 
-                except (
+                # XXX: relay ALL (non-cancel) errors to the
-                    RequestError,
+                # `trio`-side caller (which re-raises in the
-
+                # `MethodProxy._run_method()` frame) instead
-                    # TODO: relay all errors to trio?
+                # of crashing this relay task and thus the
-                    # BaseException,
+                # whole proxy/dialog; critical post the
-                ) as err:
+                # (datad|brokerd)-split where eg. contract
-                    chan.send_nowait({'exception': err})
+                # qualification failures are expected to be
+                # caught per-request by order/warmup code!
+                except Exception as err:
+                    chan.send_nowait({
+                        'mid': mid,
+                        'exception': err,
+                    })
 
             case {'error': content}:
                 chan.send_nowait({'exception': content})

piker/brokers/ib/broker.py

@@ -89,8 +89,9 @@ from .api import (
     MethodProxy,
 )
 from .symbols import (
+    cache_contract,
     con2fqme,
-    # get_mkt_info,
+    get_mkt_info,
 )
 from .ledger import (
     norm_trade_records,
@@ -138,6 +139,7 @@ async def handle_order_requests(
     ems_order_stream: tractor.MsgStream,
     accounts_def: dict[str, str],
     flows: OrderDialogs,
+    proxies: dict[str, MethodProxy],
 
 ) -> None:
 
@@ -180,6 +182,58 @@ async def handle_order_requests(
         if action in {'buy', 'sell'}:
             # validate
             order = BrokerdOrder(**request_msg)
+            fqme: str = order.symbol
+
+            # XXX: lazily qualify and cache the contract for
+            # this mkt since, post the (datad|brokerd)-split,
+            # this trading-only actor will NOT have had its
+            # (api) `Client._contracts` pre-warmed by any
+            # in-proc feed setup (which now runs in the
+            # `datad.ib` sibling); we run the SAME symbology
+            # resolution ep the feed-side uses so the cache
+            # key (`MktPair.bs_fqme`) matches what
+            # `Client.submit_limit()` looks up.
+            # NOTE: normally a no-op since
+            # `open_trade_dialog()` eagerly pre-qualifies all
+            # already-open pp/order mkts at startup; this
+            # only fires for a brand-new (to this daemon)
+            # mkt's first order.
+            if fqme not in client._contracts:
+                proxy: MethodProxy = proxies[account]
+                try:
+                    (
+                        mkt,
+                        details,
+                    ) = await get_mkt_info(
+                        fqme,
+                        proxy=proxy,
+                    )
+                    # XXX: explicit write since the lifo-cache
+                    # may deliver a hit (body skipped) when
+                    # another acct/client already qualified
+                    # this fqme.
+                    cache_contract(
+                        client,
+                        mkt,
+                        details.contract,
+                    )
+                except Exception as err:
+                    log.exception(
+                        f'Failed to qualify contract for\n'
+                        f'fqme: {fqme!r}\n'
+                    )
+                    await ems_order_stream.send(
+                        BrokerdError(
+                            oid=oid,
+                            symbol=fqme,
+                            reason=(
+                                f'No contract could be '
+                                f'qualified for {fqme!r}:\n'
+                                f'{err!r}'
+                            ),
+                        )
+                    )
+                    continue
 
             # XXX: by default 0 tells ``ib_async`` methods that
             # there is no existing order so ask the client to create
@@ -191,15 +245,34 @@ async def handle_order_requests(
                 reqid = int(reqid)
 
             # call our client api to submit the order
+            # NOTE: guard with order-error relay (vs. crashing
+            # this dialog and thus ALL order ctl for the
+            # backend) so one bad submission can't take down
+            # the daemon's clearing loop.
+            try:
                 reqid = client.submit_limit(
                     oid=order.oid,
-                symbol=order.symbol,
+                    symbol=fqme,
                     price=order.price,
                     action=order.action,
                     size=order.size,
                     account=acct_number,
                     reqid=reqid,
                 )
+            except Exception as err:
+                log.exception(
+                    f'Order submission failed for\n'
+                    f'fqme: {fqme!r}\n'
+                )
+                await ems_order_stream.send(
+                    BrokerdError(
+                        oid=oid,
+                        symbol=fqme,
+                        reason=f'Submission error: {err!r}',
+                    )
+                )
+                continue
+
             str_reqid: str = str(reqid)
             if reqid is None:
                 err_msg = BrokerdError(
@@ -548,10 +621,14 @@ async def open_trade_dialog(
 
 ) -> AsyncIterator[dict[str, Any]]:
 
-    get_console_log(
+    # XXX: ONLY adjust the level of this (sub)mod's logger;
-        level=loglevel,
+    # attaching a (stderr) handler (via `get_console_log()`)
-        name=__name__,
+    # here would DOUBLE-print every record since the daemon
-    )
+    # fixture (`.._daemon._setup_persistent_brokerd()`)
+    # already enables the console handler on the parent
+    # subsys logger which all our records propagate to!
+    if loglevel:
+        log.setLevel(loglevel.upper())
 
     # task local msg dialog tracking
     flows = OrderDialogs()
@@ -788,6 +865,64 @@ async def open_trade_dialog(
                 for msg in order_msgs:
                     await ems_stream.send(msg)
 
+                # XXX: eagerly pre-qualify (and cache) the
+                # contracts for all already-open pps and
+                # orders so that (live) order submission
+                # NEVER pays a first-request qualification
+                # delay; any brand-new mkt is still lazily
+                # qualified by `handle_order_requests()` on
+                # its first submission. NOTE: since this
+                # runs BEFORE the order handler task is
+                # even started, no order can clear until
+                # the warmup completes (early reqs just
+                # buffer in the ems IPC stream) B)
+                warmup_fqmes: set[str] = {
+                    msg.symbol
+                    for msg in all_positions
+                }
+                warmup_fqmes.update(
+                    msg.req.symbol
+                    for msg in order_msgs
+                )
+                unique_clients: set[Client] = set(
+                    aioclients.values()
+                )
+                if (
+                    warmup_fqmes
+                    and
+                    proxies
+                ):
+                    a_proxy: MethodProxy = next(
+                        iter(proxies.values())
+                    )
+                    for fqme in warmup_fqmes:
+                        try:
+                            (
+                                mkt,
+                                details,
+                            ) = await get_mkt_info(
+                                fqme,
+                                proxy=a_proxy,
+                            )
+                        except Exception as err:
+                            # XXX: non-fatal; an
+                            # un-warmed mkt just falls
+                            # back to the lazy qualify
+                            # in the order handler.
+                            log.warning(
+                                f'Failed to pre-qualify\n'
+                                f'fqme: {fqme!r}\n'
+                                f'err: {err!r}\n'
+                            )
+                            continue
+
+                        for _client in unique_clients:
+                            cache_contract(
+                                _client,
+                                mkt,
+                                details.contract,
+                            )
+
                 for client in set(aioclients.values()):
                     trade_event_stream: LinkedTaskChannel = await tn.start(
                         open_trade_event_stream,
@@ -801,6 +936,7 @@ async def open_trade_dialog(
                         ems_stream,
                         accounts_def,
                         flows,
+                        proxies,
                     )
 
                     # allocate event relay tasks for each client connection

piker/brokers/ib/symbols.py

@@ -639,12 +639,40 @@ async def get_mkt_info(
 
     # if possible register the bs_mktid to the just-built
     # mkt so that it can be retreived by order mode tasks later.
-    # TODO NOTE: this is going to be problematic if/when we split
-    # out the datatd vs. brokerd actors since the mktmap lookup
-    # table will now be inaccessible..
     if proxy is not None:
-        client: Client = proxy._aio_ns
+        cache_contract(
-        client._contracts[mkt.bs_fqme] = con
+            proxy._aio_ns,
-        client._cons2mkts[con] = mkt
+            mkt,
+            con,
+        )
 
     return mkt, details
+
+
+def cache_contract(
+    client: Client,
+    mkt: MktPair,
+    con: ibis.Contract,
+
+) -> None:
+    '''
+    Register a (qualified) contract + mkt-info pair on the
+    given (api) `Client`'s actor-local lookup tables.
+
+    Cached under BOTH fqme key-forms since consumers vary:
+    - `mkt.bs_fqme` (eg. 'nvda.nasdaq'): read by
+      `Client.submit_limit()` for order requests,
+    - `mkt.fqme` (eg. 'nvda.nasdaq.ib'): read by the
+      fill-time pp-update (symcache-backup-table) path in
+      `.broker.emit_pp_update()`.
+
+    NOTE: post the (datad|brokerd)-actor-split this MUST be
+    run (in the trading actor) for every mkt either eagerly
+    at `.broker.open_trade_dialog()` startup or lazily per
+    order request; there is no in-proc feed setup doing it
+    implicitly anymore!
+
+    '''
+    client._contracts[mkt.bs_fqme] = con
+    client._contracts[mkt.fqme] = con
+    client._cons2mkts[con] = mkt

piker/data/_daemon.py

@@ -93,6 +93,18 @@ async def _setup_persistent_datad(
     )
     assert log.name == _util.subsys
 
+    # XXX: ALSO enable console logging for the provider
+    # backend's mod subtree (eg. `piker.brokers.ib.*`)
+    # since (pre the datad|brokerd-split) that was done
+    # by `brokerd`'s fixture; without this all backend
+    # records here fall through to the stdlib's bare
+    # (non-colorized) `logging.lastResort` handler!
+    get_console_log(
+        level=loglevel or tll,
+        name=f'piker.brokers.{brokername}',
+        with_tractor_log=bool(tll),
+    )
+
     from piker.data import feed
     assert not feed._bus