.clearing: backport owned dark quote subscriptions

Import the source-equivalent EMS code and its original wrapper
regression. `clear_dark_triggers()` owns a non-strict child from
`quote_stream.subscribe()` for the lifetime of its clearing loop,
instead of changing a shared receiver's private lag policy.

LOCAL ADDITION,
Add a second regression using the production clearing loop and a
real broadcaster. Deliver a tick to EMS and a sibling, cancel EMS,
and prove the sibling remains usable with strict lag policy.
No order trigger or live broker is involved in this test.

Downstream replay,
The EMS code has the source patch's stable patch ID. The test file
also contains the local real-broadcaster case, so the complete
commit is not patch-identical. Preserve that extra coverage when
reconciling an already-applied EMS delta.

Backport-of: 12b0d66e85
Source-author: goodboy <goodboy_foss@protonmail.com>
Prompt-IO: ai/prompt-io/opencode/20261022T212248Z_dc618229_ib_recovery_prompt_io.md

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

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: tractor.MsgStream,
+    quote_stream: trionics.AsyncReceiver,
     broker: str,
     fqme: str,
 
@@ -182,7 +182,6 @@ 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():
@@ -314,6 +313,32 @@ 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

tests/test_ems_broadcast.py

@@ -0,0 +1,133 @@
+'''
+EMS broadcast ownership regressions.
+
+'''
+from contextlib import asynccontextmanager as acm
+
+import trio
+from trio.testing import wait_all_tasks_blocked
+from tractor.trionics import broadcast_receiver
+
+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),
+    ]
+
+
+def test_dark_clearing_with_real_broadcast() -> None:
+    '''
+    Dark clearing must consume quotes without owning sibling lifetime.
+
+    Exercise the production trigger loop and real Tractor broadcaster
+    together. Deliver one trade tick to both EMS and a sibling cursor;
+    DarkBook.lasts proves EMS processed it. Cancel only the EMS task
+    and verify its subscription disappears while the sibling still
+    receives the next tick with its strict lag policy intact.
+
+    The book has no order triggers, so this requires no broker or
+    account. Unlike the imported wrapper test, no stream or clearing
+    function is replaced.
+
+    '''
+    async def main() -> None:
+        symbol = 'qqq.nasdaq.ib'
+        book = _ems.DarkBook(broker='ib', triggers={}, lasts={})
+        tx, rx = trio.open_memory_channel(8)
+        root = broadcast_receiver(rx, 8)
+        quote = {symbol: {'ticks': [{'type': 'trade', 'price': 100}]}}
+
+        with trio.fail_after(2):
+            async with tx, rx, root, root.subscribe() as sibling:
+                async with trio.open_nursery() as nursery:
+                    nursery.start_soon(
+                        _ems.clear_dark_triggers,
+                        None,
+                        None,
+                        root,
+                        'ib',
+                        symbol,
+                        book,
+                    )
+                    await wait_all_tasks_blocked()
+                    assert len(root._state.subs) == 3
+                    await tx.send(quote)
+                    assert await sibling.receive() == quote
+                    await wait_all_tasks_blocked()
+                    assert book.lasts[symbol] == 100
+                    assert root._raise_on_lag
+                    assert sibling._raise_on_lag
+                    nursery.cancel_scope.cancel()
+
+                assert len(root._state.subs) == 2
+                assert not sibling._closed
+                await tx.send(quote)
+                assert await sibling.receive() == quote
+
+    trio.run(main)