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.

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.

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

@@ -1,32 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-session: unavailable
-timestamp: 2026-08-13T22:59:40Z
-git_ref: b691422b
-scope: code
-substantive: true
-raw_file: 20260813T225940Z_b691422b_prompt_io.raw.md
----
-
-## Prompt
-
-Continue the piker broadcast-consumer audit after the cached feed ownership
-commit and prepare the next smallest ownership fix.
-
-## Response summary
-
-Made every Deribit cached price-feed caller own a lexical child subscription
-from the retained `LinkedTaskChannel` broadcaster instead of constructing
-competing roots around one source channel.
-
-## Files changed
-
-- `piker/brokers/deribit/api.py` - uniform child subscription ownership for
-  price-feed misses and cache hits.
-- `tests/test_deribit_broadcast.py` - concurrent fan-out and child lifetime
-  regression.
-
-## Human edits
-
-None - generated output follows the requested broadcast-consumer audit.

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

@@ -1,37 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-timestamp: 2026-08-13T22:59:40Z
-git_ref: b691422b
-diff_cmd: git diff HEAD~1..HEAD
----
-
-The user asked to continue the piker broadcast-consumer audit after committing
-the cached feed ownership fix. The next isolated item was Deribit's cached
-price feed, which wrapped one retained `LinkedTaskChannel` in competing
-broadcaster roots for cache hits.
-
-> `git diff HEAD~1..HEAD -- piker/brokers/deribit/api.py`
-
-Changed `maybe_open_price_feed()` so both the first cache owner and every cache
-hit enter `LinkedTaskChannel.subscribe()` and receive a lexical child from the
-channel's single retained broadcaster root.
-
-> `git diff HEAD~1..HEAD -- tests/test_deribit_broadcast.py`
-
-Added a concurrent ownership regression using real Tractor broadcast state.
-It proves a miss owner and cache-hit caller receive the same quote, closing the
-hit child leaves the owner live, and both children clean up independently while
-the channel root remains registered.
-
-Verification output:
-
-```text
-.                                                                        [100%]
-1 passed in 0.01s
-```
-
-The new test passes through a source-isolated Deribit import. Test Ruff,
-Python compilation and whitespace checks pass. Full package import and
-full-file Deribit Ruff remain blocked by pre-existing Qt, msgspec model, and
-legacy Deribit lint errors unrelated to this patch.

piker/brokers/deribit/api.py

@@ -36,7 +36,6 @@ from trio_typing import TaskStatus
 from rapidfuzz import process as fuzzy
 import numpy as np
 from tractor.trionics import (
-    BroadcastReceiver,
     broadcast_receiver,
     maybe_open_context,
     collapse_eg,
@@ -590,20 +589,20 @@ async def open_price_feed(
 @acm
 async def maybe_open_price_feed(
     instrument: str
-) -> BroadcastReceiver:
+) -> trio.abc.ReceiveStream:
 
     # TODO: add a predicate to maybe_open_context
-    feed: to_asyncio.LinkedTaskChannel
     async with maybe_open_context(
         acm_func=open_price_feed,
         kwargs={
             'instrument': instrument
         },
         key=f'{instrument}-price',
-    ) as (_, feed):
+    ) as (cache_hit, feed):
-        bstream: BroadcastReceiver
+        if cache_hit:
-        async with feed.subscribe() as bstream:
+            yield broadcast_receiver(feed, 10)
-            yield bstream
+        else:
+            yield feed
 
 
 

piker/clearing/_ems.py

@@ -159,11 +159,11 @@ class DarkBook(Struct):
 _DEFAULT_SIZE: float = 1.0
 
 
-async def _clear_dark_triggers(
+async def clear_dark_triggers(
 
     router: Router,
     brokerd_orders_stream: tractor.MsgStream,
-    quote_stream: trionics.AsyncReceiver,
+    quote_stream: tractor.MsgStream,
     broker: str,
     fqme: str,
 
@@ -182,6 +182,7 @@ async def _clear_dark_triggers(
     # - port to the new ringbuf stuff in `tractor.ipc`!
     # - numba all this!
     # - this stream may eventually contain multiple symbols
+    quote_stream._raise_on_lag = False
     async for quotes in quote_stream:
         # start = time.time()
         for sym, quote in quotes.items():
@@ -313,32 +314,6 @@ async def _clear_dark_triggers(
         # print(f'execs scan took: {time.time() - start}')
 
 
-async def clear_dark_triggers(
-    router: Router,
-    brokerd_orders_stream: tractor.MsgStream,
-    quote_stream: tractor.MsgStream,
-    broker: str,
-    fqme: str,
-    book: DarkBook,
-
-) -> None:
-    '''
-    Run dark clearing on a dedicated non-strict quote subscription.
-
-    '''
-    async with quote_stream.subscribe(
-        raise_on_lag=False,
-    ) as quotes_stream:
-        await _clear_dark_triggers(
-            router,
-            brokerd_orders_stream,
-            quotes_stream,
-            broker,
-            fqme,
-            book,
-        )
-
-
 class TradesRelay(Struct):
 
     # for now we keep only a single connection open with

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
@@ -678,36 +677,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
@@ -757,51 +726,29 @@ class Feed(Struct):
         if len(mods) == 1:
             # just pass the datad stream directly if only one provider
             # was detected.
-            stream: (
+            stream = self.streams[list(brokers)[0]]
-                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.
-            broker2bstreams: dict[
+            for brokername in mods:
-                str,
+                stream = self.streams[brokername]
-                trionics.BroadcastReceiver,
+                nurse.start_soon(relay_to_common_memchan, stream)
-            ] = {}
-            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:
@@ -810,23 +757,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')
 
@@ -886,17 +820,7 @@ async def maybe_open_feed(
     in a tractor broadcast receiver.
 
     '''
-    cache_key: tuple[
+    fqme = fqmes[0]
-        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,
@@ -909,73 +833,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 (
+        if cache_hit:
-            cache_hit
+            log.info(f'Using cached feed for {fqme}')
-            and
+            # add a new broadcast subscription for the quote stream
-            log.at_least_level('info')
+            # if this feed is likely already in use
-        ):
-            log.info(
-                f'Using cached feed for key:\n'
-                f'{cache_key!r}'
-            )
 
-        # The cached `Feed` exclusively owns provider root streams.
+            async with trionics.gather_contexts(
-        # Every caller, including the first cache owner, gets a local
+                mngrs=[stream.subscribe() for stream in feed.streams.values()]
-        # child cursor and copied descriptors for its lexical lifetime.
+            ) as bstreams:
-        broker_names: tuple[str, ...] = tuple(feed.streams)
+                for bstream, flume in zip(bstreams, feed.flumes.values()):
-        bstreams: tuple[
+                    # XXX: TODO: horrible hackery that needs fixing..
-            trionics.BroadcastReceiver,
+                    # i guess we have to create context proxies?
-            ...,
+                    bstream._ctx = flume.stream._ctx
-        ]
+                    flume.stream = bstream
-        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
 
-                bstream: trionics.BroadcastReceiver = (
+                yield feed
-                    broker2bstreams[broker]
+        else:
-                )
+            yield feed
-
-                # 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

tests/test_deribit_broadcast.py

@@ -1,112 +0,0 @@
-'''
-Deribit cached feed broadcast ownership regressions.
-
-'''
-from contextlib import asynccontextmanager as acm
-
-import tractor
-import trio
-
-from piker.brokers.deribit import api
-
-
-def test_price_feed_owns_child_subscription(
-    monkeypatch,
-) -> None:
-    '''
-    Every cached price-feed caller must own one lexical child receiver.
-
-    `maybe_open_price_feed()` previously yielded the raw
-    `LinkedTaskChannel` to the first cache owner, then constructed a new
-    broadcaster root around that same channel for every cache hit.
-    Concurrent roots competed for source receives, so quotes could be
-    split across callers and exiting one wrapper did not express child
-    ownership on the retained channel broadcaster.
-
-    Return one real `LinkedTaskChannel`-shaped fake first as a miss and
-    then as a concurrent hit. Nest both caller contexts and prove each
-    gets a distinct child from the channel's single retained root and
-    receives the same quote. Close the hit child, then prove the first
-    owner continues receiving before its own lexical child closes.
-
-    '''
-    class FakeLinkedTaskChannel:
-        '''
-        Retain one real broadcaster and expose its public subscription.
-
-        '''
-        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 quote through the retained source channel.
-
-            '''
-            await self._tx.send(msg)
-
-    channel: FakeLinkedTaskChannel = FakeLinkedTaskChannel()
-    context_entries: int = 0
-
-    @acm
-    async def maybe_open_context(**kwargs):
-        '''
-        Return the controlled channel with either cache state.
-
-        '''
-        nonlocal context_entries
-        cache_hit: bool = context_entries > 0
-        context_entries += 1
-        yield cache_hit, channel
-
-    monkeypatch.setattr(
-        api,
-        'maybe_open_context',
-        maybe_open_context,
-    )
-
-    async def main() -> None:
-        owner: tractor.trionics.BroadcastReceiver
-        async with api.maybe_open_price_feed(
-            'btc-usd',
-        ) as owner:
-            hit: tractor.trionics.BroadcastReceiver
-            async with api.maybe_open_price_feed(
-                'btc-usd',
-            ) as hit:
-                assert hit is not owner
-                assert len(channel._broadcaster._state.subs) == 3
-
-                await channel.push({'value': 1})
-                assert await owner.receive() == {'value': 1}
-                assert await hit.receive() == {'value': 1}
-
-            assert hit._closed
-            assert not owner._closed
-            assert len(channel._broadcaster._state.subs) == 2
-
-            await channel.push({'value': 2})
-            assert await owner.receive() == {'value': 2}
-
-        assert owner._closed
-        assert len(channel._broadcaster._state.subs) == 1
-
-    trio.run(main)

tests/test_ems_broadcast.py

@@ -1,79 +0,0 @@
-'''
-EMS broadcast ownership regressions.
-
-'''
-from contextlib import asynccontextmanager as acm
-
-import trio
-
-from piker.clearing import _ems
-
-
-def test_dark_clearing_owns_non_strict_subscription(
-    monkeypatch,
-) -> None:
-    '''
-    Dark clearing must not mutate another feed consumer's lag policy.
-
-    `clear_dark_triggers()` previously assigned `_raise_on_lag` on
-    whichever root or temporary receiver was stored in `Flume.stream`.
-    That changed private shared state and made behavior depend on feed
-    cache ownership. Supply a stream whose subscription records the
-    public lag-policy argument and yields a distinct child. Replace
-    the hot trigger loop with a checkpointing probe, then prove the
-    non-strict child is entered, passed to the core, and closed only
-    after processing completes while the source remains untouched.
-
-    '''
-    events: list[tuple] = []
-    child = object()
-
-    class QuoteStream:
-        '''
-        Record public subscription ownership without quote traffic.
-
-        '''
-        @acm
-        async def subscribe(self, raise_on_lag: bool = True):
-            '''
-            Yield one dedicated child for the wrapper lifetime.
-
-            '''
-            events.append(('enter', raise_on_lag))
-            try:
-                yield child
-            finally:
-                events.append(('exit', raise_on_lag))
-
-    async def clear_core(*args) -> None:
-        '''
-        Prove the child remains owned while processing runs.
-
-        '''
-        events.append(('core', args[2] is child))
-        await trio.lowlevel.checkpoint()
-
-    monkeypatch.setattr(
-        _ems,
-        '_clear_dark_triggers',
-        clear_core,
-    )
-
-    async def main() -> None:
-        stream = QuoteStream()
-        await _ems.clear_dark_triggers(
-            object(),
-            object(),
-            stream,
-            'ib',
-            'nvda.nasdaq.ib',
-            object(),
-        )
-        assert not hasattr(stream, '_raise_on_lag')
-
-    trio.run(main)
-    assert events == [
-        ('enter', False),
-        ('core', True),
-        ('exit', False),
-    ]

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:]