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

@@ -1,60 +0,0 @@
----
-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

@@ -1,57 +0,0 @@
----
-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,26 +967,8 @@ 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(
+                raise ValueError(f"No contract could be found {con}")
-                    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:
@@ -1164,16 +1146,10 @@ class Client:
         try:
             con: Contract = self._contracts[symbol]
         except KeyError:
-            # require that the mkt's contract has been previously
+            # require that the symbol has been previously cached by
-            # qualified and cached (see `.symbols.get_mkt_info()`
+            # a data feed request - ensure we aren't making orders
-            # which is run for any feed-init OR lazily by the
+            # against non-known prices.
-            # order-request handler in `.broker`) - ensures we
+            raise RuntimeError("Can not order {symbol}, no live feed?")
-            # 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(
@@ -1632,17 +1608,6 @@ 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,
         *,
@@ -1656,10 +1621,10 @@ class MethodProxy:
 
         '''
         chan = self.chan
-        mid: int = next(self._mids)
+        await chan.send((meth, kwargs))
-        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
@@ -1669,51 +1634,24 @@ class MethodProxy:
             # except ConnectionError:
             #     self.reset()
 
-            match msg:
+            # print(f'NEXT MSG: {msg}')
-                # OUR method-call response B)
-                case {'mid': resp_mid, 'result': res} if (
-                    resp_mid == mid
-                ):
-                    return res
 
-                case {'mid': resp_mid, 'exception': err} if (
+            # TODO: py3.10 ``match:`` syntax B)
-                    resp_mid == mid
+            if 'result' in msg:
-                ):
+                res = msg.get('result')
-                    raise err
+                return res
 
-                # an "orphaned" response to some prior
+            elif 'exception' in msg:
-                # (cancelled) caller's request; drop it and
+                err = msg.get('exception')
-                # keep waiting for ours.
+                raise err
-                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
+            elif 'error' in msg:
-                # to the current caller as before.
+                etype, emsg = msg
-                case {'exception': err}:
+                log.warning(f'IB error relay: {emsg}')
-                    raise err
+                continue
 
-                case ('error', emsg):
+            else:
-                    log.warning(f'IB error relay: {emsg}')
+                log.warning(f'UNKNOWN IB MSG: {msg}')
-                    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,
@@ -1766,31 +1704,22 @@ async def open_aio_client_method_relay(
                 log.info('asyncio `Client` method-proxy SHUTDOWN!')
                 break
 
-            case (meth_name, kwargs, mid):
+            case (meth_name, kwargs):
+                meth_name, kwargs = msg
                 meth = getattr(client, meth_name)
 
                 try:
                     resp = await meth(**kwargs)
-                    # echo the msg back tagged with the
+                    # echo the msg back
-                    # req-id for caller-side correlation.
+                    chan.send_nowait({'result': resp})
-                    chan.send_nowait({
-                        'mid': mid,
-                        'result': resp,
-                    })
 
-                # XXX: relay ALL (non-cancel) errors to the
+                except (
-                # `trio`-side caller (which re-raises in the
+                    RequestError,
-                # `MethodProxy._run_method()` frame) instead
+
-                # of crashing this relay task and thus the
+                    # TODO: relay all errors to trio?
-                # whole proxy/dialog; critical post the
+                    # BaseException,
-                # (datad|brokerd)-split where eg. contract
+                ) as err:
-                # qualification failures are expected to be
+                    chan.send_nowait({'exception': err})
-                # 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,9 +89,8 @@ from .api import (
     MethodProxy,
 )
 from .symbols import (
-    cache_contract,
     con2fqme,
-    get_mkt_info,
+    # get_mkt_info,
 )
 from .ledger import (
     norm_trade_records,
@@ -139,7 +138,6 @@ async def handle_order_requests(
     ems_order_stream: tractor.MsgStream,
     accounts_def: dict[str, str],
     flows: OrderDialogs,
-    proxies: dict[str, MethodProxy],
 
 ) -> None:
 
@@ -182,58 +180,6 @@ 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
@@ -245,34 +191,15 @@ async def handle_order_requests(
                 reqid = int(reqid)
 
             # call our client api to submit the order
-            # NOTE: guard with order-error relay (vs. crashing
+            reqid = client.submit_limit(
-            # this dialog and thus ALL order ctl for the
+                oid=order.oid,
-            # backend) so one bad submission can't take down
+                symbol=order.symbol,
-            # the daemon's clearing loop.
+                price=order.price,
-            try:
+                action=order.action,
-                reqid = client.submit_limit(
+                size=order.size,
-                    oid=order.oid,
+                account=acct_number,
-                    symbol=fqme,
+                reqid=reqid,
-                    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(
@@ -621,14 +548,10 @@ async def open_trade_dialog(
 
 ) -> AsyncIterator[dict[str, Any]]:
 
-    # XXX: ONLY adjust the level of this (sub)mod's logger;
+    get_console_log(
-    # attaching a (stderr) handler (via `get_console_log()`)
+        level=loglevel,
-    # here would DOUBLE-print every record since the daemon
+        name=__name__,
-    # 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()
@@ -865,64 +788,6 @@ 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,
@@ -936,7 +801,6 @@ 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,40 +639,12 @@ 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:
-        cache_contract(
+        client: Client = proxy._aio_ns
-            proxy._aio_ns,
+        client._contracts[mkt.bs_fqme] = con
-            mkt,
+        client._cons2mkts[con] = 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,18 +93,6 @@ 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