Expose stream subscriber lag policy

`MsgStream.subscribe()` and `LinkedTaskChannel.subscribe()` omitted
`BroadcastReceiver.raise_on_lag`, forcing downstream consumers to
mutate a private receiver attribute when overruns were acceptable.

Add `raise_on_lag` to both public wrappers. The first subscription
sets the irreversible root broadcaster's policy, while every child
selects its own strict or warn/drop/resume behavior independently.

Document both fan-out APIs. Cover policy forwarding plus real IPC
and infected-asyncio paths.

Prompt-IO: ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.md

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

ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.md

@@ -0,0 +1,34 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: ses_0799212ebffe42arY96czXn89F
+timestamp: 2026-08-12T21:31:17Z
+git_ref: 51185487
+scope: code
+substantive: true
+raw_file: 20260812T213117Z_51185487_prompt_io.raw.md
+---
+
+## Prompt
+
+For Tractor PR 490 item four, expose `raise_on_lag` through the public IPC and
+asyncio linked-channel subscription wrappers. Review, verify and return a
+complete commit plan before applying the API downstream in piker.
+
+## Response summary
+
+Added public per-subscription lag policy to both wrappers, preserved first-call
+root policy, documented the semantics, and covered forwarding plus real fan-out
+paths.
+
+## Files changed
+
+- `tractor/_streaming.py` - `MsgStream` lag policy forwarding.
+- `tractor/to_asyncio.py` - linked-channel lag policy forwarding.
+- `docs/guide/streaming.rst` - IPC fan-out policy docs.
+- `docs/guide/asyncio.rst` - linked-channel fan-out policy docs.
+- `tests/test_task_broadcasting.py` - wrapper policy regression.
+
+## Human edits
+
+None - generated output follows the requested fourth iterative item.

ai/prompt-io/opencode/20260812T213117Z_51185487_prompt_io.raw.md

@@ -0,0 +1,53 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-12T21:31:17Z
+git_ref: 51185487
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+The user requested the fourth iterative refinement for Tractor PR 490: expose
+subscriber lag policy through the public `MsgStream.subscribe()` and
+`LinkedTaskChannel.subscribe()` wrappers, then stop for a complete commit
+plan. This enables piker to replace private receiver mutation.
+
+> `git diff HEAD~1..HEAD -- tractor/_streaming.py`
+
+Added `raise_on_lag: bool = True` to `MsgStream.subscribe()`. The first call
+passes the policy to the irreversibly allocated root broadcaster and its
+child; later calls configure each child independently while retaining the
+root's first-call policy.
+
+> `git diff HEAD~1..HEAD -- tractor/to_asyncio.py`
+
+Added equivalent lag-policy forwarding to `LinkedTaskChannel.subscribe()`.
+
+> `git diff HEAD~1..HEAD -- docs/guide/streaming.rst`
+
+> `git diff HEAD~1..HEAD -- docs/guide/asyncio.rst`
+
+Documented strict versus warn/drop/resume behavior, independent child policy,
+and first-call root policy for both wrapper types.
+
+> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py`
+
+Added a parameterized wrapper-level regression using minimal receive-compatible
+handles. It verifies a first non-raising subscription configures root and
+child, then a later strict child does not mutate the sticky root policy.
+
+Verification output:
+
+```text
+...................                                                      [100%]
+19 passed in 5.71s
+
+....................                                                     [100%]
+20 passed in 6.88s
+
+.                                                                        [100%]
+1 passed in 0.86s
+```
+
+The second and third runs cover actual `MsgStream` and infected-asyncio
+`LinkedTaskChannel` fan-out respectively. Python compilation and
+`git diff --check` passed. Adversarial review found no actionable issues.

docs/guide/asyncio.rst

@@ -209,6 +209,11 @@ The underlying broadcast machinery is lazily allocated on first
 use and is *not* reversible for the channel's remaining lifetime,
 so only reach for it when you actually want the fan-out.
 
+As with ``MsgStream``, pass ``raise_on_lag=False`` for a consumer
+which may warn, drop old values and resume from the retained window.
+Each child chooses independently; the first subscription also fixes
+the linked channel's root receive policy.
+
 One-shot calls with ``run_task()``
 ----------------------------------
 When you just want a single ``asyncio`` result and no streaming

docs/guide/streaming.rst

@@ -171,6 +171,12 @@ keeps pace with the *fastest* subscriber; a task falling more
 than the buffered window behind has its next receive raise
 ``tractor.trionics.Lagged`` to say it lost data.
 
+Pass ``raise_on_lag=False`` when a consumer may drop old values and
+resume from the oldest retained item instead. The receiver logs the
+overrun rather than raising. Each child subscription chooses its own
+policy; the first call also fixes the policy of the stream's root
+receive handle because broadcaster allocation is irreversible.
+
 The broadcast handle stays duplex btw: it proxies ``send()``
 through to the underlying stream, so each subscriber task can
 keep talking upstream while consuming its fan-out copy.

tests/test_task_broadcasting.py

@@ -8,6 +8,7 @@ from contextlib import (
 from functools import partial
 from itertools import cycle
 import time
+from types import SimpleNamespace
 from typing import Optional
 import warnings
 
@@ -15,6 +16,7 @@ import pytest
 import trio
 from trio.lowlevel import current_task
 import tractor
+from tractor.to_asyncio import LinkedTaskChannel
 from tractor.trionics import (
     broadcast_receiver,
     BroadcastReceiveError,
@@ -952,3 +954,74 @@ def test_no_raise_on_lag():
 
     with pytest.raises(KeyboardInterrupt):
         trio.run(main)
+
+
+@pytest.mark.parametrize(
+    ('subscribe', 'chan_attr'),
+    [
+        (tractor.MsgStream.subscribe, '_rx_chan'),
+        (LinkedTaskChannel.subscribe, '_from_aio'),
+    ],
+    ids=['msg-stream', 'linked-task-channel'],
+)
+def test_stream_subscribe_forwards_lag_policy(
+    subscribe,
+    chan_attr: str,
+) -> None:
+    '''
+    Stream wrappers must expose per-subscriber lag policy.
+
+    `MsgStream.subscribe()` and `LinkedTaskChannel.subscribe()`
+    previously omitted `BroadcastReceiver.raise_on_lag`, forcing
+    downstream users to mutate a private receiver attribute. Invoke
+    each public wrapper against a minimal receive-compatible handle.
+    Prove the first non-raising subscription configures both the
+    irreversible root broadcaster and its child, while a later strict
+    child selects its own policy without changing that root.
+
+    '''
+    class StreamHandle:
+        '''
+        Provide the wrapper fields needed for local fan-out.
+
+        '''
+        def __init__(self) -> None:
+            self._broadcaster = None
+            setattr(
+                self,
+                chan_attr,
+                SimpleNamespace(
+                    _state=SimpleNamespace(max_buffer_size=1),
+                ),
+            )
+
+        async def receive(self):
+            '''
+            Block if a regression unexpectedly enters source receive.
+
+            '''
+            await trio.sleep_forever()
+
+        async def send(self, value) -> None:
+            '''
+            Satisfy `MsgStream` duplex-handle patching.
+
+            '''
+
+    async def main() -> None:
+        stream = StreamHandle()
+        async with subscribe(
+            stream,
+            raise_on_lag=False,
+        ) as first:
+            assert not stream._broadcaster._raise_on_lag
+            assert not first._raise_on_lag
+
+        async with subscribe(
+            stream,
+            raise_on_lag=True,
+        ) as second:
+            assert not stream._broadcaster._raise_on_lag
+            assert second._raise_on_lag
+
+    trio.run(main)

tractor/_streaming.py

@@ -512,6 +512,7 @@ class MsgStream(trio.abc.Channel):
     @acm
     async def subscribe(
         self,
+        raise_on_lag: bool = True,
 
     ) -> AsyncIterator[BroadcastReceiver]:
         '''
@@ -526,6 +527,11 @@ class MsgStream(trio.abc.Channel):
         value from the far end via the internally created broudcast
         receiver wrapper.
 
+        ``raise_on_lag=False`` makes this subscription warn and resume
+        at the oldest retained value after an overrun. The first call
+        also sets that policy for this stream's root receive handle;
+        later child subscriptions choose their policy independently.
+
         '''
         # NOTE: This operation is indempotent and non-reversible, so be
         # sure you can deal with any (theoretical) overhead of the the
@@ -541,6 +547,7 @@ class MsgStream(trio.abc.Channel):
                 # TODO: can remove this kwarg right since
                 # by default behaviour is to do this anyway?
                 receive_afunc=self.receive,
+                raise_on_lag=raise_on_lag,
             )
 
             # NOTE: we override the original stream instance's receive
@@ -552,7 +559,9 @@ class MsgStream(trio.abc.Channel):
             # seems there's no graceful way to type this with ``mypy``?
             # https://github.com/python/mypy/issues/708
 
-        async with self._broadcaster.subscribe() as bstream:
+        async with self._broadcaster.subscribe(
+            raise_on_lag=raise_on_lag,
+        ) as bstream:
             assert bstream.key != self._broadcaster.key
             assert bstream._recv == self._broadcaster._recv
 

tractor/to_asyncio.py

@@ -324,6 +324,7 @@ class LinkedTaskChannel(
     @acm
     async def subscribe(
         self,
+        raise_on_lag: bool = True,
 
     ) -> AsyncIterator[BroadcastReceiver]:
         '''
@@ -335,6 +336,11 @@ class LinkedTaskChannel(
 
         See ``tractor._streaming.MsgStream.subscribe()`` for further
         similar details.
+
+        ``raise_on_lag=False`` makes this subscription warn and resume
+        at the oldest retained value after an overrun. The first call
+        also sets that policy for this channel's root receive handle;
+        later child subscriptions choose their policy independently.
         '''
         if self._broadcaster is None:
 
@@ -343,11 +349,14 @@ class LinkedTaskChannel(
                 # use memory channel size by default
                 self._from_aio._state.max_buffer_size,  # type: ignore
                 receive_afunc=self.receive,
+                raise_on_lag=raise_on_lag,
             )
 
             self.receive = bcast.receive  # type: ignore
 
-        async with self._broadcaster.subscribe() as bstream:
+        async with self._broadcaster.subscribe(
+            raise_on_lag=raise_on_lag,
+        ) as bstream:
             assert bstream.key != self._broadcaster.key
             assert bstream._recv == self._broadcaster._recv
             yield bstream