.clearing: own dark quote broadcast subscription

Give `clear_dark_triggers()` a dedicated
`MsgStream.subscribe()` handle with `raise_on_lag=False` instead
of mutating the private lag policy on whichever root or cached
child sits in `Flume.stream`.

Also,
- factor the hot quote loop into `_clear_dark_triggers()` so the
  public task owns the child receiver for its full lifetime.
- add an EMS regression proving the lag policy, child lifetime,
  and source-stream state remain isolated.

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`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,79 @@
+'''
+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),
+    ]