.data: isolate cached feed stream consumers

Keep provider root streams private to the cached `Feed` and give every
`maybe_open_feed()` caller, including the first owner, a local
subscription view with copied `Flume` descriptors.

Also,
- key cached resources by the normalized full symbol set and stream
  policy instead of only the first FQME.
- subscribe each `Feed.open_multi_stream()` provider relay and wait for
  all relays to start before yielding the merged channel.
- give each provider relay an independent send-channel clone so one
  provider EOC cannot terminate its siblings.
- document `Feed.pause()` and `Feed.resume()` as shared remote-provider
  controls rather than caller-local cursor operations.
- cover miss/hit ownership, multi-market provider mapping, relay
  fan-out, lexical teardown and cache-key separation.

Prompt-IO: ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.md

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

ai/prompt-io/20261007T165453Z_bde2457b_prompt_io.md

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

ai/prompt-io/20261007T165453Z_bde2457b_prompt_io.raw.md

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

ai/prompt-io/20261010T232852Z_1fa45dfa_prompt_io.md

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

ai/prompt-io/20261010T232852Z_1fa45dfa_prompt_io.raw.md

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

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

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

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

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

ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.md

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

ai/prompt-io/opencode/20260813T202402Z_12b0d66e_prompt_io.raw.md

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

ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.md

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

ai/prompt-io/opencode/20260813T230744Z_ddac3553_prompt_io.raw.md

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

config/conf.toml

@@ -1,14 +1,17 @@
 [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',
 
-  # same but UDS
-  '/unix/run/user/1000/piker/pikerd.sock',
+  # UDS alternative: replace TCP and select the UDS transport.
+  # '/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

piker/brokers/binance/feed.py

@@ -364,8 +364,6 @@ 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:

piker/brokers/ib/api.py

@@ -1584,6 +1584,72 @@ 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,
         *,
@@ -1596,38 +1662,30 @@ class MethodProxy:
         ``tractor.to_asyncio`` layer.
 
         '''
-        chan = self.chan
-        await chan.send((meth, kwargs))
+        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)
 
-        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')
+        msg: dict = slot[0]
+        match msg:
+            # OUR method-call response B)
+            case {'result': res}:
                 return res
 
-            elif 'exception' in msg:
-                err = msg.get('exception')
+            case {'exception': err}:
                 raise err
 
-            elif 'error' in msg:
-                etype, emsg = msg
-                log.warning(f'IB error relay: {emsg}')
-                continue
-
-            else:
-                log.warning(f'UNKNOWN IB MSG: {msg}')
+            case _:
+                raise RuntimeError(
+                    f'Invalid method response for mid={mid}: {msg!r}'
+                )
 
     def status_event(
         self,
@@ -1680,22 +1738,31 @@ async def open_aio_client_method_relay(
                 log.info('asyncio `Client` method-proxy SHUTDOWN!')
                 break
 
-            case (meth_name, kwargs):
-                meth_name, kwargs = msg
+            case (meth_name, kwargs, mid):
                 meth = getattr(client, meth_name)
 
                 try:
                     resp = await meth(**kwargs)
-                    # echo the msg back
-                    chan.send_nowait({'result': resp})
+                    # echo the msg back tagged with the
+                    # req-id for caller-side correlation.
+                    chan.send_nowait({
+                        'mid': mid,
+                        'result': resp,
+                    })
 
-                except (
-                    RequestError,
-
-                    # TODO: relay all errors to trio?
-                    # BaseException,
-                ) as err:
-                    chan.send_nowait({'exception': err})
+                # 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,
+                    })
 
             case {'error': content}:
                 chan.send_nowait({'exception': content})
@@ -1704,6 +1771,73 @@ async def open_aio_client_method_relay(
                 raise ValueError(f'Unhandled msg {msg}')
 
 
+async def relay_client_proxy_messages(
+    chan: tractor.to_asyncio.LinkedTaskChannel,
+    proxy: MethodProxy,
+
+) -> None:
+    '''
+    Route one proxy's asyncio-client msgs from a single reader.
+
+    Each `open_client_proxy()` allocates its own linked channel and
+    starts this task in its relay nursery. Any number of Trio caller
+    tasks may concurrently write tagged requests through
+    `MethodProxy._run_method()` and wait on their per-`mid` slots.
+    This task alone reads the channel and dispatches each response to
+    its waiting caller.
+
+    Separate proxy contexts own separate channels and reader tasks;
+    no cross-proxy response broadcast is intended. IB status events
+    still reach each proxy through its own `Client.inline_errors()`
+    handler.
+
+    '''
+    try:
+        while not chan.closed():
+            # TODO: implement reconnect functionality like
+            # in our `.data._web_bs.NoBsWs`
+            # try:
+            msg = await chan.receive()
+            # except ConnectionError:
+            #     proxy.reset()
+
+            match msg:
+                # Correlated response from a method call.
+                case {'mid': _}:
+                    proxy._deliver_method_response(msg)
+
+                # routine (api-farm conn) status events
+                # relayed inline by `Client.inline_errors()`;
+                # not a response to a method call so just
+                # log at info and keep waiting.
+                case ('event', status_msg):
+                    reason = status_msg['reason']
+                    event = proxy.event_table.pop(reason, None)
+                    if (
+                        event
+                        and
+                        event.statistics().tasks_waiting
+                    ):
+                        log.info(f'Relaying ib status message: {msg}')
+                        event.set()
+
+                # Inline API error with no request correlation.
+                case ('error', emsg):
+                    log.warning(f'IB error relay: {emsg}')
+
+                # Out-of-band client error with no method `mid`.
+                case {'exception': err}:
+                    log.error(
+                        f'Uncorrelated IB client error: {err!r}'
+                    )
+
+                # Preserve visibility into unexpected relay traffic.
+                case _:
+                    log.warning(f'UNKNOWN IB MSG: {msg}')
+    except trio.EndOfChannel:
+        log.info('IB client proxy relay closed')
+
+
 @acm
 async def open_client_proxy(
     client: Client,
@@ -1740,28 +1874,11 @@ async def open_client_proxy(
                 continue
             setattr(proxy, name, partial(proxy._run_method, meth=name))
 
-        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)
+        relay_tn.start_soon(
+            relay_client_proxy_messages,
+            chan,
+            proxy,
+        )
 
         yield proxy
 

piker/brokers/ib/feed.py

@@ -738,10 +738,6 @@ 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,
@@ -770,8 +766,6 @@ 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:
@@ -834,9 +828,6 @@ 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,
@@ -953,54 +944,52 @@ async def _setup_quote_stream(
 
 
 @acm
-async def open_aio_quote_stream(
+async def _open_aio_quote_channel(
     symbol: str,
     contract: Contract|None = None,
 
-) -> (
-    trio.abc.Channel|  # iface
-    tractor.to_asyncio.LinkedTaskChannel  # actually
-):
+) -> tractor.to_asyncio.LinkedTaskChannel:
     '''
-    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.
+    Open one raw inter-event-loop quote channel for `symbol`.
 
     '''
-    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
+    chan: tractor.to_asyncio.LinkedTaskChannel
+    started_contract: Contract
     async with tractor.to_asyncio.open_channel_from(
         _setup_quote_stream,
         symbol=symbol,
         contract=contract,
 
-    ) as (from_aio, contract):
+    ) as (chan, started_contract):
+        assert started_contract
+        yield chan
 
-        assert contract
 
-        # 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
+@acm
+async def open_aio_quote_stream(
+    symbol: str,
+    contract: Contract|None = None,
 
-        yield 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
 
 
 # TODO: cython/mypyc/numba this!

piker/cli/__init__.py

@@ -47,7 +47,7 @@ log = get_logger('piker.cli')
 
 def load_trans_eps(
     network: dict|None = None,
-    maddrs: list[str]|None = None,
+    maddrs: str|tuple[str, ...]|list[str]|None = None,
 
 ) -> dict[str, list]:
     '''
@@ -60,6 +60,8 @@ def load_trans_eps(
         return parse_endpoints(network)
 
     elif maddrs:
+        if isinstance(maddrs, str):
+            maddrs = [maddrs]
         return parse_endpoints(
             {'pikerd': list(maddrs)}
         )
@@ -87,11 +89,11 @@ def load_trans_eps(
 @click.option(
     '--maddrs',
     '-m',
-    default=None,
+    multiple=True,
     help='Multiaddrs to bind or contact',
 )
 def pikerd(
-    maddrs: list[str]|None,
+    maddrs: tuple[str, ...],
     loglevel: str,
     tl: bool,
     pdb: bool,

piker/data/feed.py

@@ -30,6 +30,7 @@ from collections import (
     defaultdict,
     abc,
 )
+from copy import copy
 from contextlib import asynccontextmanager as acm
 from functools import partial
 import time
@@ -674,6 +675,36 @@ 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
@@ -723,29 +754,51 @@ class Feed(Struct):
         if len(mods) == 1:
             # just pass the brokerd stream directly if only one provider
             # was detected.
-            stream = self.streams[list(brokers)[0]]
+            stream: (
+                tractor.MsgStream
+                |trionics.BroadcastReceiver
+            ) = self.streams[list(brokers)[0]]
+            bstream: trionics.BroadcastReceiver
             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.
-            for brokername in mods:
-                stream = self.streams[brokername]
-                nurse.start_soon(relay_to_common_memchan, stream)
+            broker2bstreams: dict[
+                str,
+                trionics.BroadcastReceiver,
+            ] = {}
+            broker: str
+            for broker in mods:
+                stream: (
+                    tractor.MsgStream
+                    |trionics.BroadcastReceiver
+                ) = self.streams[broker]
 
+                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:
@@ -754,10 +807,23 @@ 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')
 
@@ -817,7 +883,17 @@ async def maybe_open_feed(
     in a tractor broadcast receiver.
 
     '''
-    fqme = fqmes[0]
+    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),
+    )
 
     async with trionics.maybe_open_context(
         acm_func=open_feed,
@@ -830,27 +906,73 @@ async def maybe_open_feed(
             'allow_overruns': kwargs.get('allow_overruns', True),
             'start_stream': kwargs.get('start_stream', True),
         },
-        key=fqme,
+        key=cache_key,
 
     ) as (cache_hit, feed):
+        feed: Feed
 
-        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
+        if (
+            cache_hit
+            and
+            log.at_least_level('info')
+        ):
+            log.info(
+                f'Using cached feed for key:\n'
+                f'{cache_key!r}'
+            )
 
-            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
+        # 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
 
-                yield feed
-        else:
-            yield feed
+                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
 
 
 @acm

piker/service/_actor_runtime.py

@@ -214,7 +214,27 @@ async def open_pikerd(
         tractor.trionics.collapse_eg(),
         trio.open_nursery() as service_tn,
     ):
-        for addr in reg_addrs:
+        # `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:
             # normalize to a wrapped `tractor` addr-type;
             # entries may be raw `tuple`s when passed in
             # from (test) client code.
@@ -222,9 +242,16 @@ async def open_pikerd(
                 addr,
             )
             uaddr: tuple = wladdr.unwrap()
-            if (
-                uaddr not in root_actor.accept_addrs
-            ):
+            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:
                 raise RuntimeError(
                     f'`pikerd` failed to bind on {addr}!\n'
                     'Maybe you have another daemon already running?'

piker/service/_daemon.py

@@ -78,14 +78,19 @@ async def maybe_spawn_daemon(
     lock = Services.locks[service_name]
     await lock.acquire()
 
-    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:
+        # 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'])
+            )
+
         async with find_service(
             service_name,
             registry_addrs=registry_addrs,
@@ -217,11 +222,13 @@ async def maybe_spawn_daemon(
             lock.statistics().owner is current_task()
         ):
             log.exception(
-                f'Releasing stale lock after crash..?'
+                f'Releasing stale lock after crash..?\n'
+                f'\n'
                 f'{err!r}\n'
             )
             lock.release()
-            raise err
+
+        raise
 
 
 async def spawn_emsd(

piker/ui/cli.py

@@ -199,6 +199,7 @@ def chart(
             'registry_addrs',
             [],
         )
+        chart_eps: list|None = None
         network: dict = conf.get('network')
         if network:
             eps: dict = load_trans_eps(

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 = "piker_pin" }
+tractor = { git = "https://github.com/goodboy/tractor.git", branch = "wkt/addr_unpacking" }
 # tractor = { git = "https://pikers.dev/goodboy/tractor", branch = "piker_pin" }
 # ------ goodboy ------
 # hackin dev-envs, usually there's something new he's hackin in..

tests/test_feed_broadcast.py

@@ -0,0 +1,347 @@
+'''
+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:]

tests/test_ib_broadcast.py

@@ -0,0 +1,120 @@
+'''
+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)

tests/test_ib_method_proxy.py

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

tests/test_services.py

@@ -120,41 +120,6 @@ 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,

tests/test_startup.py

@@ -0,0 +1,214 @@
+'''
+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)

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=piker_pin" },
+    { name = "tractor", git = "https://github.com/goodboy/tractor.git?branch=wkt%2Faddr_unpacking" },
     { name = "trio", specifier = ">=0.27" },
     { name = "trio-typing", specifier = ">=0.10.0" },
     { name = "trio-util", specifier = ">=0.7.0,<0.8.0" },
@@ -1560,30 +1560,22 @@ wheels = [
 
 [[package]]
 name = "setproctitle"
-version = "1.3.7"
+version = "1.3.8"
 source = { registry = "https://pypi.org/simple" }
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]
 
 [[package]]
@@ -1677,7 +1669,7 @@ wheels = [
 [[package]]
 name = "tractor"
 version = "0.1.0a6.dev0"
-source = { git = "https://github.com/goodboy/tractor.git?branch=piker_pin#0d95a9e63c3a250bb9f48862c828db16b09bb486" }
+source = { git = "https://github.com/goodboy/tractor.git?branch=wkt%2Faddr_unpacking#a966928dead4cbf369ad244834ce9dbe988ac8ca" }
 dependencies = [
     { name = "bidict" },
     { name = "colorlog" },