26 changed files with 210 additions and 1853 deletions
--- a/ai/prompt-io/20261007T165453Z_bde2457b_prompt_io.md
+++ b/ai/prompt-io/20261007T165453Z_bde2457b_prompt_io.md
@ -1,52 +0,0 @@
---
-created_at: 2026-10-07T16:54:53Z
-generated_by:
-  harness: opencode
-  provider: openai
-  model: gpt-6.1-sol
-git_ref: bde2457b
-scope: code
-substantive: true
-raw_file: 20261007T165453Z_bde2457b_prompt_io.raw.md
---
-
-## Prompt
-
-The user requested fixing all seven local PR94 review findings and
-exporting the review for human inspection in Tuicr. They asked that a
-Tractor pin adjustment be considered before adding Piker workarounds.
-
-## Response summary
-
-Correct the endpoint defaults, Click argument shape, daemon binding and
-registrar option handling, chart fallback, and stray debugger pause.
-Keep the absent datad implementation's test outside this landing
-branch. Add offline startup coverage and export original findings
-without modifying review states. The checked newer Tractor address
-branch still rejects mixed transports, so leave the dependency pin.
-
-## Files changed
-
- `config/conf.toml`: TCP-only default endpoints.
- `piker/cli/__init__.py`: repeatable and intact multiaddresses.
- `piker/service/_actor_runtime.py`: validate daemon listeners.
- `piker/service/_daemon.py`: initialize and consume registrar options.
- `piker/ui/cli.py`: fallback to automatic chart binding.
- `piker/brokers/binance/feed.py`: remove unconditional debugger entry.
- `tests/test_services.py`: defer test requiring missing datad API.
- `tests/test_startup.py`: CLI/config and service lifecycle regressions.
-
-## Human edits
-
-The human identified that datad tests arrived through the agent's
-mixed cherry-picks and explicitly requested deferring that test. They
-requested a chart default for existing configs, expanded the task to
-all review findings, and directed checking whether a newer Tractor pin
-could reduce the fix scope. No direct manual source edits were observed
-during these changes; acceptance remains with the human.
-
-During the human review pass, the human confirmed removing the Binance
-pause and correcting CLI endpoint parsing. They requested an explicit
-template note about this pin's single-transport limitation and detailed
-comments distinguishing registrar discovery endpoints from daemon bind
-endpoints. Those explanations were added at the human's direction.
--- a/ai/prompt-io/20261007T165453Z_bde2457b_prompt_io.raw.md
+++ b/ai/prompt-io/20261007T165453Z_bde2457b_prompt_io.raw.md
@ -1,61 +0,0 @@
---
-created_at: 2026-10-07T16:54:53Z
-generated_by:
-  harness: opencode
-  provider: openai
-  model: gpt-6.1-sol
-git_ref: bde2457b
-diff_cmd: git diff HEAD~1..HEAD
---
-
-## Prompt
-
-Fix all findings from the local PR94 review, prefer a Tractor pin
-change if it reduces work, and export the original review for Tuicr.
-
-## Generated changes
-
-> `git diff HEAD~1..HEAD -- config/conf.toml`
-
-Use TCP-only defaults, retaining commented UDS alternatives. Both
-checked Tractor address branch tips still reject mixed transports.
-
-> `git diff HEAD~1..HEAD -- piker/cli/__init__.py`
-
-Make Click multiaddresses repeatable and normalize single strings.
-
-> `git diff HEAD~1..HEAD -- piker/service/_actor_runtime.py`
-
-Validate requested daemon bind addresses instead of remote registrars.
-Normalize accepted endpoints and allow TCP ephemeral port requests.
-
-> `git diff HEAD~1..HEAD -- piker/service/_daemon.py`
-
-Resolve explicit or inherited registrars independently of other daemon
-options and consume the explicit keyword before forwarding kwargs.
-
-> `git diff HEAD~1..HEAD -- piker/ui/cli.py`
-
-Default chart bind endpoints to None when network config is absent.
-
-> `git diff HEAD~1..HEAD -- piker/brokers/binance/feed.py`
-
-Remove the unconditional market-info debugger pause.
-
-> `git diff HEAD~1..HEAD -- tests/test_services.py tests/test_startup.py`
-
-Remove the prematurely cherry-picked datad test and add offline startup
-regressions, including a real registrar/child daemon lifecycle check.
-
-## Verification
-
- Startup and logging tests: 13 passed.
- Existing runtime boot and tagged registrar tests: 2 passed.
- Scoped native Ruff and diff checks pass.
- The first new actor test used a removed nursery API; it was corrected
-  to the installed start_actor/Portal.run interface before passing.
- Import initially failed without native libraries. Tests use the
-  existing flake's GCC, zlib, Qt and PyQt paths without executing its
-  environment-rebuilding shell hook.
- Original seven findings exported locally as Tuicr review-add payloads
-  bound to bde2457b; no session comments or remote PR were changed.
--- a/ai/prompt-io/20261010T232852Z_1fa45dfa_prompt_io.md
+++ b/ai/prompt-io/20261010T232852Z_1fa45dfa_prompt_io.md
@ -1,45 +0,0 @@
---
-created_at: 2026-10-10T23:28:52Z
-generated_by:
-  harness: opencode
-  provider: openai
-  model: openai/gpt-6.1-sol
-git_ref: 1fa45dfa
-scope: config
-substantive: true
-raw_file: 20261010T232852Z_1fa45dfa_prompt_io.raw.md
---
-
-## Prompt
-
-The human rejected keeping the April Tractor pin as a landing baseline.
-They requested the reviewed `wkt/addr_unpacking` branch, removal of the
-new rollback adaptations, reuse of original Piker histories where
-possible, and a candidate prepared within their gateway relocation
-window. They asked specifically whether the runtime validation review
-hunk duplicated older endpoint history.
-
-## Response summary
-
-Select the reviewed address branch at an immutable locked revision and
-preserve all committed application/test compatibility code. Verify the
-minimal dependency boundary offline and preview existing consumer
-fixes without importing unrelated datad/backfill changes. Distinguish
-original-source test results from live teardown acceptance, which is
-not available in this preparation window.
-
-## Files changed
-
- `pyproject.toml`: select Tractor `wkt/addr_unpacking`.
- `uv.lock`: resolve reviewed `a966928d` and setproctitle 1.3.8 while
-  preserving Polars 1.36.0 and the rest of the locked graph.
-
-## Human edits
-
-The human redirected the task from an older known-good runtime to a
-modern landing stack and insisted on reusing original history rather
-than building new compatibility adapters. They proposed stricter
-exception-group behavior as a hypothesis and set the live-test/time
-constraints. Those directions materially changed the dependency
-boundary. No direct manual edits to this final boundary were observed;
-runtime acceptance and landing remain human-owned.
--- a/ai/prompt-io/20261010T232852Z_1fa45dfa_prompt_io.raw.md
+++ b/ai/prompt-io/20261010T232852Z_1fa45dfa_prompt_io.raw.md
@ -1,52 +0,0 @@
---
-created_at: 2026-10-10T23:28:52Z
-generated_by:
-  harness: opencode
-  provider: openai
-  model: openai/gpt-6.1-sol
-git_ref: 1fa45dfa
-diff_cmd: git diff HEAD~1..HEAD
---
-
-## Prompt
-
-Undo the April Tractor rollback and select the reviewed
-`wkt/addr_unpacking` branch. Prefer original downstream compatibility
-commits to new adaptations, and prepare an offline-verified candidate
-while live IB verification is deferred for gateway relocation.
-
-## Generated changes
-
-> `git diff HEAD~1..HEAD -- pyproject.toml uv.lock`
-
-Change Tractor's source from `piker_pin` to `wkt/addr_unpacking` and
-lock it at `a966928dead4cbf369ad244834ce9dbe988ac8ca`. Restore application
-and test sources byte-for-byte to the committed PR94 tree. Keep the
-rejected rollback journal only in ignored local experiment reports.
-Preserve Polars 1.36.0 by retaining its exact existing lock blocks;
-setproctitle re-resolves from 1.3.7 to 1.3.8. `uv sync --locked` accepts
-the resulting graph. Actual imports confirm the reviewed Tractor pin
-and Trio 0.32.0, not the hypothesized Trio 0.33.
-
-## Original-history previews
-
-The existing MethodProxy stack is retained. Read-only merge previews
-show original `4af885d9` and `b691422b` can be applied without the datad
-cutover. Original-source regressions run through temporary in-memory
-overlays against the real pinned runtime: seven cases pass and one
-cache-hit case fails because `log.at_least_level('info')` looks up INFO
-in a custom-level mapping. A separate standard logging correction
-produces eight passing cases. Its patch and execution handoff are saved
-under ignored `.ai/code-review/reports/pr94_modern_stack/`; no consumer
-cherry-picks or source edits are performed by this boundary.
-
-## Verification
-
- Startup and logging: 13 passed.
- Runtime boot and tagged registrar normalization: 2 passed.
- Locked sync and source whitespace checks pass.
- Original application/test trees are unchanged from PR94 HEAD.
- Consumer overlay with explicit logging correction: 8 passed.
- Correction patch validates against its exact preview tree using a
-  private temporary index; the real index is untouched.
- Live IB startup, SIGINT and surviving-pikerd reattachment are pending.
--- a/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md
+++ b/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md
@ -1,36 +0,0 @@
---
-model: openai/gpt-5.6-sol
-service: opencode
-session: ses_0799212ebffe42arY96czXn89F
-timestamp: 2026-08-11T20:58:26Z
-git_ref: 06d5ea5d
-scope: code
-substantive: true
-raw_file: 20260811T205826Z_06d5ea5d_prompt_io.raw.md
---
-
-## Prompt
-
-Diagnose and fix an IB live-order failure where lazy NVDA contract
-qualification raised `tractor.trionics.Lagged` after the order reached
-`brokerd.ib`. Determine whether recent history changes or cancelled symbol
-searches caused the stale state.
-
-## Response summary
-
-Traced the failure to competing consumers of a one-slot Trio broadcast, not
-the recent SHM change or stale asyncio client state. Replaced broadcast
-fan-out with one response/status demultiplexer, retained request-ID
-correlation, handled cancellation and teardown, documented the task/channel
-ownership model, warned on diagnostic metadata eviction, and added
-deterministic proxy regressions.
-
-## Files changed
-
- `piker/brokers/ib/api.py` - sole-reader IB proxy response routing.
- `tests/test_ib_method_proxy.py` - status, concurrency, cancellation and EOF regressions.
- `.claude/skills/run-tests/test-harness-reference.md` - proxy test mapping.
-
-## Human edits
-
-None - generated output follows the user's live failure diagnosis.
--- a/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.raw.md
+++ b/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.raw.md
@ -1,66 +0,0 @@
---
-model: openai/gpt-5.6-sol
-service: opencode
-timestamp: 2026-08-11T20:58:26Z
-git_ref: 06d5ea5d
-diff_cmd: git diff HEAD~1..HEAD
---
-
-The user reported that certain IB order entries might have broken and
-provided a live failure for a first NVDA sell order. EMS delivered the
-request to `brokerd.ib`, but lazy `get_mkt_info('nvda.nasdaq')` contract
-qualification failed in `MethodProxy._run_method()` with:
-
-```text
-tractor.trionics._broadcast.Lagged:
-Task `piker.brokers.ib.broker.handle_order_requests` overrun and
-dropped `0` values
-```
-
-The user suspected a cancelled symbol-search task might have left stale
-asyncio state. Cancellation can leave an orphaned method response, but the
-observed failure was stale Trio broadcast state. `open_client_proxy()`
-broadcast a one-slot `LinkedTaskChannel` to an always-running IB status relay
-and an idle method receiver. Status traffic advanced only the relay, so the
-first method call for an uncached symbol entered an already-lagged receiver.
-Orders for eagerly cached contracts bypassed this path.
-
-The immediately preceding sparse-history commit did not modify IB order,
-symbol, or proxy code; it exposed no direct dependency on this failure.
-
-> `git diff HEAD~1..HEAD -- piker/brokers/ib/api.py`
-
-Generated a sole-reader message demultiplexer for each IB client proxy.
-`relay_client_proxy_messages()` owns every asyncio-to-Trio receive, handles
-status/error traffic, and dispatches `mid`-tagged responses into per-call
-`MethodProxy` wait slots. Cancelled calls remove their pending slot, causing
-late responses to be logged and dropped without skewing later calls. Clean
-channel EOF is accepted during normal proxy teardown.
-
-Follow-up review restored the masked reconnect TODO and per-message case
-rationale, documented the one-reader/many-writer task hierarchy and separate
-per-proxy channels, and added a warning with request method metadata when the
-bounded unresolved-request diagnostic table evicts its oldest entry.
-
-> `git diff HEAD~1..HEAD -- tests/test_ib_method_proxy.py`
-
-Generated deterministic regressions for idle status traffic before first
-qualification, two concurrent calls with reverse-order responses, cancelled
-calls with late responses followed by a successful call, and graceful
-channel EOF. Every synchronization wait is timeout-bounded.
-
-> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
-
-Registered the deterministic IB method-proxy target and source mapping.
-
-Verification output:
-
-```text
-........................                                                 [100%]
-25 passed, 4 warnings in 3.44s
-```
-
-The warnings are existing Tractor/Trio deprecations from the EMS test.
-Adversarial review drove a clean-EOF guard and bounded polling. Final review
-found no actionable issues. Live IB order validation remains pending and
-requires restarting `brokerd.ib`.
--- a/ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.md
+++ b/ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.md
@ -1,34 +0,0 @@
---
-model: openai/gpt-5.6-sol
-service: opencode
-session: unavailable
-timestamp: 2026-08-13T20:24:02Z
-git_ref: 12b0d66e
-scope: code
-substantive: true
-raw_file: 20260813T202402Z_12b0d66e_prompt_io.raw.md
---
-
-## Prompt
-
-Continue the paired piker broadcast-consumer work after Tractor PR 490
-stabilization. Stop cached feed callers from mutating or consuming shared
-quote-stream roots, then review, verify and prepare a commit boundary.
-
-## Response summary
-
-Made provider root streams private to the cached `Feed`, gave every caller a
-local subscribed view, corrected feed cache identity, and made multi-provider
-relays own synchronized child subscriptions and independent send-channel
-clones.
-
-## Files changed
-
- `piker/data/feed.py` - caller-local feed views, complete cache keys, and
-  subscribed multi-provider relays with independent channel lifetimes.
- `tests/test_feed_broadcast.py` - real broadcast ownership, lifecycle and
-  cache-identity regressions, including sibling survival after provider EOC.
-
-## Human edits
-
-None - generated output follows the requested cached-feed ownership item.
--- a/ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.raw.md
+++ b/ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.raw.md
@ -1,51 +0,0 @@
---
-model: openai/gpt-5.6-sol
-service: opencode
-timestamp: 2026-08-13T20:24:02Z
-git_ref: 12b0d66e
-diff_cmd: git diff HEAD~1..HEAD
---
-
-The user asked to continue the paired piker broadcast-consumer work after
-Tractor PR 490 stabilization. The next item was to stop cached feed callers
-from mutating or consuming shared `Feed`/`Flume` quote-stream roots, then
-review, verify and prepare a commit boundary.
-
-> `git diff HEAD~1..HEAD -- piker/data/feed.py`
-
-Changed `maybe_open_feed()` so the cached `Feed` exclusively owns provider
-root streams and every caller, including the first cache owner, receives a
-caller-local `Feed`/`Flume` view backed by one child subscription per provider.
-The cache identity now includes normalized full symbol shape and stream policy.
-
-Changed `Feed.open_multi_stream()` so every provider relay subscribes instead
-of directly consuming a shared stream, and synchronizes relay startup before
-yielding the merged channel. Each relay owns a cloned send channel so one
-provider EOC cannot terminate its siblings. The module-level relay returns its
-live child receiver through `TaskStatus` for typed retention and introspection.
-Documented `Feed.pause()`/`Feed.resume()` as provider-context-wide controls
-shared by local cache users.
-
-> `git diff HEAD~1..HEAD -- tests/test_feed_broadcast.py`
-
-Added real `Feed`, `Flume`, and Tractor broadcast regressions covering both
-cache-miss owner and cache-hit paths, one-provider-many-market mapping,
-multi-provider relay fan-out and startup, duplex controls, lexical child
-closure, sibling survival after provider EOC, cached-root immutability, and
-full cache-key differentiation.
-
-Verification output:
-
-```text
-...                                                                      [100%]
-3 passed in 0.02s
-
-.                                                                        [100%]
-1 passed in 0.01s
-```
-
-The second result is the existing EMS broadcast ownership regression. Ruff,
-E501, compilation and whitespace checks passed. Repeated adversarial review
-found and resolved shared `Feed.streams`, incomplete cache-key, fake-lifecycle,
-relay-startup and first-owner asymmetry issues; final review reported no
-findings.
--- a/ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.md
+++ b/ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.md
@ -1,32 +0,0 @@
---
-model: openai/gpt-5.6-sol
-service: opencode
-session: unavailable
-timestamp: 2026-08-13T23:07:44Z
-git_ref: ddac3553
-scope: code
-substantive: true
-raw_file: 20260813T230744Z_ddac3553_prompt_io.raw.md
---
-
-## Prompt
-
-Continue the piker broadcast-consumer audit after the Deribit ownership commit
-and prepare the next smallest ownership fix.
-
-## Response summary
-
-Replaced IB's manual raw quote-channel registry with a symbol-keyed cached
-context and gave every caller a lexical child from one retained
-`LinkedTaskChannel` broadcaster root.
-
-## Files changed
-
- `piker/brokers/ib/feed.py` - cached raw-channel lifetime and lexical child
-  ownership.
- `tests/test_ib_broadcast.py` - concurrent fan-out, cache-key and child
-  lifetime regression.
-
-## Human edits
-
-None - generated output follows the requested broadcast-consumer audit.
--- a/ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.raw.md
+++ b/ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.raw.md
@ -1,39 +0,0 @@
---
-model: openai/gpt-5.6-sol
-service: opencode
-timestamp: 2026-08-13T23:07:44Z
-git_ref: ddac3553
-diff_cmd: git diff HEAD~1..HEAD
---
-
-The user asked to continue the piker broadcast-consumer audit after committing
-the Deribit ownership fix. The next isolated item was IB's actor-local
-`_quote_streams` cache, which retained a raw channel owned by its first caller
-and constructed competing broadcaster roots for later callers.
-
-> `git diff HEAD~1..HEAD -- piker/brokers/ib/feed.py`
-
-Factored `_open_aio_quote_channel()` as the raw asyncio-channel context and
-made `open_aio_quote_stream()` cache that context by symbol through
-`maybe_open_context()`. Every caller now receives a lexical child from the
-retained `LinkedTaskChannel` broadcaster, while the raw source remains alive
-until its final cache user exits. Removed the manual `_quote_streams` registry
-and teardown mutation.
-
-> `git diff HEAD~1..HEAD -- tests/test_ib_broadcast.py`
-
-Added a concurrent ownership regression using real Tractor broadcast state.
-It proves miss and hit callers use the same symbol/context factory, receive the
-same quote through distinct children, close independently, and leave the owner
-live after the hit exits.
-
-Verification output:
-
-```text
-.                                                                        [100%]
-1 passed in 0.01s
-```
-
-The new test passes through a source-isolated IB feed import. Test Ruff, E501,
-Python compilation and whitespace checks pass. Normal package collection
-remains blocked by the existing Qt environment mismatch.
--- a/config/conf.toml
+++ b/config/conf.toml
@ -1,17 +1,14 @@
 [network]
-# This Tractor pin supports one transport per actor, not mixed TCP/UDS.
-# Use the single TCP endpoint shown in each list for this version.
-# Mixed-transport configs need a future Tractor pin with that support.
 pikerd = [
  # TCP localhost loopback
  '/ip4/127.0.0.1/tcp/6116',

-  # UDS alternative: replace TCP and select the UDS transport.
-  # '/unix/run/user/1000/piker/pikerd.sock',
+  # same but UDS
+  '/unix/run/user/1000/piker/pikerd.sock',
 ]
 chart = [
  '/ip4/127.0.0.1/tcp/3003',
-  # '/unix/run/user/1000/piker/chart.sock',
+  '/unix/run/user/1000/piker/chart.sock',
 ]
 # the service-actor registry endpoint;
 # other pikerd trees contact this to discover
--- a/piker/brokers/binance/feed.py
+++ b/piker/brokers/binance/feed.py
@ -364,6 +364,8 @@ async def get_mkt_info(
        if not mkt_mode:
            mkt_mode: str = f'{venue_lower}_futes'

+    await tractor.pause()
+
    async with open_cached_client(
        'binance',
    ) as client:
--- a/piker/brokers/ib/api.py
+++ b/piker/brokers/ib/api.py
@ -1584,72 +1584,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()
-        self._pending: dict[
-            int,
-            tuple[trio.Event, list[dict]],
-        ] = {}
-        self._request_methods: dict[int, str|None] = {}
-
-    def _deliver_method_response(
-        self,
-        msg: dict,
-
-    ) -> bool:
-        '''
-        Deliver one correlated response to its waiting caller.
-
-        '''
-        mid: int = msg['mid']
-        meth: str|None = self._request_methods.pop(mid, None)
-        pending = self._pending.get(mid)
-        if pending is None:
-            log.warning(
-                f'Dropping stale method-resp,\n'
-                f'mid: {mid}\n'
-                f'meth: {meth!r}\n'
-                f'(caller prolly got cancelled '
-                f'before its resp?)\n'
-            )
-            return False
-
-        event, slot = pending
-        slot.append(msg)
-        event.set()
-        return True
-
-    def _track_request_method(
-        self,
-        mid: int,
-        meth: str|None,
-
-    ) -> None:
-        '''
-        Retain bounded request metadata for stale-response logs.
-
-        '''
-        self._request_methods[mid] = meth
-        if len(self._request_methods) > 256:
-            oldest_mid: int = next(iter(self._request_methods))
-            oldest_meth: str|None = self._request_methods.pop(
-                oldest_mid
-            )
-            log.warning(
-                f'Evicting unresolved method-request metadata,\n'
-                f'mid: {oldest_mid}\n'
-                f'meth: {oldest_meth!r}\n'
-                f'pending metadata limit: 256\n'
-            )
-
    async def _run_method(
        self,
        *,
@ -1662,30 +1596,38 @@ class MethodProxy:
        ``tractor.to_asyncio`` layer.

        '''
-        mid: int = next(self._mids)
-        event = trio.Event()
-        slot: list[dict] = []
-        self._pending[mid] = event, slot
-        self._track_request_method(mid, meth)
-        try:
-            await self.chan.send((meth, kwargs, mid))
-            await event.wait()
-        finally:
-            self._pending.pop(mid, None)
+        chan = self.chan
+        await chan.send((meth, kwargs))

-        msg: dict = slot[0]
-        match msg:
-            # OUR method-call response B)
-            case {'result': res}:
+        while not chan.closed():
+            # send through method + ``kwargs: dict`` as pair
+            msg = await chan.receive()
+
+            # TODO: implement reconnect functionality like
+            # in our `.data._web_bs.NoBsWs`
+            # try:
+            #     msg = await chan.receive()
+            # except ConnectionError:
+            #     self.reset()
+
+            # print(f'NEXT MSG: {msg}')
+
+            # TODO: py3.10 ``match:`` syntax B)
+            if 'result' in msg:
+                res = msg.get('result')
                return res

-            case {'exception': err}:
+            elif 'exception' in msg:
+                err = msg.get('exception')
                raise err

-            case _:
-                raise RuntimeError(
-                    f'Invalid method response for mid={mid}: {msg!r}'
-                )
+            elif 'error' in msg:
+                etype, emsg = msg
+                log.warning(f'IB error relay: {emsg}')
+                continue
+
+            else:
+                log.warning(f'UNKNOWN IB MSG: {msg}')

    def status_event(
        self,
@ -1738,31 +1680,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
-                    # req-id for caller-side correlation.
-                    chan.send_nowait({
-                        'mid': mid,
-                        'result': resp,
-                    })
+                    # echo the msg back
+                    chan.send_nowait({'result': resp})

-                # XXX: relay ALL (non-cancel) errors to the
-                # `trio`-side caller (which re-raises in the
-                # `MethodProxy._run_method()` frame) instead
-                # of crashing this relay task and thus the
-                # whole proxy/dialog; critical post the
-                # (datad|brokerd)-split where eg. contract
-                # qualification failures are expected to be
-                # caught per-request by order/warmup code!
-                except Exception as err:
-                    chan.send_nowait({
-                        'mid': mid,
-                        'exception': err,
-                    })
+                except (
+                    RequestError,
+
+                    # TODO: relay all errors to trio?
+                    # BaseException,
+                ) as err:
+                    chan.send_nowait({'exception': err})

            case {'error': content}:
                chan.send_nowait({'exception': content})
@ -1771,73 +1704,6 @@ async def open_aio_client_method_relay(
                raise ValueError(f'Unhandled msg {msg}')


-async def relay_client_proxy_messages(
-    chan: tractor.to_asyncio.LinkedTaskChannel,
-    proxy: MethodProxy,
-
-) -> None:
-    '''
-    Route one proxy's asyncio-client msgs from a single reader.
-
-    Each `open_client_proxy()` allocates its own linked channel and
-    starts this task in its relay nursery. Any number of Trio caller
-    tasks may concurrently write tagged requests through
-    `MethodProxy._run_method()` and wait on their per-`mid` slots.
-    This task alone reads the channel and dispatches each response to
-    its waiting caller.
-
-    Separate proxy contexts own separate channels and reader tasks;
-    no cross-proxy response broadcast is intended. IB status events
-    still reach each proxy through its own `Client.inline_errors()`
-    handler.
-
-    '''
-    try:
-        while not chan.closed():
-            # TODO: implement reconnect functionality like
-            # in our `.data._web_bs.NoBsWs`
-            # try:
-            msg = await chan.receive()
-            # except ConnectionError:
-            #     proxy.reset()
-
-            match msg:
-                # Correlated response from a method call.
-                case {'mid': _}:
-                    proxy._deliver_method_response(msg)
-
-                # routine (api-farm conn) status events
-                # relayed inline by `Client.inline_errors()`;
-                # not a response to a method call so just
-                # log at info and keep waiting.
-                case ('event', status_msg):
-                    reason = status_msg['reason']
-                    event = proxy.event_table.pop(reason, None)
-                    if (
-                        event
-                        and
-                        event.statistics().tasks_waiting
-                    ):
-                        log.info(f'Relaying ib status message: {msg}')
-                        event.set()
-
-                # Inline API error with no request correlation.
-                case ('error', emsg):
-                    log.warning(f'IB error relay: {emsg}')
-
-                # Out-of-band client error with no method `mid`.
-                case {'exception': err}:
-                    log.error(
-                        f'Uncorrelated IB client error: {err!r}'
-                    )
-
-                # Preserve visibility into unexpected relay traffic.
-                case _:
-                    log.warning(f'UNKNOWN IB MSG: {msg}')
-    except trio.EndOfChannel:
-        log.info('IB client proxy relay closed')
-
-
@acm
 async def open_client_proxy(
    client: Client,
@ -1874,11 +1740,28 @@ async def open_client_proxy(
                continue
            setattr(proxy, name, partial(proxy._run_method, meth=name))

-        relay_tn.start_soon(
-            relay_client_proxy_messages,
-            chan,
-            proxy,
-        )
+        async def relay_events():
+
+            async with chan.subscribe() as msg_stream:
+
+                async for msg in msg_stream:
+                    if 'event' not in msg:
+                        continue
+
+                    # if 'event' in msg:
+                    # wake up any system event waiters.
+                    etype, status_msg = msg
+                    reason = status_msg['reason']
+
+                    ev = proxy.event_table.pop(reason, None)
+
+                    if ev and ev.statistics().tasks_waiting:
+                        log.info(f'Relaying ib status message: {msg}')
+                        ev.set()
+
+                    continue
+
+        relay_tn.start_soon(relay_events)

        yield proxy

--- a/piker/brokers/ib/feed.py
+++ b/piker/brokers/ib/feed.py
@ -738,6 +738,10 @@ async def get_bars(
    )


+# per-actor cache of inter-eventloop-chans
+_quote_streams: dict[str, trio.abc.ReceiveStream] = {}
+
+
 async def _setup_quote_stream(
    chan: tractor.to_asyncio.LinkedTaskChannel,
    symbol: str,
@ -766,6 +770,8 @@ async def _setup_quote_stream(
    and is thus run via `tractor.to_asyncio.open_channel_from()`.

    '''
+    global _quote_streams
+
    async with load_aio_clients(
        disconnect_on_exit=False,
    ) as accts2clients:
@ -828,6 +834,9 @@ async def _setup_quote_stream(

            client.ib.cancelMktData(contract)

+            # decouple broadcast mem chan
+            _quote_streams.pop(symbol, None)
+
        def push(
            t: Ticker,
            tries_before_raise: int = 6,
@ -944,52 +953,54 @@ async def _setup_quote_stream(


@acm
-async def _open_aio_quote_channel(
+async def open_aio_quote_stream(
    symbol: str,
    contract: Contract|None = None,

-) -> tractor.to_asyncio.LinkedTaskChannel:
+) -> (
+    trio.abc.Channel|  # iface
+    tractor.to_asyncio.LinkedTaskChannel  # actually
+):
    '''
-    Open one raw inter-event-loop quote channel for `symbol`.
+    Open a real-time `Ticker` quote stream from an `asyncio.Task`
+    spawned via `tractor.to_asyncio.open_channel_from()`, deliver the
+    inter-event-loop channel to the `trio.Task` caller and cache it
+    globally for re-use.

    '''
-    chan: tractor.to_asyncio.LinkedTaskChannel
-    started_contract: Contract
+    from tractor.trionics import broadcast_receiver
+    global _quote_streams
+
+    from_aio = _quote_streams.get(symbol)
+    if from_aio:
+
+        # if we already have a cached feed deliver a rx side clone
+        # to consumer
+        async with broadcast_receiver(
+            from_aio,
+            2**6,
+        ) as from_aio:
+            yield from_aio
+            return
+
+    from_aio: tractor.to_asyncio.LinkedTaskChannel
    async with tractor.to_asyncio.open_channel_from(
        _setup_quote_stream,
        symbol=symbol,
        contract=contract,

-    ) as (chan, started_contract):
-        assert started_contract
-        yield chan
+    ) as (from_aio, contract):

+        assert contract

-@acm
-async def open_aio_quote_stream(
-    symbol: str,
-    contract: Contract|None = None,
+        # TODO? de-reg on teardown of last consumer task?
+        # -> why aren't we using `.trionics.maybe_open_context()`
+        # here again?? (we are in `open_client_proxies()` tho?)
+        #
+        # cache feed for later consumers
+        _quote_streams[symbol] = from_aio

-) -> tractor.trionics.BroadcastReceiver:
-    '''
-    Open a caller-owned child of the cached quote channel for `symbol`.
-
-    Keep the raw `LinkedTaskChannel` alive until its last cache user
-    exits, and retain exactly one broadcaster root on that channel.
-
-    '''
-    chan: tractor.to_asyncio.LinkedTaskChannel
-    async with tractor.trionics.maybe_open_context(
-        acm_func=_open_aio_quote_channel,
-        kwargs={
-            'symbol': symbol,
-            'contract': contract,
-        },
-        key=symbol,
-    ) as (_, chan):
-        bstream: tractor.trionics.BroadcastReceiver
-        async with chan.subscribe() as bstream:
-            yield bstream
+        yield from_aio


 # TODO: cython/mypyc/numba this!
--- a/piker/cli/init.py
+++ b/piker/cli/init.py
@ -47,7 +47,7 @@ log = get_logger('piker.cli')

 def load_trans_eps(
    network: dict|None = None,
-    maddrs: str|tuple[str, ...]|list[str]|None = None,
+    maddrs: list[str]|None = None,

 ) -> dict[str, list]:
    '''
@ -60,8 +60,6 @@ def load_trans_eps(
        return parse_endpoints(network)

    elif maddrs:
-        if isinstance(maddrs, str):
-            maddrs = [maddrs]
        return parse_endpoints(
            {'pikerd': list(maddrs)}
        )
@ -89,11 +87,11 @@ def load_trans_eps(
@click.option(
    '--maddrs',
    '-m',
-    multiple=True,
+    default=None,
    help='Multiaddrs to bind or contact',
 )
 def pikerd(
-    maddrs: tuple[str, ...],
+    maddrs: list[str]|None,
    loglevel: str,
    tl: bool,
    pdb: bool,
--- a/piker/data/feed.py
+++ b/piker/data/feed.py
@ -30,7 +30,6 @@ from collections import (
    defaultdict,
    abc,
 )
-from copy import copy
 from contextlib import asynccontextmanager as acm
 from functools import partial
 import time
@ -675,36 +674,6 @@ async def open_feed_bus(
                bus.remove_subs(bs_fqme, subs)


-async def relay_to_common_memchan(
-    stream: (
-        tractor.MsgStream
-        |trionics.BroadcastReceiver
-    ),
-    tx: trio.MemorySendChannel[dict[str, Any]],
-    task_status: TaskStatus[
-        trionics.BroadcastReceiver,
-    ] = trio.TASK_STATUS_IGNORED,
-
-) -> None:
-    '''
-    Relay one provider subscription into a common memory channel.
-
-    Deliver the live child receiver through `task_status` so the
-    caller can retain and inspect each provider relay's cursor.
-
-    '''
-    bstream: trionics.BroadcastReceiver
-    async with (
-        stream.subscribe() as bstream,
-        tx,
-    ):
-        task_status.started(bstream)
-
-        msg: dict[str, Any]
-        async for msg in bstream:
-            await tx.send(msg)
-
-
 class Feed(Struct):
    '''
    A per-provider API for client-side consumption from real-time data
@ -754,51 +723,29 @@ class Feed(Struct):
        if len(mods) == 1:
            # just pass the brokerd stream directly if only one provider
            # was detected.
-            stream: (
-                tractor.MsgStream
-                |trionics.BroadcastReceiver
-            ) = self.streams[list(brokers)[0]]
-            bstream: trionics.BroadcastReceiver
+            stream = self.streams[list(brokers)[0]]
            async with stream.subscribe() as bstream:
                yield bstream
                return

        # start multiplexing task tree
-        tx: trio.MemorySendChannel[dict[str, Any]]
-        rx: trio.MemoryReceiveChannel[dict[str, Any]]
        tx, rx = trio.open_memory_channel(616)

+        async def relay_to_common_memchan(stream: tractor.MsgStream):
+            async with tx:
+                async for msg in stream:
+                    await tx.send(msg)
+
        async with (
            tractor.trionics.collapse_eg(),
            trio.open_nursery() as nurse
        ):
            # spawn a relay task for each stream so that they all
            # multiplex to a common channel.
-            broker2bstreams: dict[
-                str,
-                trionics.BroadcastReceiver,
-            ] = {}
-            broker: str
-            for broker in mods:
-                stream: (
-                    tractor.MsgStream
-                    |trionics.BroadcastReceiver
-                ) = self.streams[broker]
+            for brokername in mods:
+                stream = self.streams[brokername]
+                nurse.start_soon(relay_to_common_memchan, stream)

-                relay_tx: trio.MemorySendChannel[
-                    dict[str, Any]
-                ] = tx.clone()
-
-                bstream: trionics.BroadcastReceiver = await nurse.start(
-                    relay_to_common_memchan,
-                    stream,
-                    relay_tx,
-                )
-                broker2bstreams[broker] = bstream
-
-            # Each relay now owns one `tx` clone. Closing the original
-            # lets `rx` reach EOC only after every provider relay exits.
-            await tx.aclose()
            try:
                yield rx
            finally:
@ -807,23 +754,10 @@ class Feed(Struct):
    _max_sample_rate: int = 1

    async def pause(self) -> None:
-        '''
-        Pause the shared remote provider context for this cached feed.
-
-        This controls provider publication for every local cache user;
-        it does not pause only this caller's receive cursor.
-
-        '''
-        stream: tractor.MsgStream|trionics.BroadcastReceiver
        for stream in set(self.streams.values()):
            await stream.send('pause')

    async def resume(self) -> None:
-        '''
-        Resume the shared remote provider context for this cached feed.
-
-        '''
-        stream: tractor.MsgStream|trionics.BroadcastReceiver
        for stream in set(self.streams.values()):
            await stream.send('resume')

@ -883,17 +817,7 @@ async def maybe_open_feed(
    in a tractor broadcast receiver.

    '''
-    cache_key: tuple[
-        tuple[str, ...],
-        float|None,
-        bool,
-        bool,
-    ] = (
-        tuple(sorted(fqmes)),
-        kwargs.get('tick_throttle'),
-        kwargs.get('allow_overruns', True),
-        kwargs.get('start_stream', True),
-    )
+    fqme = fqmes[0]

    async with trionics.maybe_open_context(
        acm_func=open_feed,
@ -906,73 +830,27 @@ async def maybe_open_feed(
            'allow_overruns': kwargs.get('allow_overruns', True),
            'start_stream': kwargs.get('start_stream', True),
        },
-        key=cache_key,
+        key=fqme,

    ) as (cache_hit, feed):
-        feed: Feed

-        if (
-            cache_hit
-            and
-            log.at_least_level('info')
-        ):
-            log.info(
-                f'Using cached feed for key:\n'
-                f'{cache_key!r}'
-            )
+        if cache_hit:
+            log.info(f'Using cached feed for {fqme}')
+            # add a new broadcast subscription for the quote stream
+            # if this feed is likely already in use

-        # The cached `Feed` exclusively owns provider root streams.
-        # Every caller, including the first cache owner, gets a local
-        # child cursor and copied descriptors for its lexical lifetime.
-        broker_names: tuple[str, ...] = tuple(feed.streams)
-        bstreams: tuple[
-            trionics.BroadcastReceiver,
-            ...,
-        ]
-        async with trionics.gather_contexts(
-            mngrs=[
-                feed.streams[broker].subscribe()
-                for broker in broker_names
-            ]
-        ) as bstreams:
-            broker2bstreams: dict[
-                str,
-                trionics.BroadcastReceiver,
-            ] = dict(zip(
-                broker_names,
-                bstreams,
-                strict=True,
-            ))
-            flumes: dict[str, Flume] = {}
-            fqme: str
-            flume: Flume
-            for fqme, flume in feed.flumes.items():
-                broker: str = flume.mkt.broker
+            async with trionics.gather_contexts(
+                mngrs=[stream.subscribe() for stream in feed.streams.values()]
+            ) as bstreams:
+                for bstream, flume in zip(bstreams, feed.flumes.values()):
+                    # XXX: TODO: horrible hackery that needs fixing..
+                    # i guess we have to create context proxies?
+                    bstream._ctx = flume.stream._ctx
+                    flume.stream = bstream

-                bstream: trionics.BroadcastReceiver = (
-                    broker2bstreams[broker]
-                )
-
-                # XXX: TODO: horrible hackery that needs fixing..
-                # i guess we have to create context proxies?
-                root_stream: tractor.MsgStream = feed.streams[broker]
-
-                ctx: tractor.Context = root_stream.ctx
-                bstream._ctx = ctx
-
-                local_flume: Flume = copy(flume)
-                local_flume.stream = bstream
-                flumes[fqme] = local_flume
-
-            # `feed` is the cache-owned resource and must retain only
-            # provider root streams. A shallow `Feed` copy gives this
-            # caller its own descriptor mappings while intentionally
-            # sharing immutable market metadata, SHM handles and
-            # provider controls for the cache resource's lifetime.
-            local_feed: Feed = copy(feed)
-            local_feed.flumes = flumes
-            local_feed.streams = broker2bstreams
-            yield local_feed
+                yield feed
+        else:
+            yield feed


@acm
--- a/piker/service/_actor_runtime.py
+++ b/piker/service/_actor_runtime.py
@ -214,27 +214,7 @@ async def open_pikerd(
        tractor.trionics.collapse_eg(),
        trio.open_nursery() as service_tn,
    ):
-        # `reg_addrs` locates the registrar used for actor discovery;
-        # `tpt_bind_addrs` requests this daemon's own IPC listeners.
-        # An existing registrar can live elsewhere, so its endpoints
-        # must not be checked against `root_actor.accept_addrs`.
-        bind_addrs = tpt_bind_addrs
-        if bind_addrs is None:
-            # Only a registrar actor owns the discovery endpoints.
-            # A client with no explicit bind request lets Tractor pick
-            # its listener; there is no caller-specified bind to check.
-            bind_addrs = (
-                reg_addrs if root_actor.is_registrar else []
-            )
-
-        # `root_actor.accept_addrs` contains the actual bound endpoints.
-        # Normalize both sides with Tractor's address API so legacy
-        # pairs and tagged transport tuples compare in the same form.
-        accept_addrs = [
-            tractor.discovery._addr.wrap_address(addr).unwrap()
-            for addr in root_actor.accept_addrs
-        ]
-        for addr in bind_addrs:
+        for addr in reg_addrs:
            # normalize to a wrapped `tractor` addr-type;
            # entries may be raw `tuple`s when passed in
            # from (test) client code.
@ -242,16 +222,9 @@ async def open_pikerd(
                addr,
            )
            uaddr: tuple = wladdr.unwrap()
-            bound = uaddr in accept_addrs
-            if wladdr.proto_key == 'tcp' and uaddr[-1] == 0:
-                # TCP port zero asks the OS for an ephemeral listener.
-                # Its actual port differs from the request, but the
-                # tagged transport and host must still match exactly.
-                bound = any(
-                    actual[:-1] == uaddr[:-1]
-                    for actual in accept_addrs
-                )
-            if not bound:
+            if (
+                uaddr not in root_actor.accept_addrs
+            ):
                raise RuntimeError(
                    f'`pikerd` failed to bind on {addr}!\n'
                    'Maybe you have another daemon already running?'
--- a/piker/service/_daemon.py
+++ b/piker/service/_daemon.py
@ -78,19 +78,14 @@ async def maybe_spawn_daemon(
    lock = Services.locks[service_name]
    await lock.acquire()

-    try:
-        # Discovery and `maybe_open_pikerd()` must use the same
-        # registrar list, independently of other runtime options.
-        # Consume an override here: startup forwards it explicitly
-        # below, so leaving it in `pikerd_kwargs` duplicates the keyword.
-        registry_addrs = pikerd_kwargs.pop('registry_addrs', None)
-        if registry_addrs is None:
-            # Use the registrar addresses inherited by this actor.
-            rtvs: dict = tractor.get_runtime_vars()
-            registry_addrs: list[tuple] = list(
-                map(tuple, rtvs['_registry_addrs'])
-            )
+    if not pikerd_kwargs:
+        # XXX NOTE, pin to apprope `tractor` branch!
+        rtvs: dict = tractor.get_runtime_vars()
+        registry_addrs: list[tuple] = list(
+            map(tuple, rtvs['_registry_addrs'])
+        )

+    try:
        async with find_service(
            service_name,
            registry_addrs=registry_addrs,
@ -222,13 +217,11 @@ async def maybe_spawn_daemon(
            lock.statistics().owner is current_task()
        ):
            log.exception(
-                f'Releasing stale lock after crash..?\n'
-                f'\n'
+                f'Releasing stale lock after crash..?'
                f'{err!r}\n'
            )
            lock.release()
-
-        raise
+            raise err


 async def spawn_emsd(
--- a/piker/ui/cli.py
+++ b/piker/ui/cli.py
@ -199,7 +199,6 @@ def chart(
            'registry_addrs',
            [],
        )
-        chart_eps: list|None = None
        network: dict = conf.get('network')
        if network:
            eps: dict = load_trans_eps(
--- a/pyproject.toml
+++ b/pyproject.toml
@ -202,7 +202,7 @@ pyvnc = { git = "https://github.com/regulad/pyvnc.git" }
 # xonsh = { git = 'https://github.com/xonsh/xonsh.git', branch = 'main' }

 # XXX since, we're like, always hacking new shite all-the-time. Bp
-tractor = { git = "https://github.com/goodboy/tractor.git", branch = "wkt/addr_unpacking" }
+tractor = { git = "https://github.com/goodboy/tractor.git", branch = "piker_pin" }
 # tractor = { git = "https://pikers.dev/goodboy/tractor", branch = "piker_pin" }
 # ------ goodboy ------
 # hackin dev-envs, usually there's something new he's hackin in..
--- a/tests/test_feed_broadcast.py
+++ b/tests/test_feed_broadcast.py
@ -1,347 +0,0 @@
-'''
-Cached feed broadcast ownership regressions.
-
-'''
-from contextlib import asynccontextmanager as acm
-from types import SimpleNamespace
-
-import tractor
-import trio
-import pytest
-
-from piker.data import feed as feed_mod
-from piker.data.flows import Flume
-
-
-class FakeMsgStream:
-    '''
-    Duplex test stream backed by a real Tractor broadcaster.
-
-    '''
-    def __init__(self, broker: str) -> None:
-        self.broker: str = broker
-        self._ctx: object = object()
-        self._tx, rx = trio.open_memory_channel(8)
-        self._broadcaster = tractor.trionics.broadcast_receiver(
-            rx,
-            8,
-        )
-        self.children: list[
-            tractor.trionics.BroadcastReceiver,
-        ] = []
-        self.controls: list[str] = []
-
-    @property
-    def ctx(self) -> object:
-        '''
-        Expose the public context ref provided by `MsgStream.ctx`.
-
-        '''
-        return self._ctx
-
-    @acm
-    async def subscribe(self):
-        '''
-        Yield a real child receiver with duplex send delegation.
-
-        '''
-        child: tractor.trionics.BroadcastReceiver
-        async with self._broadcaster.subscribe() as child:
-            child.send = self.send
-            self.children.append(child)
-            yield child
-
-    async def send(self, msg: str) -> None:
-        '''
-        Record a pause or resume control sent through a child.
-
-        '''
-        self.controls.append(msg)
-
-    async def push(self, msg: dict) -> None:
-        '''
-        Send one provider quote into the broadcast source.
-
-        '''
-        await self._tx.send(msg)
-
-    async def close_source(self) -> None:
-        '''
-        End this provider without closing sibling relay channels.
-
-        '''
-        await self._tx.aclose()
-
-
-def make_feed() -> tuple[
-    feed_mod.Feed,
-    dict[str, FakeMsgStream],
-]:
-    '''
-    Build a real `Feed` and `Flume` graph for two providers.
-
-    '''
-    roots: dict[str, FakeMsgStream] = {
-        broker: FakeMsgStream(broker)
-        for broker in ('ib', 'deribit')
-    }
-    flumes: dict[str, Flume] = {
-        fqme: Flume(
-            mkt=SimpleNamespace(broker=broker),
-            first_quote={},
-            _rt_shm_token=SimpleNamespace(),
-            stream=roots[broker],
-        )
-        for fqme, broker in (
-            ('nvda.nasdaq.ib', 'ib'),
-            ('aapl.nasdaq.ib', 'ib'),
-            ('btc-usd.deribit', 'deribit'),
-        )
-    }
-    return (
-        feed_mod.Feed(
-            mods={
-                broker: SimpleNamespace(name=broker)
-                for broker in roots
-            },
-            portals={},
-            flumes=flumes,
-            streams=roots,
-            status={},
-        ),
-        roots,
-    )
-
-
-@pytest.mark.parametrize('cache_hit', [False, True])
-def test_cached_feed_returns_local_streams(
-    monkeypatch,
-    cache_hit: bool,
-) -> None:
-    '''
-    A cached feed caller must not consume or replace shared streams.
-
-    `maybe_open_feed()` previously paired provider subscriptions with
-    `Feed.flumes.values()` using `zip()` and installed each child on the
-    cached `Flume`. One caller therefore exposed its lexical receiver
-    to every holder of the shared `Feed`; after exit those descriptors
-    pointed at closed children. The positional pairing also dropped a
-    market whenever one provider supplied multiple flumes.
-
-    Model two real provider broadcasters with two IB markets and one
-    Deribit market. Enter both first-owner and cache-hit paths, then
-    prove the yielded `Feed`, its `Feed.streams` mapping and all
-    `Flume`s are caller-local views. Exercise provider-context-wide
-    pause/resume and `Feed.open_multi_stream()` to show duplex sends
-    still reach the roots while each relay owns another child. Finally
-    prove every lexical child closes and all cached descriptors still
-    point at live roots.
-
-    '''
-    cached_feed: feed_mod.Feed
-    roots: dict[str, FakeMsgStream]
-    cached_feed, roots = make_feed()
-    cached_flumes: dict[str, Flume] = cached_feed.flumes
-
-    @acm
-    async def maybe_open_context(**kwargs):
-        '''
-        Return the controlled shared feed as a cache hit.
-
-        '''
-        yield cache_hit, cached_feed
-
-    monkeypatch.setattr(
-        feed_mod.trionics,
-        'maybe_open_context',
-        maybe_open_context,
-    )
-
-    local_feed: feed_mod.Feed|None = None
-    local_streams: dict[
-        str,
-        tractor.trionics.BroadcastReceiver,
-    ]|None = None
-
-    async def main() -> None:
-        nonlocal local_feed, local_streams
-        feed: feed_mod.Feed
-        async with feed_mod.maybe_open_feed(
-            list(cached_flumes),
-        ) as feed:
-            local_feed = feed
-            local_streams = feed.streams
-            assert feed is not cached_feed
-            assert feed.streams is not roots
-            assert feed.flumes is not cached_flumes
-
-            broker: str
-            stream: tractor.trionics.BroadcastReceiver
-            for broker, stream in feed.streams.items():
-                assert stream is roots[broker].children[0]
-                assert stream._ctx is roots[broker].ctx
-
-            fqme: str
-            flume: Flume
-            for fqme, flume in feed.flumes.items():
-                cached: Flume = cached_flumes[fqme]
-
-                stream: tractor.trionics.BroadcastReceiver = (
-                    feed.streams[flume.mkt.broker]
-                )
-
-                assert flume is not cached
-                assert flume.stream is stream
-                assert cached.stream is roots[flume.mkt.broker]
-
-            assert (
-                feed.flumes['nvda.nasdaq.ib'].stream
-                is feed.flumes['aapl.nasdaq.ib'].stream
-            )
-
-            await feed.pause()
-            await feed.resume()
-            assert all(
-                root.controls == ['pause', 'resume']
-                for root in roots.values()
-            )
-
-            stream: trio.MemoryReceiveChannel
-            async with feed.open_multi_stream() as stream:
-                assert all(
-                    len(child._state.subs) == 3
-                    for child in feed.streams.values()
-                )
-                await roots['ib'].push({'provider': 'ib'})
-                await roots['deribit'].push({
-                    'provider': 'deribit',
-                })
-                with trio.fail_after(1):
-                    msgs = {
-                        (await stream.receive())['provider'],
-                        (await stream.receive())['provider'],
-                    }
-                assert msgs == {'ib', 'deribit'}
-
-                # One provider's EOC must close only its `tx` clone;
-                # the sibling relay continues publishing to `stream`.
-                await roots['ib'].close_source()
-                with trio.fail_after(1):
-                    while (
-                        len(feed.streams['ib']._state.subs)
-                        != 2
-                    ):
-                        await trio.lowlevel.checkpoint()
-
-                    await roots['deribit'].push({
-                        'provider': 'deribit-after-ib-eoc',
-                    })
-                    msg: dict
-                    msg = await stream.receive()
-
-                assert msg['provider'] == 'deribit-after-ib-eoc'
-
-            assert all(
-                len(child._state.subs) == 2
-                for child in feed.streams.values()
-            )
-
-    trio.run(main)
-
-    assert local_feed is not None
-    assert local_streams is not None
-    assert all(stream._closed for stream in local_streams.values())
-    assert all(
-        flume.stream is roots[flume.mkt.broker]
-        for flume in cached_flumes.values()
-    )
-    assert all(
-        len(root._broadcaster._state.subs) == 1
-        for root in roots.values()
-    )
-
-
-def test_feed_cache_key_includes_shape_and_policy(
-    monkeypatch,
-) -> None:
-    '''
-    Feed cache identity must include every resource-defining input.
-
-    The old `fqmes[0]` key aliased `[A]` with `[A, B]` and ignored
-    `tick_throttle`, `allow_overruns` and `start_stream`. Whichever
-    caller entered first silently determined later feed shape and
-    stream behavior, including reuse of a history-only remote feed for
-    a live quote request.
-
-    Capture the key passed to `maybe_open_context()` for reordered and
-    varied requests. Symbol order must normalize to one key, while the
-    full symbol set and each stream policy must produce distinct keys.
-
-    '''
-    feed: feed_mod.Feed = feed_mod.Feed(
-        mods={},
-        portals={},
-        flumes={},
-        streams={},
-        status={},
-    )
-    keys: list[tuple] = []
-
-    @acm
-    async def maybe_open_context(**kwargs):
-        '''
-        Capture cache identity without opening a remote feed.
-
-        '''
-        keys.append(kwargs['key'])
-        yield False, feed
-
-    @acm
-    async def gather_contexts(mngrs):
-        '''
-        Permit the metadata-only empty feed used by this key probe.
-
-        '''
-        assert not mngrs
-        yield ()
-
-    monkeypatch.setattr(
-        feed_mod.trionics,
-        'maybe_open_context',
-        maybe_open_context,
-    )
-    monkeypatch.setattr(
-        feed_mod.trionics,
-        'gather_contexts',
-        gather_contexts,
-    )
-
-    async def open_once(
-        fqmes: list[str],
-        **kwargs,
-    ) -> None:
-        async with feed_mod.maybe_open_feed(
-            fqmes,
-            **kwargs,
-        ):
-            pass
-
-    async def main() -> None:
-        await open_once(['a.ib', 'b.ib'])
-        await open_once(['b.ib', 'a.ib'])
-        await open_once(['a.ib'])
-        await open_once(['a.ib', 'b.ib'], tick_throttle=10)
-        await open_once(
-            ['a.ib', 'b.ib'],
-            allow_overruns=False,
-        )
-        await open_once(
-            ['a.ib', 'b.ib'],
-            start_stream=False,
-        )
-
-    trio.run(main)
-
-    assert keys[0] == keys[1]
-    assert len(set(keys[2:])) == 4
-    assert keys[0] not in keys[2:]
--- a/tests/test_ib_broadcast.py
+++ b/tests/test_ib_broadcast.py
@ -1,120 +0,0 @@
-'''
-IB cached quote-channel ownership regressions.
-
-'''
-from contextlib import asynccontextmanager as acm
-
-import tractor
-import trio
-
-from piker.brokers.ib import feed
-
-
-def test_quote_stream_caches_channel_and_owns_child(
-    monkeypatch,
-) -> None:
-    '''
-    IB quote callers must share one cached channel through child cursors.
-
-    `open_aio_quote_stream()` previously stored the first caller's raw
-    `LinkedTaskChannel` in `_quote_streams`, yielded that root directly,
-    and wrapped the same source in a new broadcaster for each later
-    caller. The first caller therefore owned source lifetime while
-    competing roots split quote receives between concurrent users.
-
-    Stub `maybe_open_context()` with one retained channel and enter a
-    miss owner plus concurrent hit. Prove both receive the same quote
-    through distinct children of one broadcaster, the hit closes
-    independently, and the owner keeps receiving until its own exit.
-    Also verify both calls request the same symbol cache key.
-
-    '''
-    class FakeLinkedTaskChannel:
-        '''
-        Retain one real broadcaster and expose child subscriptions.
-
-        '''
-        def __init__(self) -> None:
-            self._tx: trio.MemorySendChannel
-            rx: trio.MemoryReceiveChannel
-            self._tx, rx = trio.open_memory_channel(8)
-            self._broadcaster: (
-                tractor.trionics.BroadcastReceiver
-            ) = tractor.trionics.broadcast_receiver(
-                rx,
-                8,
-            )
-
-        @acm
-        async def subscribe(self):
-            '''
-            Yield one caller-owned child from the retained root.
-
-            '''
-            child: tractor.trionics.BroadcastReceiver
-            async with self._broadcaster.subscribe() as child:
-                yield child
-
-        async def push(self, msg: dict) -> None:
-            '''
-            Send one ticker through the retained source channel.
-
-            '''
-            await self._tx.send(msg)
-
-    chan: FakeLinkedTaskChannel = FakeLinkedTaskChannel()
-    context_entries: int = 0
-    keys: list[str] = []
-    acm_funcs: list = []
-
-    @acm
-    async def maybe_open_context(**kwargs):
-        '''
-        Return one retained channel as a miss followed by a hit.
-
-        '''
-        nonlocal context_entries
-        keys.append(kwargs['key'])
-        acm_funcs.append(kwargs['acm_func'])
-        cache_hit: bool = context_entries > 0
-        context_entries += 1
-        yield cache_hit, chan
-
-    monkeypatch.setattr(
-        feed.tractor.trionics,
-        'maybe_open_context',
-        maybe_open_context,
-    )
-
-    async def main() -> None:
-        owner: tractor.trionics.BroadcastReceiver
-        async with feed.open_aio_quote_stream(
-            'NVDA',
-        ) as owner:
-            hit: tractor.trionics.BroadcastReceiver
-            async with feed.open_aio_quote_stream(
-                'NVDA',
-            ) as hit:
-                assert hit is not owner
-                assert len(chan._broadcaster._state.subs) == 3
-
-                await chan.push({'value': 1})
-                assert await owner.receive() == {'value': 1}
-                assert await hit.receive() == {'value': 1}
-
-            assert hit._closed
-            assert not owner._closed
-            assert len(chan._broadcaster._state.subs) == 2
-
-            await chan.push({'value': 2})
-            assert await owner.receive() == {'value': 2}
-
-        assert owner._closed
-        assert len(chan._broadcaster._state.subs) == 1
-        assert keys == ['NVDA', 'NVDA']
-        assert acm_funcs == [
-            feed._open_aio_quote_channel,
-            feed._open_aio_quote_channel,
-        ]
-
-    trio.run(main)
--- a/tests/test_ib_method_proxy.py
+++ b/tests/test_ib_method_proxy.py
@ -1,271 +0,0 @@
-'''
-IB asyncio method-proxy regressions.
-
-'''
-from types import SimpleNamespace
-from typing import Any
-
-import pytest
-import trio
-
-from piker.brokers.ib import api as ib_api
-from piker.brokers.ib.api import (
-    MethodProxy,
-    relay_client_proxy_messages,
-)
-
-
-class FakeChannel:
-    '''
-    Model the Trio side of an asyncio linked-task channel.
-
-    '''
-    def __init__(self) -> None:
-        self.requests: list[tuple] = []
-        (
-            self._tx,
-            self._rx,
-        ) = trio.open_memory_channel[Any](10)
-
-    async def send(self, msg: tuple) -> None:
-        '''
-        Capture one Trio-to-asyncio method request.
-
-        '''
-        self.requests.append(msg)
-
-    async def receive(self) -> Any:
-        '''
-        Receive one simulated asyncio-to-Trio response.
-
-        '''
-        return await self._rx.receive()
-
-    def closed(self) -> bool:
-        '''
-        Report an open channel for the lifetime of each test.
-
-        '''
-        return False
-
-    async def respond(self, msg: Any) -> None:
-        '''
-        Send one response or status event to the proxy relay.
-
-        '''
-        await self._tx.send(msg)
-
-    async def close_responses(self) -> None:
-        '''
-        Close the simulated asyncio-to-Trio response channel.
-
-        '''
-        await self._tx.aclose()
-
-
-def test_method_proxy_routes_after_idle_status_events() -> None:
-    '''
-    Idle IB status traffic must not break first-symbol qualification.
-
-    ``open_client_proxy()`` previously broadcast a one-slot channel
-    to an always-running status relay and an idle method receiver.
-    Status events advanced only the relay, so a first order for an
-    uncached symbol raised ``Lagged`` before receiving
-    ``get_sym_details()``. Feed several idle events through the
-    sole-reader relay, then start two method calls and return their
-    responses in reverse order. Both exact results prove status
-    traffic cannot overrun a dormant caller and concurrent responses
-    remain correlated by request ID.
-
-    '''
-    async def main() -> None:
-        chan = FakeChannel()
-        proxy = MethodProxy(
-            chan,
-            event_table={},
-            asyncio_ns=SimpleNamespace(),
-        )
-        results: dict[str, str] = {}
-
-        async def call(name: str) -> None:
-            results[name] = await proxy._run_method(meth=name)
-
-        async with trio.open_nursery() as nursery:
-            nursery.start_soon(
-                relay_client_proxy_messages,
-                chan,
-                proxy,
-            )
-            for i in range(3):
-                await chan.respond((
-                    'event',
-                    {'reason': f'farm-status-{i}'},
-                ))
-
-            nursery.start_soon(call, 'first')
-            nursery.start_soon(call, 'second')
-            with trio.fail_after(0.1):
-                while len(chan.requests) < 2:
-                    await trio.lowlevel.checkpoint()
-
-            mids: dict[str, int] = {
-                meth: mid
-                for meth, _kwargs, mid in chan.requests
-            }
-            await chan.respond({
-                'mid': mids['second'],
-                'result': 'second-result',
-            })
-            await chan.respond({
-                'mid': mids['first'],
-                'result': 'first-result',
-            })
-            with trio.fail_after(0.1):
-                while len(results) < 2:
-                    await trio.lowlevel.checkpoint()
-            nursery.cancel_scope.cancel()
-
-        assert results == {
-            'first': 'first-result',
-            'second': 'second-result',
-        }
-
-    trio.run(main)
-
-
-def test_method_proxy_relay_accepts_clean_eof() -> None:
-    '''
-    Normal asyncio relay completion must not fail proxy teardown.
-
-    The sole-reader relay blocks in ``LinkedTaskChannel.receive()``
-    until the asyncio task exits and closes its Trio send channel.
-    Close the fake response sender before entering the relay and
-    prove the resulting ``EndOfChannel`` is graceful completion.
-
-    '''
-    async def main() -> None:
-        chan = FakeChannel()
-        proxy = MethodProxy(
-            chan,
-            event_table={},
-            asyncio_ns=SimpleNamespace(),
-        )
-        await chan.close_responses()
-        with trio.fail_after(0.1):
-            await relay_client_proxy_messages(chan, proxy)
-
-    trio.run(main)
-
-
-def test_method_proxy_drops_cancelled_call_response() -> None:
-    '''
-    A cancelled symbol search must not skew the next method response.
-
-    Search timeouts can cancel a Trio caller after its request
-    reaches the asyncio client. Wait until one request is sent,
-    cancel its caller, and then deliver the orphaned response. The
-    sole-reader relay must drop that stale ``mid`` without failing; a
-    subsequent call must still receive its own exact response.
-
-    '''
-    async def main() -> None:
-        chan = FakeChannel()
-        proxy = MethodProxy(
-            chan,
-            event_table={},
-            asyncio_ns=SimpleNamespace(),
-        )
-        call_scope = trio.CancelScope()
-        cancelled = trio.Event()
-        result: list[str] = []
-
-        async def cancelled_call() -> None:
-            with call_scope:
-                try:
-                    await proxy._run_method(meth='cancelled')
-                finally:
-                    cancelled.set()
-
-        async def next_call() -> None:
-            value = await proxy._run_method(meth='next')
-            result.append(value)
-
-        async with trio.open_nursery() as nursery:
-            nursery.start_soon(
-                relay_client_proxy_messages,
-                chan,
-                proxy,
-            )
-            nursery.start_soon(cancelled_call)
-            with trio.fail_after(0.1):
-                while len(chan.requests) < 1:
-                    await trio.lowlevel.checkpoint()
-            cancelled_mid: int = chan.requests[0][2]
-            call_scope.cancel()
-            with trio.fail_after(0.1):
-                await cancelled.wait()
-            assert proxy._pending == {}
-            assert (
-                proxy._request_methods[cancelled_mid]
-                ==
-                'cancelled'
-            )
-
-            await chan.respond({
-                'mid': cancelled_mid,
-                'result': 'stale',
-            })
-            with trio.fail_after(0.1):
-                while cancelled_mid in proxy._request_methods:
-                    await trio.lowlevel.checkpoint()
-            nursery.start_soon(next_call)
-            with trio.fail_after(0.1):
-                while len(chan.requests) < 2:
-                    await trio.lowlevel.checkpoint()
-            next_mid: int = chan.requests[1][2]
-            await chan.respond({
-                'mid': next_mid,
-                'result': 'fresh',
-            })
-            with trio.fail_after(0.1):
-                while not result:
-                    await trio.lowlevel.checkpoint()
-            nursery.cancel_scope.cancel()
-
-        assert result == ['fresh']
-
-    trio.run(main)
-
-
-def test_method_proxy_warns_when_metadata_is_evicted(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    '''
-    Unresolved metadata eviction must remain operator-visible.
-
-    Cancelled calls whose asyncio responses never arrive retain a
-    bounded method name solely for later stale-response diagnostics.
-    Fill that table beyond its limit and prove the oldest entry is
-    evicted with its request ID, method name, and configured bound in
-    the warning.
-
-    '''
-    proxy = MethodProxy(
-        FakeChannel(),
-        event_table={},
-        asyncio_ns=SimpleNamespace(),
-    )
-    warnings: list[str] = []
-    monkeypatch.setattr(ib_api.log, 'warning', warnings.append)
-
-    for mid in range(257):
-        proxy._track_request_method(mid, f'method-{mid}')
-
-    assert len(proxy._request_methods) == 256
-    assert 0 not in proxy._request_methods
-    assert warnings == [
-        'Evicting unresolved method-request metadata,\n'
-        'mid: 0\n'
-        "meth: 'method-0'\n"
-        'pending metadata limit: 256\n'
-    ]
--- a/tests/test_services.py
+++ b/tests/test_services.py
@ -120,6 +120,41 @@ def test_registry_normalizes_tagged_addresses(
        _registry.Registry.addrs = []


+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,
--- a/tests/test_startup.py
+++ b/tests/test_startup.py
@ -1,214 +0,0 @@
-'''
-Offline regressions for endpoint configuration and service startup.
-
-'''
-from contextlib import asynccontextmanager as acm
-from pathlib import Path
-import sys
-import tomllib
-from types import ModuleType
-
-from click.testing import CliRunner
-import pytest
-import tractor
-import trio
-
-from piker import config
-from piker.cli import load_trans_eps, pikerd
-from piker.service import _daemon, _actor_runtime
-
-
-@pytest.mark.parametrize('repeat', [False, True])
-def test_maddrs_click_collection(repeat):
-    '''
-    Click previously produced a string for `--maddrs`, which endpoint
-    loading split into individual characters. Parse one or repeated
-    options through Click and then the real Tractor parser, without
-    entering the indefinitely running daemon callback. Every address
-    must survive intact and in the declared order.
-
-    '''
-    addresses = ['/ip4/127.0.0.1/tcp/6116']
-    if repeat:
-        addresses.append('/ip4/127.0.0.1/tcp/6117')
-    args = [
-        arg for address in addresses for arg in ['--maddrs', address]
-    ]
-    with pikerd.make_context('pikerd', args) as ctx:
-        endpoints = load_trans_eps(maddrs=ctx.params['maddrs'])
-    assert [addr.unwrap() for addr in endpoints['pikerd']] == [
-        ('tcp', '127.0.0.1', 6116 + index)
-        for index in range(len(addresses))
-    ]
-
-
-def test_single_string_multiaddr():
-    '''
-    Direct endpoint callers may provide one string, independently of
-    Click's collection conversion. That input must denote a single
-    endpoint rather than a sequence of invalid character endpoints.
-
-    '''
-    endpoints = load_trans_eps(maddrs='/ip4/127.0.0.1/tcp/6116')
-    assert endpoints['pikerd'][0].unwrap() == (
-        'tcp', '127.0.0.1', 6116,
-    )
-
-
-def test_template_single_transport():
-    '''
-    The shipped template enabled TCP and UDS together although the
-    locked Tractor runtime permits only one transport per actor.
-    Parse the real template and ensure both daemon and chart defaults
-    use the TCP transport selected by their startup paths.
-
-    '''
-    path = Path(__file__).parents[1] / 'config' / 'conf.toml'
-    conf = tomllib.loads(path.read_text())
-    endpoints = load_trans_eps(conf['network'])
-    for name in ['pikerd', 'chart']:
-        assert endpoints[name]
-        assert {addr.proto_key for addr in endpoints[name]} == {'tcp'}
-
-
-@pytest.mark.parametrize('network', [None, {}])
-def test_chart_without_network(monkeypatch, network):
-    '''
-    Existing configs without a populated network table left chart
-    bind endpoints unbound, crashing before the UI entrypoint. Invoke
-    the actual Click callback with absent and empty tables, replacing
-    only config IO and the UI entrypoint. Automatic binding must be
-    passed as None while preserving the configured registrar address.
-
-    '''
-    from piker.ui.cli import chart
-
-    conf = {} if network is None else {'network': network}
-    monkeypatch.setattr(config, 'load', lambda **_: (conf, None))
-    app = ModuleType('piker.ui._app')
-    calls = []
-    app._main = lambda **kwargs: calls.append(kwargs)
-    monkeypatch.setitem(sys.modules, 'piker.ui._app', app)
-    result = CliRunner().invoke(
-        chart,
-        ['btcusdt.binance'],
-        obj={
-            'pdb': False,
-            'brokermods': [object()],
-            'tractorloglevel': 'warning',
-            'loglevel': 'warning',
-            'registry_addrs': [('127.0.0.1', 6116)],
-        },
-    )
-    assert result.exit_code == 0, result.exception
-    runtime = calls[0]['tractor_kwargs']
-    assert runtime['tpt_bind_addrs'] is None
-    assert runtime['registry_addrs'] == [('127.0.0.1', 6116)]
-
-
-@pytest.mark.parametrize('explicit_registry', [False, True])
-def test_daemon_options_resolve_registrars(
-    monkeypatch,
-    explicit_registry,
-):
-    '''
-    Any nonempty pikerd kwargs previously skipped registrar address
-    initialization. Exercise the full discovery and spawn path with
-    runtime options, using protocol-boundary fakes instead of remote
-    actors. An explicit override must reach discovery and startup
-    exactly once; otherwise the inherited registrar list is used.
-    The service lock must be released after leaving the context.
-
-    '''
-    inherited = [('tcp', '127.0.0.1', 6116)]
-    override = [('tcp', '127.0.0.1', 6117)]
-    expected = override if explicit_registry else inherited
-    calls = []
-    portal = object()
-
-    @acm
-    async def find(name, registry_addrs):
-        calls.append(('find', registry_addrs))
-        yield None
-
-    @acm
-    async def open_root(**kwargs):
-        calls.append(('open', kwargs))
-        yield None
-
-    @acm
-    async def wait(name):
-        yield portal
-
-    async def start(**kwargs):
-        return True
-
-    monkeypatch.setattr(_daemon, 'find_service', find)
-    monkeypatch.setattr(_daemon, 'maybe_open_pikerd', open_root)
-    monkeypatch.setattr(tractor, 'wait_for_actor', wait)
-    monkeypatch.setattr(
-        tractor,
-        'get_runtime_vars',
-        lambda: {'_registry_addrs': inherited},
-    )
-
-    async def main():
-        lock = trio.Lock()
-        monkeypatch.setitem(_daemon.Services.locks, 'probe', lock)
-        kwargs = {'debug_mode': False}
-        if explicit_registry:
-            kwargs['registry_addrs'] = override
-        async with _daemon.maybe_spawn_daemon(
-            'probe', start, {}, **kwargs,
-        ) as found:
-            assert found is portal
-        assert not lock.locked()
-
-    trio.run(main)
-    assert calls[0] == ('find', expected)
-    assert calls[1][1]['registry_addrs'] == expected
-    assert calls[1][1]['debug_mode'] is False
-
-
-async def _check_distinct_daemon_bind():
-    '''
-    Open Piker's service context on a non-registrar child actor.
-
-    '''
-    actor = tractor.current_actor()
-    assert not actor.is_registrar
-    assert actor.accept_addrs != actor.reg_addrs
-    async with _actor_runtime.open_pikerd(
-        registry_addrs=actor.reg_addrs,
-        tpt_bind_addrs=[('127.0.0.1', 0)],
-    ):
-        return actor.accept_addrs
-
-
-def test_daemon_bind_distinct_from_registrar(reg_addr):
-    '''
-    Separate daemon and registrar endpoints were rejected because
-    startup checked registrar addresses against daemon listeners.
-    Run a real TCP registrar and a distinct child daemon with an
-    ephemeral listener. Its service context must start and unwind
-    without taking ownership of the registrar's listening address.
-
-    '''
-    async def main():
-        async with (
-            tractor.open_root_actor(
-                registry_addrs=[reg_addr],
-                enable_transports=['tcp'],
-            ),
-            tractor.open_nursery() as nursery,
-        ):
-            portal = await nursery.start_actor(
-                'daemon_probe',
-                enable_modules=[__name__],
-                bind_addrs=[('127.0.0.1', 0)],
-            )
-            addresses = await portal.run(_check_distinct_daemon_bind)
-            assert addresses
-            await nursery.cancel()
-
-    trio.run(main)
--- a/uv.lock
+++ b/uv.lock
@ -990,7 +990,7 @@ requires-dist = [
    { name = "tomli", specifier = ">=2.0.1,<3.0.0" },
    { name = "tomli-w", specifier = ">=1.0.0,<2.0.0" },
    { name = "tomlkit", git = "https://github.com/pikers/tomlkit.git?branch=piker_pin" },
-    { name = "tractor", git = "https://github.com/goodboy/tractor.git?branch=wkt%2Faddr_unpacking" },
+    { name = "tractor", git = "https://github.com/goodboy/tractor.git?branch=piker_pin" },
    { name = "trio", specifier = ">=0.27" },
    { name = "trio-typing", specifier = ">=0.10.0" },
    { name = "trio-util", specifier = ">=0.7.0,<0.8.0" },
@ -1560,22 +1560,30 @@ wheels = [

 [[package]]
 name = "setproctitle"
-version = "1.3.8"
+version = "1.3.7"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/49/b0/6b8a516c5a9e9630bd5293db78314ac012f690305fe93beadea388626efb/setproctitle-1.3.8.tar.gz", hash = "sha256:cafe209d064a6efb88cb45a03e97981ff8832802b2b5d009dde0197a3b7b41c8", size = 27214, upload-time = "2026-10-01T21:09:27.121Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/8d/48/49393a96a2eef1ab418b17475fb92b8fcfad83d099e678751b05472e69de/setproctitle-1.3.7.tar.gz", hash = "sha256:bc2bc917691c1537d5b9bca1468437176809c7e11e5694ca79a9ca12345dcb9e", size = 27002, upload-time = "2025-09-05T12:51:25.278Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/58/46/125791eb19c85340c6d43de8c821b143aecbfa0882343f244f5744474229/setproctitle-1.3.8-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:a4a782f9ab26e662a890c1a7d865c452c83ed554de44aa83da35fa5bdcc79921", size = 18100, upload-time = "2026-10-01T20:59:33.41Z" },
-    { url = "https://files.pythonhosted.org/packages/47/51/73b1e62c3e019e683e58d31f9f8af57e6750f36ce079004d29b7790e22ce/setproctitle-1.3.8-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:d63adfa4be8f9dbac4be6b359e2a8eb15a733545ef613d02a12ae27a4d40e975", size = 13116, upload-time = "2026-10-01T20:59:35.159Z" },
-    { url = "https://files.pythonhosted.org/packages/78/ae/095c6ad159e3a7b95bc6d9a3eae18abfea1ebc79ed3377617406a1668150/setproctitle-1.3.8-cp313-cp313-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:d4289e209e805d7d276a534f062092974fbe6814889d4c132c247205ebb49a65", size = 33275, upload-time = "2026-10-01T20:59:37.235Z" },
-    { url = "https://files.pythonhosted.org/packages/f7/6e/5196258efd941cb874b3f4da91460601b867fdc434eba48a70455736d4be/setproctitle-1.3.8-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8bb543b4b58ad839f7f0d335a7f4b40cef6989952e7239de3532de0b7771607e", size = 34124, upload-time = "2026-10-01T20:59:39.529Z" },
-    { url = "https://files.pythonhosted.org/packages/36/01/e9754e0916e6dda232a8c337bda69d2fea33acee0c5d3bdf580b8155a50d/setproctitle-1.3.8-cp313-cp313-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:b705f01a531dba8fbbab90e30d432af4cc70a79c0a33b238e2c061816140cad0", size = 36006, upload-time = "2026-10-01T20:59:41.501Z" },
-    { url = "https://files.pythonhosted.org/packages/ee/a9/c63d2d745f18423937429a8611743e7b3e17259173150fbcdaa59daa4a51/setproctitle-1.3.8-cp313-cp313-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:8e286e4c4e90063072a6e088a8d6526a7ffceb47429db88ba50ba58eeebcc758", size = 32268, upload-time = "2026-10-01T20:59:43.795Z" },
-    { url = "https://files.pythonhosted.org/packages/36/0c/a72818902f32ea9b848db5ccc828eb9cd9bc8661975610e1d46ea4aacbe2/setproctitle-1.3.8-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:7b77c8671974288cf192e481ed63e928f1086762e97e8845d09d65ba121f8aaa", size = 33179, upload-time = "2026-10-01T20:59:45.809Z" },
-    { url = "https://files.pythonhosted.org/packages/bb/55/fa59463ae6b957048e0e842d4b5513eced1b71554d8af0f5d91cd23cde23/setproctitle-1.3.8-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:0e85b55eac9abfa3166ab00dac21e718ac2624a1df7d0b75559a2d3e3781cb33", size = 34472, upload-time = "2026-10-01T20:59:47.858Z" },
-    { url = "https://files.pythonhosted.org/packages/b7/77/1f0cf55055788bf7eb9a146040a64d7ec42cd56153894856effc70e943ec/setproctitle-1.3.8-cp313-cp313-musllinux_1_2_riscv64.whl", hash = "sha256:f215ec9237d5f8e96d0205be6e2141292c20b133343c2153b6d4934f523ef422", size = 31428, upload-time = "2026-10-01T20:59:49.753Z" },
-    { url = "https://files.pythonhosted.org/packages/32/98/d1d8a6cd620d72a04f5d3335125e5b2fc5667bd06a76bb5bbc50b101e804/setproctitle-1.3.8-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:b0e70a5d294e4c8cf7dc94e015fca6d4913779a424937e66b429fba82d8eb5e2", size = 32776, upload-time = "2026-10-01T20:59:51.763Z" },
-    { url = "https://files.pythonhosted.org/packages/4b/34/8b9eea1e80ac7a7f622eead8a0e4d73a60098344ed6f373e111f7ce8d29d/setproctitle-1.3.8-cp313-cp313-win32.whl", hash = "sha256:f85b4d87273e0ec63568433bbb8ccf160b548c49b53576fe010bad9f698b8586", size = 12105, upload-time = "2026-10-01T20:59:53.444Z" },
-    { url = "https://files.pythonhosted.org/packages/bf/f0/98883c658859d331faf91eccac8946cb2f8feb2590905015b129fdeb09cd/setproctitle-1.3.8-cp313-cp313-win_amd64.whl", hash = "sha256:bcdf74fe8a6e3b31bed1bc7d3e2a38cfbda3263d2750492ea0725dbc2db5aacf", size = 12933, upload-time = "2026-10-01T20:59:55.045Z" },
+    { url = "https://files.pythonhosted.org/packages/5d/2f/fcedcade3b307a391b6e17c774c6261a7166aed641aee00ed2aad96c63ce/setproctitle-1.3.7-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:c3736b2a423146b5e62230502e47e08e68282ff3b69bcfe08a322bee73407922", size = 18047, upload-time = "2025-09-05T12:49:50.271Z" },
+    { url = "https://files.pythonhosted.org/packages/23/ae/afc141ca9631350d0a80b8f287aac79a76f26b6af28fd8bf92dae70dc2c5/setproctitle-1.3.7-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:3384e682b158d569e85a51cfbde2afd1ab57ecf93ea6651fe198d0ba451196ee", size = 13073, upload-time = "2025-09-05T12:49:51.46Z" },
+    { url = "https://files.pythonhosted.org/packages/87/ed/0a4f00315bc02510395b95eec3d4aa77c07192ee79f0baae77ea7b9603d8/setproctitle-1.3.7-cp313-cp313-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:0564a936ea687cd24dffcea35903e2a20962aa6ac20e61dd3a207652401492dd", size = 33284, upload-time = "2025-09-05T12:49:52.741Z" },
+    { url = "https://files.pythonhosted.org/packages/fc/e4/adf3c4c0a2173cb7920dc9df710bcc67e9bcdbf377e243b7a962dc31a51a/setproctitle-1.3.7-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a5d1cb3f81531f0eb40e13246b679a1bdb58762b170303463cb06ecc296f26d0", size = 34104, upload-time = "2025-09-05T12:49:54.416Z" },
+    { url = "https://files.pythonhosted.org/packages/52/4f/6daf66394152756664257180439d37047aa9a1cfaa5e4f5ed35e93d1dc06/setproctitle-1.3.7-cp313-cp313-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:a7d159e7345f343b44330cbba9194169b8590cb13dae940da47aa36a72aa9929", size = 35982, upload-time = "2025-09-05T12:49:56.295Z" },
+    { url = "https://files.pythonhosted.org/packages/1b/62/f2c0595403cf915db031f346b0e3b2c0096050e90e0be658a64f44f4278a/setproctitle-1.3.7-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:0b5074649797fd07c72ca1f6bff0406f4a42e1194faac03ecaab765ce605866f", size = 33150, upload-time = "2025-09-05T12:49:58.025Z" },
+    { url = "https://files.pythonhosted.org/packages/a0/29/10dd41cde849fb2f9b626c846b7ea30c99c81a18a5037a45cc4ba33c19a7/setproctitle-1.3.7-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:61e96febced3f61b766115381d97a21a6265a0f29188a791f6df7ed777aef698", size = 34463, upload-time = "2025-09-05T12:49:59.424Z" },
+    { url = "https://files.pythonhosted.org/packages/71/3c/cedd8eccfaf15fb73a2c20525b68c9477518917c9437737fa0fda91e378f/setproctitle-1.3.7-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:047138279f9463f06b858e579cc79580fbf7a04554d24e6bddf8fe5dddbe3d4c", size = 32848, upload-time = "2025-09-05T12:50:01.107Z" },
+    { url = "https://files.pythonhosted.org/packages/d1/3e/0a0e27d1c9926fecccfd1f91796c244416c70bf6bca448d988638faea81d/setproctitle-1.3.7-cp313-cp313-win32.whl", hash = "sha256:7f47accafac7fe6535ba8ba9efd59df9d84a6214565108d0ebb1199119c9cbbd", size = 12544, upload-time = "2025-09-05T12:50:15.81Z" },
+    { url = "https://files.pythonhosted.org/packages/36/1b/6bf4cb7acbbd5c846ede1c3f4d6b4ee52744d402e43546826da065ff2ab7/setproctitle-1.3.7-cp313-cp313-win_amd64.whl", hash = "sha256:fe5ca35aeec6dc50cabab9bf2d12fbc9067eede7ff4fe92b8f5b99d92e21263f", size = 13235, upload-time = "2025-09-05T12:50:16.89Z" },
+    { url = "https://files.pythonhosted.org/packages/e6/a4/d588d3497d4714750e3eaf269e9e8985449203d82b16b933c39bd3fc52a1/setproctitle-1.3.7-cp313-cp313t-macosx_10_13_universal2.whl", hash = "sha256:10e92915c4b3086b1586933a36faf4f92f903c5554f3c34102d18c7d3f5378e9", size = 18058, upload-time = "2025-09-05T12:50:02.501Z" },
+    { url = "https://files.pythonhosted.org/packages/05/77/7637f7682322a7244e07c373881c7e982567e2cb1dd2f31bd31481e45500/setproctitle-1.3.7-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:de879e9c2eab637f34b1a14c4da1e030c12658cdc69ee1b3e5be81b380163ce5", size = 13072, upload-time = "2025-09-05T12:50:03.601Z" },
+    { url = "https://files.pythonhosted.org/packages/52/09/f366eca0973cfbac1470068d1313fa3fe3de4a594683385204ec7f1c4101/setproctitle-1.3.7-cp313-cp313t-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:c18246d88e227a5b16248687514f95642505000442165f4b7db354d39d0e4c29", size = 34490, upload-time = "2025-09-05T12:50:04.948Z" },
+    { url = "https://files.pythonhosted.org/packages/71/36/611fc2ed149fdea17c3677e1d0df30d8186eef9562acc248682b91312706/setproctitle-1.3.7-cp313-cp313t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:7081f193dab22df2c36f9fc6d113f3793f83c27891af8fe30c64d89d9a37e152", size = 35267, upload-time = "2025-09-05T12:50:06.015Z" },
+    { url = "https://files.pythonhosted.org/packages/88/a4/64e77d0671446bd5a5554387b69e1efd915274686844bea733714c828813/setproctitle-1.3.7-cp313-cp313t-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:9cc9b901ce129350637426a89cfd650066a4adc6899e47822e2478a74023ff7c", size = 37376, upload-time = "2025-09-05T12:50:07.484Z" },
+    { url = "https://files.pythonhosted.org/packages/89/bc/ad9c664fe524fb4a4b2d3663661a5c63453ce851736171e454fa2cdec35c/setproctitle-1.3.7-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:80e177eff2d1ec172188d0d7fd9694f8e43d3aab76a6f5f929bee7bf7894e98b", size = 33963, upload-time = "2025-09-05T12:50:09.056Z" },
+    { url = "https://files.pythonhosted.org/packages/ab/01/a36de7caf2d90c4c28678da1466b47495cbbad43badb4e982d8db8167ed4/setproctitle-1.3.7-cp313-cp313t-musllinux_1_2_ppc64le.whl", hash = "sha256:23e520776c445478a67ee71b2a3c1ffdafbe1f9f677239e03d7e2cc635954e18", size = 35550, upload-time = "2025-09-05T12:50:10.791Z" },
+    { url = "https://files.pythonhosted.org/packages/dd/68/17e8aea0ed5ebc17fbf03ed2562bfab277c280e3625850c38d92a7b5fcd9/setproctitle-1.3.7-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:5fa1953126a3b9bd47049d58c51b9dac72e78ed120459bd3aceb1bacee72357c", size = 33727, upload-time = "2025-09-05T12:50:12.032Z" },
+    { url = "https://files.pythonhosted.org/packages/b2/33/90a3bf43fe3a2242b4618aa799c672270250b5780667898f30663fd94993/setproctitle-1.3.7-cp313-cp313t-win32.whl", hash = "sha256:4a5e212bf438a4dbeece763f4962ad472c6008ff6702e230b4f16a037e2f6f29", size = 12549, upload-time = "2025-09-05T12:50:13.074Z" },
+    { url = "https://files.pythonhosted.org/packages/0b/0e/50d1f07f3032e1f23d814ad6462bc0a138f369967c72494286b8a5228e40/setproctitle-1.3.7-cp313-cp313t-win_amd64.whl", hash = "sha256:cf2727b733e90b4f874bac53e3092aa0413fe1ea6d4f153f01207e6ce65034d9", size = 13243, upload-time = "2025-09-05T12:50:14.146Z" },
 ]

 [[package]]
@ -1669,7 +1677,7 @@ wheels = [
 [[package]]
 name = "tractor"
 version = "0.1.0a6.dev0"
-source = { git = "https://github.com/goodboy/tractor.git?branch=wkt%2Faddr_unpacking#a966928dead4cbf369ad244834ce9dbe988ac8ca" }
+source = { git = "https://github.com/goodboy/tractor.git?branch=piker_pin#0d95a9e63c3a250bb9f48862c828db16b09bb486" }
 dependencies = [
    { name = "bidict" },
    { name = "colorlog" },