brokerd: slim RPC caps + `ib` client-id offset

Caps-sec tightening now that `brokerd` is trading-only: NO `piker.data.*` (feed) mods are RPC-enabled in the (live, credentialed) trading actor anymore. Deats, - drop `_data_mods` for a minimal `_brokerd_service_mods` (just `piker.brokers._daemon`); dedup-compose with the backend's set in `spawn_brokerd()`. - `broker_init()` reads the backend's `_brokerd_mods` (fallback: `__enable_modules__` for flat backends). - fail fast in `spawn_brokerd()` via `validate.get_eps()` when a backend offers NO live order-ctl eps (eg. `kucoin`, `deribit`) -> tells the caller to use paper-mode instead of booting a dead actor; analogous warning in `datad_init()` for datad-ep-less backends. - offset `ib`'s default `client_id` per daemon-kind in `load_aio_clients()`: post-split BOTH `datad.ib` and `brokerd.ib` connect to the same gw/tws endpoint and the shared default (6116 + linear retry incrs) would collide and burn the full conn-timeout retry cycle; datad gets +16, ad-hoc (test/cli) actors +32. - drop the import-cleanup leftovers (`exceptiongroup`, `_FeedsBus` type-only import) and the now-resolved "expose datad" TODO in `.cli`. (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> Prompt-IO: ai/prompt-io/claude/20260610T171344Z_eee19de0_prompt_io.md
.data: cut feed layer over to `datad` actors
2026-06-10 13:14:24 -04:00 · 2026-06-10 13:13:44 -04:00 · 2026-06-10 13:12:59 -04:00 · 2026-06-10 13:12:26 -04:00 · 2026-06-10 13:11:42 -04:00 · 2026-06-10 13:11:05 -04:00
42 changed files with 1438 additions and 183 deletions
--- a/ai/prompt-io/claude/20260610T170859Z_75cefe10_prompt_io.md
+++ b/ai/prompt-io/claude/20260610T170859Z_75cefe10_prompt_io.md
@ -0,0 +1,59 @@
+---
+model: claude-fable-5[1m]
+service: claude
+session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
+timestamp: 2026-06-10T17:08:59Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T170859Z_75cefe10_prompt_io.md
+scope: code
+substantive: true
+raw_file: 20260610T170859Z_75cefe10_prompt_io.raw.md
+---
+
+## Prompt
+
+Session-initiating instruction (driving all 7 commits in
+this series):
+
+> ok i want you to become the distributed runtime and
+> concurrency expert for this project - namely acquire
+> a deep understanding of tractor and how it's used.
+> then i want you to attempt to factor our current
+> brokerd service daemon into 2 daemons: a brokerd
+> which only hosts live/paper trading endpoint tasks
+> [..] a new `datad` subdaemon which in a separate
+> subactor serves all the data feed service tasks [..]
+> give me a mega detailed plan on how to approach this,
+> and a staged approach for the implementation.
+
+Proximate driver for THIS commit: during the approved
+plan's stage-0 test gate the agent discovered test
+subactors were writing into the user's REAL
+`~/.config/piker/accounting/` files (a bogus paper fill
+landed in `account.kraken.paper.toml`) because the old
+test-dir override in `config.get_app_dir()` was
+commented out and could never work in spawned
+subactors anyway. Fix applied autonomously under the
+approved plan; no explicit per-fix user prompt.
+
+## Response summary
+
+Restore `pytest` config-dir isolation by resolving the
+per-test tmp dir lazily (at conf-path access time) from
+`tractor` runtime-vars, which propagate down the actor
+tree; route all conf-path derivation through
+`config.get_conf_dir()` so the override is effective in
+every (sub)actor.
+
+## Files changed
+
+- `piker/config.py` — add `_maybe_use_test_dir()`;
+  hook in `get_conf_dir()`; route `get_conf_path()` +
+  `load()` mkdir through it
+- `piker/accounting/_ledger.py` — derive ledger dir
+  via `config.get_conf_dir()` (not the global)
+- `piker/accounting/_pos.py` — same for account files
+
+## Human edits
+
+None — committed as generated.
--- a/ai/prompt-io/claude/20260610T170859Z_75cefe10_prompt_io.raw.md
+++ b/ai/prompt-io/claude/20260610T170859Z_75cefe10_prompt_io.raw.md
@ -0,0 +1,46 @@
+---
+model: claude-fable-5[1m]
+service: claude
+timestamp: 2026-06-10T17:08:59Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T170859Z_75cefe10_prompt_io.md
+---
+
+NOTE: diff-ref mode entry (code committed in the same
+commit as this log); backfilled from the live dev
+session transcript per the `/prompt-io` skill rules.
+
+> `git log -1 -p --follow -- piker/config.py`
+
+Generated: `config._maybe_use_test_dir()` — lazily
+reads `piker_test_dir` from
+`tractor.runtime._state._runtime_vars['piker_vars']`
+(pre-loaded by `open_piker_runtime()` from the
+`tests.conftest._open_test_pikerd()` overrides) and
+calls `_override_config_dir()` when set. Hooked at the
+top of `get_conf_dir()`; `get_conf_path()` and
+`load()`'s dir-creation rerouted through
+`get_conf_dir()`.
+
+> `git log -1 -p --follow -- piker/accounting/_ledger.py`
+> `git log -1 -p --follow -- piker/accounting/_pos.py`
+
+Generated: ledger/account dir derivation switched from
+the `config._config_dir` module global to
+`config.get_conf_dir()` + `mkdir(parents=True,
+exist_ok=True)` for nested tmp-dir creation.
+
+Key diagnostic reasoning (verbatim from session):
+
+The old (commented-out) `get_app_dir()` override gated
+on `'pytest' in sys.modules` which can NEVER work in
+spawned subactors (fresh procs, no pytest import); as
+a result test `paperboi`/daemon actors were writing
+into the user's REAL `~/.config/piker/accounting/`
+files. Evidence: `account.kraken.paper.toml` gained a
+single 0.001 xbtusdt clear stamped 2026-06-09T18:47Z
+(a test fill), and `trades_kraken_paper.toml` (252B)
+contained only that fill. The resolution must be
+checked lazily at config-path access time (NOT import
+time) since sub-actors only receive runtime-vars once
+their `tractor` runtime has fully booted.
--- a/ai/prompt-io/claude/20260610T171022Z_4485f2b9_prompt_io.md
+++ b/ai/prompt-io/claude/20260610T171022Z_4485f2b9_prompt_io.md
@ -0,0 +1,53 @@
+---
+model: claude-fable-5[1m]
+service: claude
+session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
+timestamp: 2026-06-10T17:10:22Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171022Z_4485f2b9_prompt_io.md
+scope: code
+substantive: true
+raw_file: 20260610T171022Z_4485f2b9_prompt_io.raw.md
+---
+
+## Prompt
+
+Same session-initiating `brokerd`-split instruction (see
+`20260610T170859Z_75cefe10_prompt_io.md`). Proximate
+driver: the approved plan's per-stage test gates could
+not run AT ALL — the branch base was broken vs.
+`tractor` git `main` (`AttributeError: 'tuple' object
+has no attribute 'unwrap'` at `pikerd` boot, stale
+`arbiter_sockaddr`/`.type`/`.size` API refs). The agent
+fixed forward autonomously to (re)establish the gate
+baseline, continuing the `repair_tests` branch lineage
+(verified already merged in ancestry via `git cherry`
+during a user-requested branch-overlap survey).
+
+## Response summary
+
+Port the service layer + test suites to current
+`tractor` APIs: addr-type normalization in
+`open_pikerd()`, `query_actor()`/`wait_for_actor()`
+kwarg renames, modern self-cancel absorption semantics
+in the cancel-method test harness, exc/position attr
+renames, a paper-EMS startup-budget bump and a syntax
+fix in `.deribit.api`.
+
+## Files changed
+
+- `piker/service/_actor_runtime.py` — `wrap_address()`
+  normalize before `.unwrap()` in `open_pikerd()`
+- `piker/service/_registry.py` — `check_for_service()`
+  -> `query_actor(regaddr=)` + 2-tuple yield +
+  `open_registry(addrs=)`
+- `piker/brokers/deribit/api.py` — missing comma in
+  `tractor.trionics` import tuple
+- `tests/test_services.py` — `registry_addr=` kwarg,
+  raddr unwraps, cancel-semantics harness rewrite,
+  `fail_after` 9 -> 19s
+- `tests/test_ems.py` — `.boxed_type`, `pp.cumsize`
+
+## Human edits
+
+None — committed as generated.
--- a/ai/prompt-io/claude/20260610T171022Z_4485f2b9_prompt_io.raw.md
+++ b/ai/prompt-io/claude/20260610T171022Z_4485f2b9_prompt_io.raw.md
@ -0,0 +1,56 @@
+---
+model: claude-fable-5[1m]
+service: claude
+timestamp: 2026-06-10T17:10:22Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171022Z_4485f2b9_prompt_io.md
+---
+
+NOTE: diff-ref mode entry (code committed in the same
+commit as this log); backfilled from the live dev
+session transcript per the `/prompt-io` skill rules.
+
+> `git log -1 -p --follow -- piker/service/_actor_runtime.py`
+
+Generated: normalize each registry addr via
+`tractor.discovery._addr.wrap_address()` before
+`.unwrap()`-ing for the `accept_addrs` bind check —
+entries may be raw `tuple`s when passed in from (test)
+client code. Import-path precedent taken from
+`piker/cli/__init__.py:336`.
+
+> `git log -1 -p --follow -- piker/service/_registry.py`
+
+Generated: `check_for_service()` ported to
+`tractor.query_actor(name, regaddr=...)` (kwarg was
+`arbiter_sockaddr=`), unpacking the new
+`(sockaddr, portal)` yield, and passing the
+now-required `open_registry(addrs=Registry.addrs)`.
+
+> `git log -1 -p --follow -- tests/test_services.py`
+> `git log -1 -p --follow -- tests/test_ems.py`
+> `git log -1 -p --follow -- piker/brokers/deribit/api.py`
+
+Key diagnostic reasoning (verbatim from session):
+
+- the "DID NOT RAISE ContextCancelled" failure: in this
+  test the client actor IS pikerd (in-proc), and
+  current `tractor` main absorbs a `ContextCancelled`
+  whose canceller is your own actor — self-requested
+  cancels now exit cleanly instead of raising; the
+  'sigint' variant propagates a bare collapsed
+  `KeyboardInterrupt` rather than a
+  `BaseExceptionGroup`.
+- the hard-coded `trio.fail_after(9)` startup budget is
+  marginal — full stack boot (pikerd -> emsd ->
+  brokerd.kraken -> paperboi + live kraken symbology
+  fetch) occasionally exceeds 9s -> bumped to 19s.
+- `RemoteActorError.type` -> `.boxed_type`;
+  `Position.size` -> `.cumsize` (the paper engine
+  populates `BrokerdPosition.size` from `pp.cumsize`).
+- overlap survey (user-requested): all of the
+  `repair_tests` branch commits are already in this
+  stack's ancestry; this commit finishes that branch's
+  port mission (its f4c4f1e2 fixed `conftest.py`'s
+  `arbiter_sockaddr` usage; this fixes the remaining
+  `test_services.py` + `check_for_service()` sites).
--- a/ai/prompt-io/claude/20260610T171105Z_bc6e18d7_prompt_io.md
+++ b/ai/prompt-io/claude/20260610T171105Z_bc6e18d7_prompt_io.md
@ -0,0 +1,54 @@
+---
+model: claude-fable-5[1m]
+service: claude
+session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
+timestamp: 2026-06-10T17:11:05Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171105Z_bc6e18d7_prompt_io.md
+scope: code
+substantive: true
+raw_file: 20260610T171105Z_bc6e18d7_prompt_io.raw.md
+---
+
+## Prompt
+
+Same session-initiating `brokerd`-split instruction (see
+`20260610T170859Z_75cefe10_prompt_io.md`); this is the
+approved plan's "stage 0" prep. Plan-shaping user
+decisions captured via in-session Q&A:
+
+- `datad.<broker>` topology: sibling of
+  `brokerd.<broker>` under `pikerd` (vs. child
+  subactor).
+- migration: hard cutover, staged by layer (no
+  dual-mode runtime flag).
+- post-split `brokerd` scope: trading-only +
+  EMS-lazy-spawned (charts/CLI never touch it).
+
+## Response summary
+
+Declare per-daemon-kind backend submod groups
+(`_datad_mods`/`_brokerd_mods`) in the split-style
+backends, keyed to the pre-existing
+`piker.data.validate._eps` contract; add a
+`validate.get_eps()` introspection helper; ws-import
+hygiene in `.kraken.broker`. Zero behavior change
+(`__enable_modules__` unions unchanged).
+
+## Files changed
+
+- `piker/data/validate.py` — add `get_eps()`
+- `piker/brokers/kraken/__init__.py` — mod groups
+- `piker/brokers/binance/__init__.py` — mod groups
+  (note: no `symbols.py`; search eps live in `.feed`)
+- `piker/brokers/deribit/__init__.py` — datad-only
+  groups (no `broker.py` yet)
+- `piker/brokers/ib/__init__.py` — add `'api'` to the
+  pre-existing `_datad_mods`
+- `piker/brokers/kraken/broker.py` — import
+  `NoBsWs`/`open_autorecon_ws` from
+  `piker.data._web_bs` directly
+
+## Human edits
+
+None — committed as generated.
--- a/ai/prompt-io/claude/20260610T171105Z_bc6e18d7_prompt_io.raw.md
+++ b/ai/prompt-io/claude/20260610T171105Z_bc6e18d7_prompt_io.raw.md
@ -0,0 +1,46 @@
+---
+model: claude-fable-5[1m]
+service: claude
+timestamp: 2026-06-10T17:11:05Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171105Z_bc6e18d7_prompt_io.md
+---
+
+NOTE: diff-ref mode entry (code committed in the same
+commit as this log); backfilled from the live dev
+session transcript per the `/prompt-io` skill rules.
+
+> `git log -1 -p --follow -- piker/data/validate.py`
+
+Generated: `get_eps(mod, kind) -> dict[str, Callable]`
+returning the daemon-kind's ep funcs defined by a
+backend mod, keyed by ep name, missing eps excluded;
+sourced from the existing `_eps` grouping table
+(`'middleware' | 'datad' | 'brokerd'`).
+
+> `git log -1 -p --follow -- piker/brokers/kraken/__init__.py`
+> `git log -1 -p --follow -- piker/brokers/binance/__init__.py`
+> `git log -1 -p --follow -- piker/brokers/deribit/__init__.py`
+> `git log -1 -p --follow -- piker/brokers/ib/__init__.py`
+> `git log -1 -p --follow -- piker/brokers/kraken/broker.py`
+
+Key exploration findings driving the design (from the
+planning phase, 3 parallel explore agents + 1 architect
+agent):
+
+- the codebase already anticipated this split:
+  `validate._eps` groups eps into 'datad' vs 'brokerd'
+  kinds and `ib/__init__.py` already declared
+  `_brokerd_mods`/`_datad_mods`; `brokers/_daemon.py`
+  carried the literal TODO "rename the daemon to datad
+  prolly once we split up broker vs. data tasks into
+  separate actors?".
+- the feared kraken feed<->broker "shared ws state"
+  hazard is mild: `NoBsWs`/`open_autorecon_ws`
+  originate in `piker.data._web_bs` (re-exported via
+  `.kraken.feed`); only `stream_messages` is feed-local
+  and it's a pure parser — in-process imports only,
+  each actor opens its own ws conn.
+- `enable_modules` gates RPC entry, NOT python imports,
+  so backend trading mods may keep importing feed-side
+  helpers in-process post-split.
--- a/ai/prompt-io/claude/20260610T171142Z_119d2c04_prompt_io.md
+++ b/ai/prompt-io/claude/20260610T171142Z_119d2c04_prompt_io.md
@ -0,0 +1,47 @@
+---
+model: claude-fable-5[1m]
+service: claude
+session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
+timestamp: 2026-06-10T17:11:42Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171142Z_119d2c04_prompt_io.md
+scope: code
+substantive: true
+raw_file: 20260610T171142Z_119d2c04_prompt_io.raw.md
+---
+
+## Prompt
+
+Same session-initiating `brokerd`-split instruction (see
+`20260610T170859Z_75cefe10_prompt_io.md`); this is the
+approved plan's "stage 1": introduce the `datad` daemon
+machinery additively (nothing routes through it yet).
+User-decided constraint applied: `datad.<broker>` is a
+SIBLING of `brokerd.<broker>` under `pikerd`, spawned
+via the existing `Services` + `maybe_spawn_daemon()`
+machinery.
+
+## Response summary
+
+New `piker/data/_daemon.py` hosting the
+`datad.<broker>` feed-only daemon-actor: lifetime
+fixture owning the actor-global `_FeedsBus`, init/spawn
+fns mirroring `.brokers._daemon` conventions and the
+`samplerd` sub-daemon precedent, plus root-mod
+registration, `piker.service` re-exports and a spawn
+test.
+
+## Files changed
+
+- `piker/data/_daemon.py` — NEW:
+  `_setup_persistent_datad()`, `datad_init()`,
+  `spawn_datad()`, `maybe_spawn_datad()`,
+  `_datad_service_mods`
+- `piker/service/_actor_runtime.py` — add
+  `piker.data._daemon` to `_root_modules`
+- `piker/service/__init__.py` — re-export spawn eps
+- `tests/test_services.py` — add `test_datad_spawn`
+
+## Human edits
+
+None — committed as generated.
--- a/ai/prompt-io/claude/20260610T171142Z_119d2c04_prompt_io.raw.md
+++ b/ai/prompt-io/claude/20260610T171142Z_119d2c04_prompt_io.raw.md
@ -0,0 +1,60 @@
+---
+model: claude-fable-5[1m]
+service: claude
+timestamp: 2026-06-10T17:11:42Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171142Z_119d2c04_prompt_io.md
+---
+
+NOTE: diff-ref mode entry (code committed in the same
+commit as this log); backfilled from the live dev
+session transcript per the `/prompt-io` skill rules.
+
+> `git log -1 -p --follow -- piker/data/_daemon.py`
+
+Generated symbols + key design decisions:
+
+- `_datad_service_mods: list[str]` — datad-always
+  enabled mods, the data-side successor to the old
+  `piker.brokers._daemon._data_mods` set; kept minimal
+  per the caps-sec model.
+- `_setup_persistent_datad()` — `@tractor.context`
+  lifetime fixture: console-log setup then allocates
+  the actor-global feed bus via
+  `feed.get_feed_bus(brokername, service_nursery)`
+  exactly as the old brokerd fixture did, pinned open
+  with `ctx.started()` + `sleep_forever()`.
+- `datad_init()` — actor name `f'datad.{brokername}'`;
+  copies backend `_spawn_kwargs` (CRITICAL for `ib`'s
+  `infect_asyncio=True`); builds `enable_modules` from
+  `getattr(brokermod, '_datad_mods',
+  getattr(brokermod, '__enable_modules__', []))` as
+  the flat-backend fallback.
+- `spawn_datad()` — `Services.actor_n.start_actor()` +
+  `Services.start_service_task()` exactly mirroring
+  `spawn_brokerd()`; dedup-composes enable mods via
+  `list(dict.fromkeys(...))`.
+- `maybe_spawn_datad()` — wraps `maybe_spawn_daemon(
+  service_name=f'datad.{brokername}', ...)`.
+
+> `git log -1 -p --follow -- piker/service/_actor_runtime.py`
+> `git log -1 -p --follow -- piker/service/__init__.py`
+> `git log -1 -p --follow -- tests/test_services.py`
+
+Design rationale (verbatim from session):
+
+- `_root_modules` must gain `piker.data._daemon` so
+  `pikerd_portal.run(spawn_datad, ...)` resolves in
+  the root.
+- the `Services`-based impl style deliberately mirrors
+  `spawn_brokerd()` so the eventual `tractor.hilevel`
+  `ServiceMngr` port (see the `service_mng_to_tractor`
+  branch's d8c21d44 prep, surfaced by the
+  user-requested branch-overlap survey) lands
+  symmetrically on both spawn fns.
+- mod placement (`piker/data/_daemon.py` vs.
+  generalizing `piker.brokers._daemon`) follows the
+  per-subsystem daemon-mod convention
+  (`.clearing._ems`, `.data._sampling`) and resolves
+  the existing TODO at `brokers/_daemon.py:49` ("move
+  this def to the `.data` subpkg").
--- a/ai/prompt-io/claude/20260610T171226Z_64181219_prompt_io.md
+++ b/ai/prompt-io/claude/20260610T171226Z_64181219_prompt_io.md
@ -0,0 +1,47 @@
+---
+model: claude-fable-5[1m]
+service: claude
+session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
+timestamp: 2026-06-10T17:12:26Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171226Z_64181219_prompt_io.md
+scope: code
+substantive: true
+raw_file: 20260610T171226Z_64181219_prompt_io.raw.md
+---
+
+## Prompt
+
+Same session-initiating `brokerd`-split instruction (see
+`20260610T170859Z_75cefe10_prompt_io.md`); this is the
+approved plan's "stage 2": decouple the clearing layer
+from `feed.portals` BEFORE the feed cutover so live
+trading works at every stage boundary. User-decided
+constraint applied: post-split `brokerd` is
+trading-only and spawned LAZILY only by `emsd`'s
+`open_brokerd_dialog()` path.
+
+## Response summary
+
+Kill the single coupling forcing feed + trading eps
+into one actor (`Router.open_trade_relays()` pulling
+its trades portal from `feed.portals[brokermod]`):
+`open_brokerd_dialog()` now (maybe) spawns/finds
+`brokerd.<broker>` itself and ONLY when a live
+trades-ep will actually open; paper mode never touches
+it. Pre-cutover this is a pure refactor (registry
+lookup finds the same feed-spawned daemon).
+
+## Files changed
+
+- `piker/clearing/_ems.py` — `open_brokerd_dialog()`
+  re-sig + inner `acquire_live_portal()`;
+  `Router.maybe_open_brokerd_dialog()` drops `portal`
+  param; `open_trade_relays()` drops the
+  `feed.portals` lookup
+- `piker/accounting/cli.py` — keyword-form `portal=`
+  override kept for the `piker ledger` ad-hoc actor
+
+## Human edits
+
+None — committed as generated.
--- a/ai/prompt-io/claude/20260610T171226Z_64181219_prompt_io.raw.md
+++ b/ai/prompt-io/claude/20260610T171226Z_64181219_prompt_io.raw.md
@ -0,0 +1,43 @@
+---
+model: claude-fable-5[1m]
+service: claude
+timestamp: 2026-06-10T17:12:26Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171226Z_64181219_prompt_io.md
+---
+
+NOTE: diff-ref mode entry (code committed in the same
+commit as this log); backfilled from the live dev
+session transcript per the `/prompt-io` skill rules.
+
+> `git log -1 -p --follow -- piker/clearing/_ems.py`
+
+Generated: new `open_brokerd_dialog()` signature
+`(brokermod, exec_mode, fqme=None, portal=None,
+loglevel=None)` with an inner `@acm
+acquire_live_portal()` that yields the caller-provided
+`portal` override (the `piker ledger` path) else
+`maybe_spawn_brokerd(brokermod.name)` — designated THE
+one place a live, credentialed `brokerd.<broker>` gets
+booted post-split. The eager
+`portal.open_context(trades_endpoint, ...)`
+construction moved inside that block so the paper
+short-circuit never acquires a live portal.
+
+> `git log -1 -p --follow -- piker/accounting/cli.py`
+
+Key analysis (verbatim from session):
+
+- `feed.portals` had exactly ONE trading consumer:
+  `piker/clearing/_ems.py:671` (`portal =
+  feed.portals[brokermod]`) — the single coupling
+  forcing feed + trading into one actor.
+- `piker ledger` (`accounting/cli.py:100-157`) is a
+  hidden consumer: it calls `broker_init()` directly,
+  spawns its own ad-hoc actor and passes the portal
+  positionally — the new signature must keep an
+  explicit `portal:` override param for this path.
+- stage sequencing: clearing decouple lands BEFORE the
+  feed cutover, otherwise `feed.portals` would hand
+  emsd a `datad` portal and live `open_trade_dialog`
+  RPC would fail.
--- a/ai/prompt-io/claude/20260610T171259Z_59d5d9a6_prompt_io.md
+++ b/ai/prompt-io/claude/20260610T171259Z_59d5d9a6_prompt_io.md
@ -0,0 +1,48 @@
+---
+model: claude-fable-5[1m]
+service: claude
+session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
+timestamp: 2026-06-10T17:12:59Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171259Z_59d5d9a6_prompt_io.md
+scope: code
+substantive: true
+raw_file: 20260610T171259Z_59d5d9a6_prompt_io.raw.md
+---
+
+## Prompt
+
+Same session-initiating `brokerd`-split instruction (see
+`20260610T170859Z_75cefe10_prompt_io.md`); this is the
+approved plan's "stage 3": the feed-layer hard cutover
+(user-decided: no dual-mode runtime flag, each stage
+lands fully working).
+
+## Response summary
+
+The topology flip: all data-feed consumers route to
+`datad.<broker>`; the `brokerd` lifetime fixture slims
+to trading-only; UI/CLI data-flavoured spawn sites
+repointed; test expectations inverted incl. an explicit
+negative assert that paper sessions spawn NO
+`brokerd.<broker>`.
+
+## Files changed
+
+- `piker/data/feed.py` — `open_feed()` ->
+  `maybe_spawn_datad()`; `open_feed_bus()` actor-name
+  assert -> `'datad'`; `install_brokerd_search` ->
+  `install_datad_search`; comment sweep
+- `piker/brokers/_daemon.py` —
+  `_setup_persistent_brokerd()` slimmed (feed-bus
+  alloc moved to the datad fixture)
+- `piker/ui/_app.py`, `piker/ui/cli.py`,
+  `piker/brokers/core.py`, `piker/brokers/cli.py`,
+  `piker/ui/kivy/option_chain.py` — spawn-site
+  repoints
+- `tests/test_services.py` — datad assertions + the
+  no-`brokerd.kraken` negative check
+
+## Human edits
+
+None — committed as generated.
--- a/ai/prompt-io/claude/20260610T171259Z_59d5d9a6_prompt_io.raw.md
+++ b/ai/prompt-io/claude/20260610T171259Z_59d5d9a6_prompt_io.raw.md
@ -0,0 +1,54 @@
+---
+model: claude-fable-5[1m]
+service: claude
+timestamp: 2026-06-10T17:12:59Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171259Z_59d5d9a6_prompt_io.md
+---
+
+NOTE: diff-ref mode entry (code committed in the same
+commit as this log); backfilled from the live dev
+session transcript per the `/prompt-io` skill rules.
+
+> `git log -1 -p --follow -- piker/data/feed.py`
+
+Generated: `open_feed()` spawn cutover with the
+`maybe_spawn_datad` import done "relative-direct" from
+`._daemon` — NOT via `piker.service` — to dodge a
+partial-init cycle: when `piker.data.feed` loads as
+part of `piker.service.__init__` executing its own
+`piker.data._daemon` import, the service pkg is
+mid-init and its `maybe_spawn_datad` binding does not
+exist yet. The `open_feed_bus()` local-state sanity
+assert flips `'brokerd' in servicename` ->
+`'datad' in servicename` (the only actor-name assert
+in the tree, verified by grep).
+
+> `git log -1 -p --follow -- piker/brokers/_daemon.py`
+
+Generated: `_setup_persistent_brokerd()` slimmed to
+console-log setup + pinned-open ctx; drops the bus
+alloc, the `assert not feed._bus`, the service nursery
+(backend `open_trade_dialog()` ctxs own their task
+trees) and the `eg.ExceptionGroup` handler. The
+`piker ledger` ad-hoc actor enters this same slimmed
+fixture — exactly what it needs.
+
+> `git log -1 -p --follow -- piker/ui/_app.py`
+> `git log -1 -p --follow -- piker/ui/cli.py`
+> `git log -1 -p --follow -- piker/brokers/core.py`
+> `git log -1 -p --follow -- piker/brokers/cli.py`
+> `git log -1 -p --follow -- piker/ui/kivy/option_chain.py`
+> `git log -1 -p --follow -- tests/test_services.py`
+
+Verification (from session): per-suite gates green at
+this commit (services 5-passed incl. the new negative
+assert, feeds 3-passed); a headless live smoke
+(`maybe_open_pikerd` + `open_feed(['xbtusdt.kraken'])`
+on an alt registry port) confirmed quotes flowing via
+`datad.kraken` + `samplerd` with `check_for_service(
+'brokerd.kraken') is None`. Known pre-existing flake
+documented: `test_multi_fill_positions`' second
+in-proc runtime boot wedges ~50% (zombie subactor w/
+unread parent-IPC bytes); reproduced with the split
+fully reverted so NOT a regression of this work.
--- a/ai/prompt-io/claude/20260610T171344Z_eee19de0_prompt_io.md
+++ b/ai/prompt-io/claude/20260610T171344Z_eee19de0_prompt_io.md
@ -0,0 +1,43 @@
+---
+model: claude-fable-5[1m]
+service: claude
+session: 32d15f9a-b2d3-4c26-bdc9-190219141a25
+timestamp: 2026-06-10T17:13:44Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171344Z_eee19de0_prompt_io.md
+scope: code
+substantive: true
+raw_file: 20260610T171344Z_eee19de0_prompt_io.raw.md
+---
+
+## Prompt
+
+Same session-initiating `brokerd`-split instruction (see
+`20260610T170859Z_75cefe10_prompt_io.md`); this is the
+approved plan's final "stage 4": caps-sec slimming of
+the (live, credentialed) trading actor + the `ib`
+dual-daemon `client_id` collision mitigation flagged in
+the plan's risk register.
+
+## Response summary
+
+`brokerd` loses ALL `piker.data.*` (feed) RPC mods;
+spawn fails fast for datad-only backends with a "use
+paper-mode" error; `ib`'s default api-gw `client_id`
+gets a per-daemon-kind offset so `datad.ib` +
+`brokerd.ib` don't collide on connect.
+
+## Files changed
+
+- `piker/brokers/_daemon.py` — `_data_mods` -> minimal
+  `_brokerd_service_mods`; `broker_init()` reads
+  `_brokerd_mods` (fallback `__enable_modules__`);
+  `spawn_brokerd()` fail-fast via `validate.get_eps()`
+- `piker/brokers/ib/api.py` — role-based `client_id`
+  offset in `load_aio_clients()`
+- `piker/cli/__init__.py` — resolved "expose datad"
+  TODO
+
+## Human edits
+
+None — committed as generated.
--- a/ai/prompt-io/claude/20260610T171344Z_eee19de0_prompt_io.raw.md
+++ b/ai/prompt-io/claude/20260610T171344Z_eee19de0_prompt_io.raw.md
@ -0,0 +1,41 @@
+---
+model: claude-fable-5[1m]
+service: claude
+timestamp: 2026-06-10T17:13:44Z
+git_ref: datad_service
+diff_cmd: git log -1 -p --follow -- ai/prompt-io/claude/20260610T171344Z_eee19de0_prompt_io.md
+---
+
+NOTE: diff-ref mode entry (code committed in the same
+commit as this log); backfilled from the live dev
+session transcript per the `/prompt-io` skill rules.
+
+> `git log -1 -p --follow -- piker/brokers/_daemon.py`
+
+Generated: the fail-fast originally landed in
+`broker_init()` but was relocated to `spawn_brokerd()`
+mid-implementation after realizing `piker ledger`
+calls `broker_init()` directly even for paper accounts
+on datad-only backends (would have crashed the cli);
+the service-spawn path is the correct enforcement
+seam. Error text:
+
+  Backend 'kucoin' offers NO `brokerd` (live
+  order-control) eps!? It is likely a datad-only
+  provider, use paper-mode for clearing instead.
+
+(verified live via a `trio.run()` unit check.)
+
+> `git log -1 -p --follow -- piker/brokers/ib/api.py`
+
+Generated: in `load_aio_clients()`, when `client_id`
+is the 6116 default: `datad`-named actors offset +16
+(disjoint from `brokerd`'s linear `client_id + i`
+retry range), other non-`brokerd` (ad-hoc test/cli)
+actors +32. Rationale from the plan's risk register:
+post-split BOTH per-broker daemons connect to the same
+TWS/gw endpoint; a shared default id collides and
+burns up to `connect_timeout * retries` (90s) in
+retry cycles.
+
+> `git log -1 -p --follow -- piker/cli/__init__.py`
--- a/ai/prompt-io/claude/README.md
+++ b/ai/prompt-io/claude/README.md
@ -0,0 +1,34 @@
+# AI Prompt I/O Log — claude
+
+This directory tracks prompt inputs and model
+outputs for AI-assisted development using
+`claude` (claude-code CLI).
+
+## Policy
+
+Prompt logging follows the
+[NLNet generative AI policy][nlnet-ai].
+All substantive AI contributions are logged
+with:
+- Model name and version
+- Timestamps
+- The prompts that produced the output
+- Unedited model output (`.raw.md` files)
+
+[nlnet-ai]: https://nlnet.nl/foundation/policies/generativeAI/
+
+## Usage
+
+Entries are created by the `/prompt-io` skill
+or automatically via `/commit-msg` integration.
+
+Each commit carrying AI-generated changes links
+to its provenance entry via a `Prompt-IO:`
+commit-msg trailer; entries use "diff-ref mode"
+(pointers into `git log -p` instead of verbatim
+code copies) to avoid duplicating committed
+code.
+
+Human contributors remain accountable for all
+code decisions. AI-generated content is never
+presented as human-authored work.
--- a/piker/accounting/_ledger.py
+++ b/piker/accounting/_ledger.py
@ -324,10 +324,13 @@ def load_ledger(
    ldir: Path = (
        dirpath
        or
-        config._config_dir / 'accounting' / 'ledgers'
+        config.get_conf_dir() / 'accounting' / 'ledgers'
    )
    if not ldir.is_dir():
-        ldir.mkdir()
+        ldir.mkdir(
+            parents=True,
+            exist_ok=True,
+        )

    fname = f'trades_{brokername}_{acctid}.toml'
    fpath: Path = ldir / fname
--- a/piker/accounting/_pos.py
+++ b/piker/accounting/_pos.py
@ -785,9 +785,16 @@ def load_account(
    legacy_fn: str = f'pps.{brokername}.{acctid}.toml'
    fn: str = f'account.{brokername}.{acctid}.toml'

-    dirpath: Path = dirpath or (config._config_dir / 'accounting')
+    dirpath: Path = (
+        dirpath
+        or
+        (config.get_conf_dir() / 'accounting')
+    )
    if not dirpath.is_dir():
-        dirpath.mkdir()
+        dirpath.mkdir(
+            parents=True,
+            exist_ok=True,
+        )

    conf, path = config.load(
        path=dirpath / fn,
--- a/piker/accounting/cli.py
+++ b/piker/accounting/cli.py
@ -143,12 +143,15 @@ def sync(
                    # (what the EMS normally does internall) B)
                    open_brokerd_dialog(
                        brokermod,
-                        portal,
                        exec_mode=(
                            'paper'
                            if account == 'paper'
                            else 'live'
                        ),
+                        # use our own ad-hoc-spawned actor,
+                        # do NOT (spawn and) use the
+                        # `brokerd.<broker>` service daemon!
+                        portal=portal,
                        loglevel=loglevel,
                    ) as (
                        brokerd_stream,
--- a/piker/brokers/_daemon.py
+++ b/piker/brokers/_daemon.py
@ -25,10 +25,8 @@ from contextlib import (
 )
 from types import ModuleType
 from typing import (
-    TYPE_CHECKING,
    AsyncContextManager,
 )
-import exceptiongroup as eg

 import tractor
 import trio
@ -40,27 +38,20 @@ from piker.log import (
 from . import _util
 from . import get_brokermod

-if TYPE_CHECKING:
-    from ..data import _FeedsBus
-
 log = get_logger(name=__name__)

-# `brokerd` enabled modules
-# TODO: move this def to the `.data` subpkg..
+# `brokerd`-actor-always-enabled mods.
 # NOTE: keeping this list as small as possible is part of our caps-sec
-# model and should be treated with utmost care!
-_data_mods: str = [
-    'piker.brokers.core',
-    'piker.brokers.data',
+# model and should be treated with utmost care! In particular NO
+# `piker.data.*` feed mods should be enabled in this (live,
+# credentialed) trading actor; all data-feed serving is the
+# domain of the `datad.<broker>` sibling daemon, see
+# `piker.data._daemon._datad_service_mods`.
+_brokerd_service_mods: list[str] = [
    'piker.brokers._daemon',
-    'piker.data',
-    'piker.data.feed',
-    'piker.data._sampling'
 ]


-# TODO: we should rename the daemon to datad prolly once we split up
-# broker vs. data tasks into separate actors?
@tractor.context
 async def _setup_persistent_brokerd(
    ctx: tractor.Context,
@ -69,9 +60,15 @@ async def _setup_persistent_brokerd(

 ) -> None:
    '''
-    Allocate a actor-wide service nursery in ``brokerd``
-    such that feeds can be run in the background persistently by
-    the broker backend as needed.
+    Trading-only daemon (lifetime) fixture: console logging
+    setup and a pinned-open context for service mgmt.
+
+    All data-feed-bus state now lives in the (data-feed-only)
+    `datad.<brokername>` sibling daemon, see
+    `piker.data._daemon._setup_persistent_datad()`; this
+    actor hosts only the backend's `open_trade_dialog()`
+    (live order-control) ep-task(s) which manage their own
+    task trees per `tractor.Context`.

    '''
    # NOTE: we only need to setup logging once (and only) here
@ -87,46 +84,12 @@ async def _setup_persistent_brokerd(
    )
    assert log.name == _util.subsys

-    # further, set the log level on any broker broker specific
-    # logger instance.
+    # unblock caller
+    await ctx.started()

-    from piker.data import feed
-    assert not feed._bus
-
-    # allocate a nursery to the bus for spawning background
-    # tasks to service client IPC requests, normally
-    # `tractor.Context` connections to explicitly required
-    # `brokerd` endpoints such as:
-    # - `stream_quotes()`,
-    # - `manage_history()`,
-    # - `allocate_persistent_feed()`,
-    # - `open_symbol_search()`
-    # NOTE: see ep invocation details inside `.data.feed`.
-    try:
-        async with (
-            # tractor.trionics.collapse_eg(),
-            trio.open_nursery() as service_nursery
-        ):
-            bus: _FeedsBus = feed.get_feed_bus(
-                brokername,
-                service_nursery,
-            )
-            assert bus is feed._bus
-
-            # unblock caller
-            await ctx.started()
-
-            # we pin this task to keep the feeds manager active until the
-            # parent actor decides to tear it down
-            await trio.sleep_forever()
-
-    except eg.ExceptionGroup:
-        # TODO: likely some underlying `brokerd` IPC connection
-        # broke so here we handle a respawn and re-connect attempt!
-        # This likely should pair with development of the OCO task
-        # nusery in dev over @ `tractor` B)
-        # https://github.com/goodboy/tractor/pull/363
-        raise
+    # we pin this task to keep the daemon active until the
+    # parent actor decides to tear it down
+    await trio.sleep_forever()


 def broker_init(
@ -147,8 +110,10 @@ def broker_init(

    This includes:
    - load the appropriate <brokername>.py pkg module,
-    - reads any declared `__enable_modules__: listr[str]` which will be
-      passed to `tractor.ActorNursery.start_actor(enabled_modules=<this>)`
+    - reads any declared `_brokerd_mods: list[str]` (falling
+      back to the full `__enable_modules__` set for
+      not-yet-split backends) which will be passed to
+      `tractor.ActorNursery.start_actor(enable_modules=)`
      at actor start time,
    - deliver a references to the daemon lifetime fixture, which
      for now is always the `_setup_persistent_brokerd()` context defined
@ -183,8 +148,14 @@ def broker_init(
    ]
    for submodname in getattr(
        brokermod,
-        '__enable_modules__',
-        [],
+        '_brokerd_mods',
+        # fallback for (flat, less mature) backends which
+        # don't yet declare a daemon-kind mod split.
+        getattr(
+            brokermod,
+            '__enable_modules__',
+            [],
+        ),
    ):
        subpath: str = f'{modpath}.{submodname}'
        enabled.append(subpath)
@ -212,6 +183,22 @@ async def spawn_brokerd(
        f'backend: {brokername!r}'
    )

+    # fail fast on (data-only) backends which don't offer
+    # ANY live order-control eps; the caller should instead
+    # be using paper-mode (and thus never spawning us)!
+    from ..data.validate import get_eps
+    brokerd_eps: dict = get_eps(
+        get_brokermod(brokername),
+        'brokerd',
+    )
+    if not brokerd_eps:
+        raise RuntimeError(
+            f'Backend {brokername!r} offers NO `brokerd` '
+            f'(live order-control) eps!?\n'
+            f'It is likely a datad-only provider, use '
+            f'paper-mode for clearing instead.\n'
+        )
+
    (
        brokermode,
        tractor_kwargs,
@ -233,7 +220,11 @@ async def spawn_brokerd(
    dname: str = tractor_kwargs.pop('name')  # f'brokerd.{brokername}'
    portal = await Services.actor_n.start_actor(
        dname,
-        enable_modules=_data_mods + tractor_kwargs.pop('enable_modules'),
+        enable_modules=list(dict.fromkeys(
+            _brokerd_service_mods
+            +
+            tractor_kwargs.pop('enable_modules')
+        )),
        debug_mode=Services.debug_mode,
        **tractor_kwargs
    )
--- a/piker/brokers/binance/init.py
+++ b/piker/brokers/binance/init.py
@ -52,9 +52,25 @@ __all__ = [
 ]


-# `brokerd` modules
-__enable_modules__: list[str] = [
+# per-daemon-kind (sub)mod groups: declares which of our
+# submods host the eps run by each daemon-actor kind as
+# defined by `piker.data.validate._eps`.
+# NOTE: `get_mkt_info` and `open_symbol_search` both live
+# in `.feed` for this backend (no `symbols.py`).
+_brokerd_mods: list[str] = [
    'api',
-    'feed',
    'broker',
 ]
+
+_datad_mods: list[str] = [
+    'api',
+    'feed',
+]
+
+
+# tractor RPC enable arg
+__enable_modules__: list[str] = list(dict.fromkeys(
+    _brokerd_mods
+    +
+    _datad_mods
+))
--- a/piker/brokers/cli.py
+++ b/piker/brokers/cli.py
@ -35,7 +35,7 @@ from piker.log import (
    get_logger,
 )
 from ..service import (
-    maybe_spawn_brokerd,
+    maybe_spawn_datad,
    maybe_open_pikerd,
 )
 from ..brokers import (
@ -187,7 +187,7 @@ def brokercheck(config, broker):

    '''
    async def bcheck_main():
-        async with maybe_spawn_brokerd(broker) as portal:
+        async with maybe_spawn_datad(broker) as portal:
            await portal.run(run_test, broker)
            await portal.cancel_actor()

@ -317,7 +317,7 @@ def record(config, rate, name, dhost, filename):
        return

    async def main(tries):
-        async with maybe_spawn_brokerd(
+        async with maybe_spawn_datad(
            tries=tries, loglevel=loglevel
        ) as portal:
            # run app "main"
--- a/piker/brokers/core.py
+++ b/piker/brokers/core.py
@ -30,7 +30,7 @@ import trio

 from piker.log import get_logger
 from . import get_brokermod
-from ..service import maybe_spawn_brokerd
+from ..service import maybe_spawn_datad
 from . import open_cached_client
 from ..accounting import MktPair

@ -172,7 +172,7 @@ async def symbol_search(
        # await tractor.devx._debug.maybe_init_greenback()
        # tractor.pause_from_sync()

-        async with maybe_spawn_brokerd(
+        async with maybe_spawn_datad(
            mod.name,
            infect_asyncio=getattr(
                mod,
--- a/piker/brokers/deribit/init.py
+++ b/piker/brokers/deribit/init.py
@ -47,13 +47,25 @@ __all__ = [
 ]


-# tractor RPC enable arg
-__enable_modules__: list[str] = [
+# per-daemon-kind (sub)mod groups: declares which of our
+# submods host the eps run by each daemon-actor kind as
+# defined by `piker.data.validate._eps`.
+# NOTE: datad-only backend (no `broker.py` yet)!
+_brokerd_mods: list[str] = []
+
+_datad_mods: list[str] = [
    'api',
    'feed',
-#   'broker',
 ]

+
+# tractor RPC enable arg
+__enable_modules__: list[str] = list(dict.fromkeys(
+    _brokerd_mods
+    +
+    _datad_mods
+))
+
 # passed to ``tractor.ActorNursery.start_actor()``
 _spawn_kwargs = {
    'infect_asyncio': True,
--- a/piker/brokers/deribit/api.py
+++ b/piker/brokers/deribit/api.py
@ -37,7 +37,7 @@ from rapidfuzz import process as fuzzy
 import numpy as np
 from tractor.trionics import (
    broadcast_receiver,
-    maybe_open_context
+    maybe_open_context,
    collapse_eg,
 )
 from tractor import to_asyncio
--- a/piker/brokers/ib/init.py
+++ b/piker/brokers/ib/init.py
@ -65,17 +65,18 @@ _brokerd_mods: list[str] = [
 ]

 _datad_mods: list[str] = [
+    'api',
    'feed',
    'symbols',
 ]


 # tractor RPC enable arg
-__enable_modules__: list[str] = (
+__enable_modules__: list[str] = list(dict.fromkeys(
    _brokerd_mods
    +
    _datad_mods
-)
+))

 # passed to ``tractor.ActorNursery.start_actor()``
 _spawn_kwargs = {
--- a/piker/brokers/ib/api.py
+++ b/piker/brokers/ib/api.py
@ -1340,6 +1340,21 @@ async def load_aio_clients(
    ib = None
    client = None

+    # XXX: post (brokerd vs. datad)-split BOTH per-broker
+    # daemons connect to the same API gw/tws endpoint(s); to
+    # avoid `clientId` collisions (and the long conn-timeout
+    # retry cycle they cause) we offset the data-daemon's
+    # default id-range to be disjoint from `brokerd.ib`'s
+    # (which also retries with `client_id + i` increments).
+    if client_id == 6116:  # the default from above
+        aname: str = tractor.current_actor().name
+        if 'datad' in aname:
+            client_id += 16
+        # ad-hoc (test/cli) actors get their own range to
+        # avoid clashing with any live daemon-tree's conns.
+        elif 'brokerd' not in aname:
+            client_id += 32
+
    # attempt to get connection info from config; if no .toml entry
    # exists, we try to load from a default localhost connection.
    localhost = '127.0.0.1'
--- a/piker/brokers/kraken/init.py
+++ b/piker/brokers/kraken/init.py
@ -66,10 +66,24 @@ __all__ = [
 ]


-# tractor RPC enable arg
-__enable_modules__: list[str] = [
+# per-daemon-kind (sub)mod groups: declares which of our
+# submods host the eps run by each daemon-actor kind as
+# defined by `piker.data.validate._eps`.
+_brokerd_mods: list[str] = [
    'api',
    'broker',
+]
+
+_datad_mods: list[str] = [
+    'api',
    'feed',
    'symbols',
 ]
+
+
+# tractor RPC enable arg
+__enable_modules__: list[str] = list(dict.fromkeys(
+    _brokerd_mods
+    +
+    _datad_mods
+))
--- a/piker/brokers/kraken/broker.py
+++ b/piker/brokers/kraken/broker.py
@ -73,10 +73,12 @@ from piker.log import (
    get_logger,
 )
 from piker.data import open_symcache
-from . import api
-from .feed import (
+from piker.data._web_bs import (
    open_autorecon_ws,
    NoBsWs,
+)
+from . import api
+from .feed import (
    stream_messages,
 )
 from .ledger import (
--- a/piker/clearing/_ems.py
+++ b/piker/clearing/_ems.py
@ -335,9 +335,14 @@ class TradesRelay(Struct):
@acm
 async def open_brokerd_dialog(
    brokermod: ModuleType,
-    portal: tractor.Portal,
    exec_mode: str,
    fqme: str|None = None,
+
+    # XXX: explicit (already spawned) trading-actor override,
+    # currently only used by the `piker ledger` cli which
+    # boots its own ad-hoc `brokerd`-like actor; normally we
+    # (lazily) spawn/find the `brokerd.<broker>` daemon here.
+    portal: tractor.Portal|None = None,
    loglevel: str|None = None,

 ) -> tuple[
@ -351,6 +356,10 @@ async def open_brokerd_dialog(
    paper engine instance depending on live trading support for the
    broker backend, configuration, or client code usage.

+    NOTE: this is now the ONE place where a (live, credentialed)
+    `brokerd.<broker>` daemon-actor gets (lazily) booted; pure
+    data/paper sessions should never spawn one!
+
    '''
    get_console_log(
        level=loglevel,
@ -416,16 +425,29 @@ async def open_brokerd_dialog(
        )
        exec_mode: str = 'paper'

-    if (
-        trades_endpoint is not None
-        or
-        exec_mode != 'paper'
-    ):
-        # open live brokerd trades endpoint
-        open_trades_endpoint = portal.open_context(
-            trades_endpoint,
-            loglevel=loglevel,
+    @acm
+    async def acquire_live_portal():
+        '''
+        Deliver a portal to the (live, credentialed) trading
+        actor hosting the backend's `open_trade_dialog()` ep:
+        either the caller-provided override or the (maybe
+        lazily spawned) `brokerd.<broker>` service daemon.
+
+        '''
+        if portal is not None:
+            yield portal
+            return
+
+        # XXX: the ONE (normal) place a `brokerd.<broker>`
+        # daemon-actor gets booted in the runtime B)
+        from piker.brokers._daemon import (
+            maybe_spawn_brokerd,
        )
+        async with maybe_spawn_brokerd(
+            brokermod.name,
+            loglevel=loglevel,
+        ) as live_portal:
+            yield live_portal

    @acm
    async def maybe_open_paper_ep():
@ -437,7 +459,14 @@ async def open_brokerd_dialog(
                return

        # open trades-dialog endpoint with backend broker
-        async with open_trades_endpoint as msg:
+        async with (
+            acquire_live_portal() as live_portal,
+
+            live_portal.open_context(
+                trades_endpoint,
+                loglevel=loglevel,
+            ) as msg,
+        ):
            ctx, first = msg

            # runtime indication that the backend can't support live
@ -581,7 +610,6 @@ class Router(Struct):
    async def maybe_open_brokerd_dialog(
        self,
        brokermod: ModuleType,
-        portal: tractor.Portal,
        exec_mode: str,
        fqme: str,
        loglevel: str,
@ -606,7 +634,6 @@ class Router(Struct):

        async with open_brokerd_dialog(
            brokermod=brokermod,
-            portal=portal,
            exec_mode=exec_mode,
            fqme=fqme,
            loglevel=loglevel,
@ -668,7 +695,6 @@ class Router(Struct):
            brokername, _, _, _ = unpack_fqme(fqme)
            brokermod = feed.mods[brokername]
            broker = brokermod.name
-            portal = feed.portals[brokermod]

            # XXX: this should be initial price quote from target provider
            flume = feed.flumes[fqme]
@ -682,7 +708,6 @@ class Router(Struct):

            async with self.maybe_open_brokerd_dialog(
                brokermod=brokermod,
-                portal=portal,
                exec_mode=exec_mode,
                fqme=fqme,
                loglevel=loglevel,
--- a/piker/cli/init.py
+++ b/piker/cli/init.py
@ -250,7 +250,6 @@ def cli(

    # TODO: load endpoints from `conf::[network].pikerd`
    # - pikerd vs. regd, separate registry daemon?
-    # - expose datad vs. brokerd?
    # - bind emsd with certain perms on public iface?
    regaddrs: list[tuple[str, int]] = regaddr or [(
        _default_registry_host,
--- a/piker/config.py
+++ b/piker/config.py
@ -188,6 +188,48 @@ def _override_config_dir(
    _config_dir = path


+def _maybe_use_test_dir() -> None:
+    '''
+    When running under the `pytest` harness, override
+    the config dir to the per-test-tmp dir "passed down"
+    the actor tree via `tractor`'s runtime-vars
+    inheritance mechanism.
+
+    See the `tractor_runtime_overrides` usage in our
+    `tests.conftest._open_test_pikerd()` as well as
+    `.service._actor_runtime.open_piker_runtime()` for
+    the root-actor's pre-loading of the var state.
+
+    NOTE: this must be checked lazily at config-path
+    access time (NOT import time) since sub-actors only
+    receive runtime-vars once their `tractor` runtime
+    has fully booted.
+
+    '''
+    global _config_dir
+    import tractor
+    actor = tractor.current_actor(
+        err_on_no_runtime=False,
+    )
+    if actor is None:
+        return
+
+    rvs: dict = tractor.runtime._state._runtime_vars
+    pvars: dict|None = rvs.get('piker_vars')
+    if (
+        pvars
+        and
+        (testdir := pvars.get('piker_test_dir'))
+    ):
+        testdirpath = Path(testdir)
+        assert testdirpath.exists(), (
+            f'piker test harness might be borked!?\n'
+            f'testdirpath: {testdirpath!r}\n'
+        )
+        if _config_dir != testdirpath:
+            _override_config_dir(testdirpath)
+
+
 def _conf_fn_w_ext(
    name: str,
 ) -> str:
@ -201,6 +243,7 @@ def get_conf_dir() -> Path:
    on the local filesystem.

    '''
+    _maybe_use_test_dir()
    return _config_dir


@ -226,7 +269,7 @@ def get_conf_path(
        assert str(conf_name) in _conf_names

    fn = _conf_fn_w_ext(conf_name)
-    return _config_dir / Path(fn)
+    return get_conf_dir() / Path(fn)


 def repodir() -> Path:
@ -271,8 +314,9 @@ def load(

    '''
    # create the $HOME/.config/piker dir if dne
-    if not _config_dir.is_dir():
-        _config_dir.mkdir(
+    conf_dir: Path = get_conf_dir()
+    if not conf_dir.is_dir():
+        conf_dir.mkdir(
            parents=True,
            exist_ok=True,
        )
--- a/piker/data/_daemon.py
+++ b/piker/data/_daemon.py
@ -0,0 +1,300 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+Data-daemon-actor "endpoint-hooks": the service task entry
+points for `datad.<brokername>`, the per-provider real-time
+and historical market-data feed daemon.
+
+The (data vs. broker)d-split mirrors the ep-groupings in
+`piker.data.validate._eps`: this daemon hosts all `'datad'`
+eps (live quotes, history loading, symbology search) while
+its `brokerd.<brokername>` sibling hosts only the live
+order-control (and thus credentialed) `'brokerd'` eps.
+
+'''
+from __future__ import annotations
+from contextlib import (
+    asynccontextmanager as acm,
+)
+from types import ModuleType
+from typing import (
+    TYPE_CHECKING,
+    AsyncContextManager,
+)
+
+import tractor
+import trio
+
+from piker.log import (
+    get_logger,
+    get_console_log,
+)
+from . import _util
+
+if TYPE_CHECKING:
+    from .feed import _FeedsBus
+
+log = get_logger(name=__name__)
+
+# `datad`-actor-always-enabled mods: the data-side successor
+# to the old `piker.brokers._daemon._data_mods` set.
+# NOTE: keeping this list as small as possible is part of
+# our caps-sec model and should be treated with utmost care!
+_datad_service_mods: list[str] = [
+    'piker.brokers.core',
+    'piker.brokers.data',
+    'piker.data',
+    'piker.data.feed',
+    'piker.data._sampling',
+    'piker.data._daemon',
+]
+
+
+@tractor.context
+async def _setup_persistent_datad(
+    ctx: tractor.Context,
+    brokername: str,
+    loglevel: str|None = None,
+
+) -> None:
+    '''
+    Allocate an actor-wide service nursery in this
+    `datad.<brokername>` actor such that data-feed tasks
+    (shm writer-samplers, history backfillers, symbology
+    loaders) can be run in the background persistently by
+    the provider backend as needed.
+
+    '''
+    # NOTE: we only need to setup logging once (and only)
+    # here since all hosted daemon tasks will reference
+    # this same log instance's (actor local) state and thus
+    # don't require any further (level) configuration on
+    # their own B)
+    actor: tractor.Actor = tractor.current_actor()
+    tll: str = actor.loglevel
+    log = get_console_log(
+        level=loglevel or tll,
+        name=f'{_util.subsys}.{brokername}',
+        with_tractor_log=bool(tll),
+    )
+    assert log.name == _util.subsys
+
+    from piker.data import feed
+    assert not feed._bus
+
+    # allocate a nursery to the bus for spawning background
+    # tasks which service client IPC requests, normally
+    # `tractor.Context` connections to explicitly required
+    # `datad` endpoints such as:
+    # - `stream_quotes()`,
+    # - `manage_history()`,
+    # - `allocate_persistent_feed()`,
+    # - `open_symbol_search()`
+    # NOTE: see ep invocation details inside `.data.feed`.
+    async with (
+        trio.open_nursery() as service_nursery
+    ):
+        bus: _FeedsBus = feed.get_feed_bus(
+            brokername,
+            service_nursery,
+        )
+        assert bus is feed._bus
+
+        # unblock caller
+        await ctx.started()
+
+        # we pin this task to keep the feeds manager active
+        # until the parent actor decides to tear it down
+        await trio.sleep_forever()
+
+
+def datad_init(
+    brokername: str,
+    loglevel: str|None = None,
+
+    **start_actor_kwargs,
+
+) -> tuple[
+    ModuleType,
+    dict,
+    AsyncContextManager,
+]:
+    '''
+    Given an input broker name, load all named arguments
+    which can be passed for daemon endpoint + context spawn
+    as required in every `datad.<brokername>` (actor)
+    service.
+
+    This includes:
+    - load the appropriate <brokername>.py pkg module,
+    - reads any declared `_datad_mods: list[str]` (falling
+      back to the full `__enable_modules__` set for
+      not-yet-split backends) which will be passed to
+      `tractor.ActorNursery.start_actor(enable_modules=)`
+      at actor start time,
+    - deliver a reference to the daemon lifetime fixture,
+      which for now is always the
+      `_setup_persistent_datad()` context defined above.
+
+    '''
+    from piker.brokers import get_brokermod
+    from .validate import get_eps
+    brokermod = get_brokermod(brokername)
+    modpath: str = brokermod.__name__
+
+    # warn (but don't bail) when the backend is missing
+    # some/all of the `datad` ep contract defined by
+    # `piker.data.validate._eps`.
+    datad_eps: dict = get_eps(brokermod, 'datad')
+    if not datad_eps:
+        log.warning(
+            f'Backend {brokername!r} offers NO `datad` '
+            f'(data-feed) eps!?\n'
+            f'Most feed/chart functionality will be '
+            f'broken for this provider..\n'
+        )
+
+    start_actor_kwargs['name'] = f'datad.{brokername}'
+
+    # XXX CRITICAL: include any backend-declared spawn
+    # kwargs, eg. `{'infect_asyncio': True}` required by
+    # `ib`'s embedded `asyncio`-mode `ib_async` usage!
+    start_actor_kwargs.update(
+        getattr(
+            brokermod,
+            '_spawn_kwargs',
+            {},
+        )
+    )
+
+    # lookup actor-enabled modules declared by the backend
+    # offering the `datad` endpoint(s).
+    enabled: list[str]
+    enabled = start_actor_kwargs['enable_modules'] = [
+        __name__,  # so that eps from THIS mod can be invoked
+        modpath,
+    ]
+    for submodname in getattr(
+        brokermod,
+        '_datad_mods',
+        # fallback for (flat, less mature) backends which
+        # don't yet declare a daemon-kind mod split.
+        getattr(
+            brokermod,
+            '__enable_modules__',
+            [],
+        ),
+    ):
+        subpath: str = f'{modpath}.{submodname}'
+        enabled.append(subpath)
+
+    return (
+        brokermod,
+        start_actor_kwargs,  # to `ActorNursery.start_actor()`
+
+        # XXX see impl above; contains all (actor global)
+        # setup/teardown expected in all `datad` actor
+        # instances.
+        _setup_persistent_datad,
+    )
+
+
+async def spawn_datad(
+    brokername: str,
+    loglevel: str|None = None,
+
+    **tractor_kwargs,
+
+) -> bool:
+
+    log.info(
+        f'Spawning data-daemon,\n'
+        f'backend: {brokername!r}'
+    )
+
+    (
+        brokermod,
+        tractor_kwargs,
+        daemon_fixture_ep,
+    ) = datad_init(
+        brokername,
+        loglevel,
+        **tractor_kwargs,
+    )
+
+    # ask `pikerd` to spawn a new sub-actor and manage it
+    # under its actor nursery
+    from piker.service import Services
+
+    dname: str = tractor_kwargs.pop('name')  # f'datad.{brokername}'
+    enable_mods: list[str] = list(dict.fromkeys(
+        _datad_service_mods
+        +
+        tractor_kwargs.pop('enable_modules')
+    ))
+    portal = await Services.actor_n.start_actor(
+        dname,
+        enable_modules=enable_mods,
+        debug_mode=Services.debug_mode,
+        **tractor_kwargs,
+    )
+
+    # NOTE: the service mngr expects an already spawned
+    # actor + its portal ref in order to do non-blocking
+    # setup of the `datad` service nursery.
+    await Services.start_service_task(
+        dname,
+        portal,
+
+        # signature of target root-task endpoint
+        daemon_fixture_ep,
+        brokername=brokername,
+        loglevel=loglevel,
+    )
+    return True
+
+
+@acm
+async def maybe_spawn_datad(
+    brokername: str,
+    loglevel: str|None = None,
+
+    **pikerd_kwargs,
+
+) -> tractor.Portal:
+    '''
+    Helper to spawn a datad service *from* a client who
+    wishes to use the sub-actor-daemon but is fine with
+    re-using any existing and contactable `datad`.
+
+    Mas o menos, acts as a cached-actor-getter factory.
+
+    '''
+    from piker.service import maybe_spawn_daemon
+
+    async with maybe_spawn_daemon(
+        service_name=f'datad.{brokername}',
+        service_task_target=spawn_datad,
+        spawn_args={
+            'brokername': brokername,
+        },
+        loglevel=loglevel,
+
+        **pikerd_kwargs,
+
+    ) as portal:
+        yield portal
--- a/piker/data/feed.py
+++ b/piker/data/feed.py
@ -17,7 +17,7 @@
 '''
 Data feed apis and infra.

-This module is enabled for ``brokerd`` daemons and includes mostly
+This module is enabled for ``datad`` daemons and includes mostly
 endpoints and middleware to support our real-time, provider agnostic,
 live market quotes layer. Historical data loading and processing is also
 initiated in parts of the feed bus startup but business logic and
@ -54,8 +54,11 @@ from piker.accounting import (
 )
 from piker.types import Struct
 from piker.brokers import get_brokermod
-from piker.service import (
-    maybe_spawn_brokerd,
+# NOTE: must be a "relative-direct" import (NOT via
+# `piker.service`) to avoid a partial-init cycle when this
+# mod is loaded as part of `piker.service.__init__`.
+from ._daemon import (
+    maybe_spawn_datad,
 )
 from piker.calc import humanize
 from ._util import (
@ -110,7 +113,7 @@ class _FeedsBus(Struct):
    '''
    Data feeds broadcaster and persistence management.

-    This is a brokerd side api used to manager persistent real-time
+    This is a datad side api used to manager persistent real-time
    streams that can be allocated and left alive indefinitely. A bus is
    associated one-to-one with a particular broker backend where the
    "bus" refers so a multi-symbol bus where quotes are interleaved in
@ -249,7 +252,7 @@ async def allocate_persistent_feed(
    '''
    Create and maintain a "feed bus" which allocates tasks for real-time
    streaming and optional historical data storage per broker/data provider
-    backend; this normally task runs *in* a `brokerd` actor.
+    backend; this normally task runs *in* a `datad` actor.

    If none exists, this allocates a ``_FeedsBus`` which manages the
    lifetimes of streaming tasks created for each requested symbol.
@ -318,8 +321,8 @@ async def allocate_persistent_feed(
    #   at max capacity.
    # - the same ideas ^ but when a local core is maxxed out (like how
    #   binance does often with hft XD
-    # - if a brokerd is non-local then we can't just allocate a mem
-    #   channel here and have the brokerd write it, we instead need
+    # - if a datad is non-local then we can't just allocate a mem
+    #   channel here and have the datad write it, we instead need
    #   a small streaming machine around the remote feed which can then
    #   do the normal work of sampling and writing shm buffers
    #   (depending on if we want sampling done on the far end or not?)
@ -499,7 +502,7 @@ async def open_feed_bus(
    # (after we also group them in a nice `/dev/shm/piker/` subdir).
    # ensure we are who we think we are
    servicename = tractor.current_actor().name
-    assert 'brokerd' in servicename
+    assert 'datad' in servicename
    assert brokername in servicename

    bus: _FeedsBus = get_feed_bus(brokername)
@ -509,12 +512,12 @@ async def open_feed_bus(
    for symbol in symbols:

        # if no cached feed for this symbol has been created for this
-        # brokerd yet, start persistent stream and shm writer task in
+        # datad yet, start persistent stream and shm writer task in
        # service nursery
        flume = bus.feeds.get(symbol)
        if flume is None:
            # allocate a new actor-local stream bus which
-            # will persist for this `brokerd`'s service lifetime.
+            # will persist for this `datad`'s service lifetime.
            async with bus.task_lock:
                await bus.nursery.start(
                    partial(
@ -721,7 +724,7 @@ class Feed(Struct):
            mods = {name: self.mods[name] for name in brokers}

        if len(mods) == 1:
-            # just pass the brokerd stream directly if only one provider
+            # just pass the datad stream directly if only one provider
            # was detected.
            stream = self.streams[list(brokers)[0]]
            async with stream.subscribe() as bstream:
@ -763,7 +766,7 @@ class Feed(Struct):


@acm
-async def install_brokerd_search(
+async def install_datad_search(

    portal: tractor.Portal,
    brokermod: ModuleType,
@ -812,7 +815,7 @@ async def maybe_open_feed(
    ReceiveChannel[dict[str, Any]],
 ):
    '''
-    Maybe open a data to a ``brokerd`` daemon only if there is no
+    Maybe open a data feed to a ``datad`` daemon only if there is no
    local one for the broker-symbol pair, if one is cached use it wrapped
    in a tractor broadcast receiver.

@ -885,14 +888,14 @@ async def open_feed(
        providers.setdefault(mod, []).append(bs_fqme)
        feed.mods[mod.name] = mod

-    # one actor per brokerd for now
-    brokerd_ctxs = []
+    # one actor per datad for now
+    datad_ctxs = []
    for brokermod, bfqmes in providers.items():

-        # if no `brokerd` for this backend exists yet we spawn
+        # if no `datad` for this backend exists yet we spawn
        # a daemon actor for it.
-        brokerd_ctxs.append(
-            maybe_spawn_brokerd(
+        datad_ctxs.append(
+            maybe_spawn_datad(
                brokermod.name,
                loglevel=loglevel
            )
@ -900,7 +903,7 @@ async def open_feed(

    portals: tuple[tractor.Portal]
    async with trionics.gather_contexts(
-        brokerd_ctxs,
+        datad_ctxs,
    ) as portals:

        bus_ctxs: list[AsyncContextManager] = []
@ -937,9 +940,9 @@ async def open_feed(
                    tick_throttle=tick_throttle,

                    # XXX: super important to avoid
-                    # the brokerd from some other
+                    # the datad from some other
                    # backend overruning the task here
-                    # bc some other brokerd took longer
+                    # bc some other datad took longer
                    # to startup before we hit the `.open_stream()`
                    # loop below XD .. really we should try to do each
                    # of these stream open sequences sequentially per
@ -1008,7 +1011,7 @@ async def open_feed(
                    assert stream
                    feed.streams[brokermod.name] = stream

-                    # apply `brokerd`-common stream to each flume
+                    # apply `datad`-common stream to each flume
                    # tracking a live market feed from that provider.
                    for fqme, flume in feed.flumes.items():
                        if brokermod.name == flume.mkt.broker:
--- a/piker/data/validate.py
+++ b/piker/data/validate.py
@ -91,6 +91,24 @@ _eps: dict[str, list[str]] = {
 }


+def get_eps(
+    mod: ModuleType,
+    kind: str,  # 'middleware' | 'datad' | 'brokerd'
+
+) -> dict[str, Callable]:
+    '''
+    Return the daemon-kind's ep funcs defined by the backend
+    `mod`, keyed by ep name; any eps from `_eps[kind]` not
+    implemented by the backend are excluded.
+
+    '''
+    return {
+        name: ep
+        for name in _eps[kind]
+        if (ep := getattr(mod, name, None))
+    }
+
+
 def validate_backend(
    mod: ModuleType,
    syms: list[str],
--- a/piker/service/init.py
+++ b/piker/service/init.py
@ -56,3 +56,7 @@ from ..brokers._daemon import (
    spawn_brokerd as spawn_brokerd,
    maybe_spawn_brokerd as maybe_spawn_brokerd,
 )
+from ..data._daemon import (
+    spawn_datad as spawn_datad,
+    maybe_spawn_datad as maybe_spawn_datad,
+)
--- a/piker/service/_actor_runtime.py
+++ b/piker/service/_actor_runtime.py
@ -157,6 +157,7 @@ _root_modules: list[str] = [
    __name__,
    'piker.service._daemon',
    'piker.brokers._daemon',
+    'piker.data._daemon',

    'piker.clearing._ems',
    'piker.clearing._client',
@ -214,7 +215,13 @@ async def open_pikerd(
        trio.open_nursery() as service_tn,
    ):
        for addr in reg_addrs:
-            uaddr: tuple = addr.unwrap()
+            # normalize to a wrapped `tractor` addr-type;
+            # entries may be raw `tuple`s when passed in
+            # from (test) client code.
+            wladdr = tractor.discovery._addr.wrap_address(
+                addr,
+            )
+            uaddr: tuple = wladdr.unwrap()
            if (
                uaddr not in root_actor.accept_addrs
            ):
--- a/piker/service/_registry.py
+++ b/piker/service/_registry.py
@ -225,10 +225,13 @@ async def check_for_service(

    '''
    async with (
-        open_registry(ensure_exists=False) as reg_addr,
+        open_registry(
+            addrs=Registry.addrs,
+            ensure_exists=False,
+        ) as reg_addrs,
        tractor.query_actor(
            service_name,
-            arbiter_sockaddr=reg_addr,
-        ) as sockaddr,
+            regaddr=reg_addrs[0],
+        ) as (sockaddr, _),
    ):
        return sockaddr
--- a/piker/ui/_app.py
+++ b/piker/ui/_app.py
@ -32,12 +32,12 @@ from . import _event
 from . import _search
 from ..accounting import unpack_fqme
 from ..data._symcache import open_symcache
-from ..data.feed import install_brokerd_search
+from ..data.feed import install_datad_search
 from ..log import (
    get_logger,
    get_console_log,
 )
-from ..service import maybe_spawn_brokerd
+from ..service import maybe_spawn_datad
 from ._exec import run_qtractor

 log = get_logger(__name__)
@ -50,16 +50,16 @@ async def load_provider_search(
 ) -> None:

    name = brokermod.name
-    log.info(f'loading brokerd for {name}..')
+    log.info(f'loading datad for {name}..')

    async with (

-        maybe_spawn_brokerd(
+        maybe_spawn_datad(
            name,
            loglevel=loglevel
        ) as portal,

-        install_brokerd_search(
+        install_datad_search(
            portal,
            brokermod,
        ),
--- a/piker/ui/cli.py
+++ b/piker/ui/cli.py
@ -27,7 +27,7 @@ from ..cli import (
    load_trans_eps,
 )
 from .. import watchlists as wl
-from ..service import maybe_spawn_brokerd
+from ..service import maybe_spawn_datad


 _config_dir = click.get_app_dir('piker')
@ -69,7 +69,7 @@ def monitor(config, rate, name, dhost, test, tl):
    from .kivy.monitor import _async_main

    async def main():
-        async with maybe_spawn_brokerd(
+        async with maybe_spawn_datad(
            brokername=brokermod.name,
            loglevel=loglevel
        ) as portal:
@ -118,7 +118,7 @@ def optschain(
    from .kivy.option_chain import _async_main

    async def main():
-        async with maybe_spawn_brokerd(
+        async with maybe_spawn_datad(
            loglevel=loglevel
        ):
            # run app "main"
--- a/piker/ui/kivy/option_chain.py
+++ b/piker/ui/kivy/option_chain.py
@ -495,7 +495,7 @@ async def _async_main(

    async with trio.open_nursery() as nursery:
        # get a portal to the data feed daemon
-        async with tractor.wait_for_actor('brokerd') as portal:
+        async with tractor.wait_for_actor('datad') as portal:

            # set up a pager view for large ticker lists
            chain = await new_chain_ui(
--- a/tests/test_ems.py
+++ b/tests/test_ems.py
@ -151,7 +151,7 @@ def load_and_check_pos(
            # is the same the fqme.
            pp: Position = table.pps[ppmsg.symbol]

-            assert ppmsg.size == pp.size
+            assert ppmsg.size == pp.cumsize
            assert ppmsg.avg_price == pp.ppu

            yield pp
@ -179,7 +179,7 @@ def test_ems_err_on_bad_broker(
        # NOTE: emsd should error on the actor's enabled modules
        # import phase, when looking for a backend named `doggy`.
        except tractor.RemoteActorError as re:
-            assert re.type is ModuleNotFoundError
+            assert re.boxed_type is ModuleNotFoundError

    run_and_tollerate_cancels(load_bad_fqme)

--- a/tests/test_services.py
+++ b/tests/test_services.py
@ -53,11 +53,12 @@ def test_runtime_boot(

            tractor.wait_for_actor(
                'pikerd',
-                arbiter_sockaddr=daemon_addr,
+                registry_addr=daemon_addr,
            ) as portal,
        ):
-            assert pikerd_portal.channel.raddr == daemon_addr
-            assert pikerd_portal.channel.raddr == portal.channel.raddr
+            uw_raddr: tuple = pikerd_portal.chan.raddr.unwrap()
+            assert uw_raddr == daemon_addr
+            assert uw_raddr == portal.chan.raddr.unwrap()

            # no service tasks should be started
            assert not services.service_tasks
@ -65,6 +66,41 @@ def test_runtime_boot(
    trio.run(main)


+def test_datad_spawn(
+    open_test_pikerd: AsyncContextManager,
+    loglevel: str,
+
+) -> None:
+    '''
+    Verify the new (data-feed-only) `datad.<broker>` daemon
+    can be spawned/registered as a `pikerd` sub-service via
+    the `maybe_spawn_datad()` factory.
+
+    '''
+    from piker.service import maybe_spawn_datad
+
+    backend: str = 'kraken'
+    datad_name: str = f'datad.{backend}'
+
+    async def main():
+        async with (
+            open_test_pikerd() as (_, _, _, services),
+
+            maybe_spawn_datad(
+                backend,
+                loglevel=loglevel,
+            ) as portal,
+        ):
+            assert portal
+
+            async with ensure_service(datad_name):
+                assert (
+                    datad_name in services.service_tasks
+                )
+
+    trio.run(main)
+
+
 def test_ensure_datafeed_actors(
    open_test_pikerd: AsyncContextManager,
    loglevel: str,
@ -72,14 +108,14 @@ def test_ensure_datafeed_actors(

 ) -> None:
    '''
-    Verify that booting a data feed starts a `brokerd`
+    Verify that booting a data feed starts a `datad`
    actor and a singleton global `samplerd` and opening
    an order mode in paper opens the `paperboi` service.

    '''
-    actor_name: str = 'brokerd'
+    actor_name: str = 'datad'
    backend: str = 'kraken'
-    brokerd_name: str = f'{actor_name}.{backend}'
+    datad_name: str = f'{actor_name}.{backend}'

    async def main():
        async with (
@ -94,7 +130,7 @@ def test_ensure_datafeed_actors(
            await feed.pause()

            async with (
-                ensure_service(brokerd_name),
+                ensure_service(datad_name),
                ensure_service('samplerd'),
            ):
                await trio.sleep(0.1)
@ -108,7 +144,7 @@ async def ensure_service(
    sockaddr: tuple[str, int] | None = None,
 ) -> None:
    async with find_service(name) as portal:
-        remote_sockaddr = portal.channel.raddr
+        remote_sockaddr: tuple = portal.chan.raddr.unwrap()
        print(f'FOUND `{name}` @ {remote_sockaddr}')

        if sockaddr:
@ -131,40 +167,49 @@ def run_test_w_cancel_method(
        "was remotely cancelled by remote actor (\'pikerd\'")

    if cancel_method == 'sigint':
-        with pytest.raises(
+        # XXX: with modern `tractor` the (single-exc)
+        # group is collapsed so a bare KBI normally
+        # propagates; tolerate either form.
+        with pytest.raises((
+            KeyboardInterrupt,
            BaseExceptionGroup,
-        ) as exc_info:
+        )) as exc_info:
            trio.run(main)

-        multi = exc_info.value
+        err = exc_info.value
+        match err:
+            case BaseExceptionGroup():
+                for suberr in err.exceptions:
+                    match suberr:
+                        # ensure we receive a remote
+                        # cancellation error caused by the
+                        # pikerd root actor.
+                        case tractor.ContextCancelled():
+                            assert (
+                                cancelled_msg
+                                in
+                                suberr.args[0]
+                            )

-        for suberr in multi.exceptions:
-            match suberr:
-                # ensure we receive a remote cancellation error caused
-                # by the pikerd root actor since we used the
-                # `.cancel_service()` API above B)
-                case tractor.ContextCancelled():
-                    assert cancelled_msg in suberr.args[0]
+                        case KeyboardInterrupt():
+                            pass

-                case KeyboardInterrupt():
-                    pass
+                        case _:
+                            pytest.fail(
+                                f'Unexpected error {suberr}'
+                            )

-                case _:
-                    pytest.fail(f'Unexpected error {suberr}')
+            case KeyboardInterrupt():
+                pass

    elif cancel_method == 'services':
-
-        # XXX NOTE: oddly, when you pass --pdb to pytest, i think since
-        # we also use that to enable the underlying tractor debug mode,
-        # it causes this to not raise for some reason? So if you see
-        # that while changing this test.. it's prolly that.
-
-        with pytest.raises(
-            tractor.ContextCancelled
-        ) as exc_info:
-            trio.run(main)
-
-        assert cancelled_msg in exc_info.value.args[0]
+        # XXX: cancelling our own sub-service via
+        # `Services.cancel_service()` is a *self*
+        # requested cancel: modern `tractor` absorbs the
+        # resulting `ContextCancelled` (canceller is our
+        # own actor) so the runtime tears down gracefully
+        # with NO error raised to the opener.
+        trio.run(main)

    else:
        pytest.fail(f'Test is broken due to {cancel_method}')
@ -182,9 +227,9 @@ def test_ensure_ems_in_paper_actors(

 ) -> None:

-    actor_name: str = 'brokerd'
    backend: str = 'kraken'
-    brokerd_name: str = f'{actor_name}.{backend}'
+    datad_name: str = f'datad.{backend}'
+    brokerd_name: str = f'brokerd.{backend}'

    async def main():

@ -197,7 +242,9 @@ def test_ensure_ems_in_paper_actors(
        # ensure we timeout after is startup is too slow.
        # TODO: something like this should be our start point for
        # benchmarking end-to-end startup B)
-        with trio.fail_after(9):
+        # NOTE: includes a live (kraken) symbology fetch so
+        # the budget needs some headroom for net latency..
+        with trio.fail_after(19):
            async with (
                open_test_pikerd() as (_, _, _, services),

@ -226,15 +273,25 @@ def test_ensure_ems_in_paper_actors(

                async with (
                    ensure_service('emsd'),
-                    ensure_service(brokerd_name),
+                    ensure_service(datad_name),
                    ensure_service(f'paperboi.{backend}'),
                ):
                    for name in pikerd_subservices:
                        assert name in services.service_tasks

-                    # brokerd.kraken actor should have been started
-                    # implicitly by the ems.
-                    assert brokerd_name in services.service_tasks
+                    # datad.kraken actor should have been
+                    # started implicitly by the feed layer.
+                    assert datad_name in services.service_tasks
+
+                    # XXX: paper-mode sessions should NEVER
+                    # boot a (live, credentialed) `brokerd`;
+                    # only emsd's `open_brokerd_dialog()`
+                    # live-ep path is allowed to spawn it!
+                    assert (
+                        brokerd_name
+                        not in
+                        services.service_tasks
+                    )

                    print('ALL SERVICES STARTED, cancelling runtime with:\n'
                          f'-> {cancel_method}')
Author	SHA1	Message	Date
Gud Boi	f15f8178a3	brokerd: slim RPC caps + `ib` client-id offset Caps-sec tightening now that `brokerd` is trading-only: NO `piker.data.*` (feed) mods are RPC-enabled in the (live, credentialed) trading actor anymore. Deats, - drop `_data_mods` for a minimal `_brokerd_service_mods` (just `piker.brokers._daemon`); dedup-compose with the backend's set in `spawn_brokerd()`. - `broker_init()` reads the backend's `_brokerd_mods` (fallback: `__enable_modules__` for flat backends). - fail fast in `spawn_brokerd()` via `validate.get_eps()` when a backend offers NO live order-ctl eps (eg. `kucoin`, `deribit`) -> tells the caller to use paper-mode instead of booting a dead actor; analogous warning in `datad_init()` for datad-ep-less backends. - offset `ib`'s default `client_id` per daemon-kind in `load_aio_clients()`: post-split BOTH `datad.ib` and `brokerd.ib` connect to the same gw/tws endpoint and the shared default (6116 + linear retry incrs) would collide and burn the full conn-timeout retry cycle; datad gets +16, ad-hoc (test/cli) actors +32. - drop the import-cleanup leftovers (`exceptiongroup`, `_FeedsBus` type-only import) and the now-resolved "expose datad" TODO in `.cli`. (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> Prompt-IO: ai/prompt-io/claude/20260610T171344Z_eee19de0_prompt_io.md	2026-06-10 13:14:24 -04:00
Gud Boi	eee19de090	.data: cut feed layer over to `datad` actors The topology flip: all data-feed consumers now route to the new `datad.<broker>` sibling daemon; `brokerd` becomes trading-only and is ONLY ever booted lazily by `emsd`'s `open_brokerd_dialog()` (see prior commit). Chart-only and paper sessions run with zero (live, credentialed) `brokerd` procs B) Deats, - `open_feed()` -> `maybe_spawn_datad()` (NB: imported relative-direct from `._daemon` to dodge a partial-init cycle via `piker.service`); flip the `open_feed_bus()` actor-name assert to `'datad'`; comment sweep. - slim `_setup_persistent_brokerd()` to a trading-only fixture: console logging + pinned-open ctx; the feed-bus alloc moves to `_setup_persistent_datad()` and backend `open_trade_dialog()` ctxs own their own task trees. (the `piker ledger` ad-hoc actor enters this same slimmed fixture - exactly what it needs.) - repoint data-flavoured spawn sites to `maybe_spawn_datad`: `.ui._app` symbol-search (+ rename `install_brokerd_search` -> `install_datad_search`), `.brokers.core.symbol_search()`, `.brokers.cli` `brokercheck`/`record`, legacy kivy `.ui.cli` + `option_chain`'s `wait_for_actor()`. - invert `tests.test_services` expectations: feed/EMS-paper flows must spawn `datad.kraken` and `paperboi.kraken` with an explicit negative assert that NO `brokerd.kraken` service task exists. (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> Prompt-IO: ai/prompt-io/claude/20260610T171259Z_59d5d9a6_prompt_io.md	2026-06-10 13:13:44 -04:00
Gud Boi	59d5d9a66d	.clearing: lazily spawn `brokerd` from `emsd` Drop the ONE coupling that forces feed + trading eps into the same actor: `Router.open_trade_relays()` pulling its trades-dialog portal from `feed.portals[brokermod]`. Instead `open_brokerd_dialog()` now (maybe) spawns/finds `brokerd.<broker>` itself via `maybe_spawn_brokerd()` and ONLY when a live trades-ep will actually be opened; the paper-mode short-circuit never touches it, so post feed-cutover paper sessions will run with zero `brokerd` procs. Pre-cutover this is a pure refactor: the registry lookup just finds the same feed-spawned daemon. Deats, - new `open_brokerd_dialog()` sig: portal acquisition moves inside via an `acquire_live_portal()` helper; keep an explicit `portal: Portal\|None` override for the `piker ledger` cli which boots its own ad-hoc actor. - `Router.maybe_open_brokerd_dialog()` drops its `portal` param; `open_trade_relays()` drops the `feed.portals` lookup entirely. - `.accounting.cli`: pass `portal=` by keyword. (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> Prompt-IO: ai/prompt-io/claude/20260610T171226Z_64181219_prompt_io.md	2026-06-10 13:12:59 -04:00
Gud Boi	6418121923	Add `datad` daemon machinery to `.data` First half of the `brokerd` split: a new per-provider data-feed-only daemon-actor `datad.<broker>` to (soon) host all `validate._eps['datad']` eps (live quotes, history loading, symbology search) leaving `brokerd` for live order ctl only. Purely additive; nothing routes through it yet. Deats, - new `piker.data._daemon` mod mirroring the `.brokers._daemon` conventions (and the `samplerd` sub-daemon precedent): - `_setup_persistent_datad()` lifetime fixture owning the actor-global `_FeedsBus` alloc. - `datad_init()` building `enable_modules` from the backend's `_datad_mods` (falling back to `__enable_modules__` for not-yet-split backends) and copying `_spawn_kwargs` (critical for `ib`'s `infect_asyncio`). - `spawn_datad()`/`maybe_spawn_datad()` wrapping `Services` + `maybe_spawn_daemon()`. - add `piker.data._daemon` to `_root_modules` so `pikerd` can run `spawn_datad()` requests. - re-export the spawn eps from `piker.service`. - add `test_datad_spawn` verifying actor boot + service registration via `ensure_service('datad.kraken')`. Note the `Services`-based impl style deliberately mirrors `spawn_brokerd()` so the eventual `tractor.hilevel` `ServiceMngr` port (see the `service_mng_to_tractor` branch's d8c21d44 prep work) lands symmetrically on both. (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> Prompt-IO: ai/prompt-io/claude/20260610T171142Z_119d2c04_prompt_io.md	2026-06-10 13:12:26 -04:00
Gud Boi	119d2c0495	Declare per-daemon-kind backend mod groups Prep for the `brokerd` -> (`datad` + `brokerd`) actor split by having each (split-style) backend declare which of its submods host which daemon-kind's eps, exactly per the `piker.data.validate._eps` groupings; `ib` already had `_brokerd_mods`/`_datad_mods` so extend the convention to `kraken`, `binance` and `deribit` (and add `'api'` to ib's datad set since both kinds need the `Client` layer). `__enable_modules__` stays as the (deduped) union so this is a ZERO behavior change; flat backends (`kucoin` etc.) just don't declare the split yet. Also, - add `validate.get_eps()` returning a backend's defined eps per daemon-kind for spawn-time introspection. - import `NoBsWs`/`open_autorecon_ws` from `piker.data._web_bs` directly in `.kraken.broker` (they were only re-exported via `.kraken.feed`) so the trading mod doesn't depend on the feed mod for ws primitives. (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> Prompt-IO: ai/prompt-io/claude/20260610T171105Z_bc6e18d7_prompt_io.md	2026-06-10 13:11:42 -04:00
Gud Boi	bc6e18d7b4	Port service+tests to latest `tractor` APIs Continue the `repair_tests`-branch mission (already merged in this stack's ancestry, see `f4c4f1e2` which ported `conftest.py`) by fixing the remaining drift breakage vs. `tractor` git `main`; without these NOTHING boots since the `tractor.Address` port in `604e5fcf`. Deats, - normalize reg addrs via `wrap_address()` in `open_pikerd()` before `.unwrap()`-ing; entries may be raw `tuple`s when passed in from (test) client code. - port `check_for_service()` to `query_actor(regaddr=)` (was `arbiter_sockaddr=`) incl. its 2-tuple yield and the now-required `open_registry(addrs=)` arg. - `wait_for_actor(registry_addr=)` + `.chan.raddr.unwrap()` raw-tuple compares in `test_runtime_boot` and `ensure_service()`. - update `run_test_w_cancel_method()` for modern `tractor` cancel semantics: self-requested sub-service cancels are absorbed (no `ContextCancelled` raised to the opener) and single-exc groups collapse to a bare KBI. - `RemoteActorError.boxed_type` (was `.type`) and `Position.cumsize` (was `.size`) renames in tests. - bump the paper-EMS startup budget 9 -> 19s; it includes a live (kraken) symbology fetch so needs net headroom. - woops, add the missing comma in `.deribit.api`'s `tractor.trionics` import tuple.. (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> Prompt-IO: ai/prompt-io/claude/20260610T171022Z_4485f2b9_prompt_io.md	2026-06-10 13:11:05 -04:00
Gud Boi	4485f2b9ce	Fix `pytest` config-dir isolation in subactors The old (commented-out) `get_app_dir()` override gated on `'pytest' in sys.modules` which can NEVER work in spawned subactors (fresh procs, no pytest import); as a result test `paperboi`/daemon actors were writing into the user's REAL `~/.config/piker/accounting/` files.. friggin yikes. Deats, - add `config._maybe_use_test_dir()` which lazily (at conf-path access time, NOT import time) reads the `piker_test_dir` entry from `tractor.runtime._state._runtime_vars['piker_vars']` as pre-loaded by `open_piker_runtime()` from the `tests.conftest._open_test_pikerd()` overrides. - hook it in `get_conf_dir()` and route `get_conf_path()` + `load()`'s mkdir through `get_conf_dir()`. - route `.accounting._ledger` / `._pos` dir derivation through `config.get_conf_dir()` (was reading the `_config_dir` global directly, bypassing the override); also `mkdir(parents=True, exist_ok=True)` for nested tmp-dir creation. (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> Prompt-IO: ai/prompt-io/claude/20260610T170859Z_75cefe10_prompt_io.md	2026-06-10 13:10:09 -04:00