.ib.feed: cache quote channel contexts

Replace the manual `_quote_streams` raw-channel registry with a
symbol-keyed `maybe_open_context()` resource whose lifetime extends to
the final quote consumer.

Give every `open_aio_quote_stream()` caller a lexical child from the
cached `LinkedTaskChannel` broadcaster so concurrent users receive the
same quotes and close independently.

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

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

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.

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,
 
-) -> (
+) -> tractor.to_asyncio.LinkedTaskChannel:
-    trio.abc.Channel|  # iface
-    tractor.to_asyncio.LinkedTaskChannel  # actually
-):
     '''
-    Open a real-time `Ticker` quote stream from an `asyncio.Task`
+    Open one raw inter-event-loop quote channel for `symbol`.
-    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.
 
     '''
-    from tractor.trionics import broadcast_receiver
+    chan: tractor.to_asyncio.LinkedTaskChannel
-    global _quote_streams
+    started_contract: Contract
-
-    from_aio = _quote_streams.get(symbol)
-    if from_aio:
-
-        # if we already have a cached feed deliver a rx side clone
-        # to consumer
-        async with broadcast_receiver(
-            from_aio,
-            2**6,
-        ) as from_aio:
-            yield from_aio
-            return
-
-    from_aio: tractor.to_asyncio.LinkedTaskChannel
     async with tractor.to_asyncio.open_channel_from(
         _setup_quote_stream,
         symbol=symbol,
         contract=contract,
 
-    ) as (from_aio, contract):
+    ) as (chan, started_contract):
+        assert started_contract
+        yield chan
 
-        assert contract
 
-        # TODO? de-reg on teardown of last consumer task?
+@acm
-        # -> why aren't we using `.trionics.maybe_open_context()`
+async def open_aio_quote_stream(
-        # here again?? (we are in `open_client_proxies()` tho?)
+    symbol: str,
-        #
+    contract: Contract|None = None,
-        # cache feed for later consumers
-        _quote_streams[symbol] = from_aio
 
-        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!

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)