Reject concurrent `BroadcastReceiver.receive()` calls

Two tasks could receive through one `BroadcastReceiver` and race its
single `_wait_scope`. A later call could replace the scope, leaving
the first blocked through `.aclose()`.

Track the active `trio.lowlevel.Task` per receiver and reject an
overlapping call with `trio.BusyResourceError`. Include the owner's
task name and repr in the error, then clear ownership in `finally`.

Define each receiver as one logical subscription cursor. Concurrent
consumers must use separate `.subscribe()` receivers so each retains
an independent cursor and lag policy. Strict mode reads every retained
value or raises `Lagged`; `raise_on_lag=False` explicitly permits
drops.

Cover the regression with a checkpoint loop synchronized on
`recv_ready`; explain the interleaving inline, assert the error names
the active consumer, then release the source and prove the original
receive remains usable.

Prompt-IO: ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.md

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

ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.md

@@ -0,0 +1,47 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: pr490-landing-review
+timestamp: 2026-08-28T01:42:47Z
+git_ref: wkt/fix_broadcast_lag_count
+scope: code
+substantive: true
+raw_file: 20260828T014247Z_d06ee34b_prompt_io.raw.md
+---
+
+## Prompt
+
+Prime PR #490 for landing, review its concurrent broadcast changes,
+fix valid findings, and keep the landing pass focused.
+
+## Response summary
+
+Rejected an invalid mutable-default finding under `msgspec.Struct`
+semantics. Found and fixed a close-time race where concurrent receives
+on one `BroadcastReceiver` could overwrite its sole cancellation scope
+and strand an earlier waiter. The receiver now rejects overlapping
+calls with `trio.BusyResourceError`. Public docs now define each
+receiver as one logical subscription cursor and require one receiver
+per concurrent lossless consumer. Busy errors identify the active Trio
+task which owns the receiver.
+
+## Files changed
+
+- `tractor/trionics/_broadcast.py` - reject concurrent receives on one
+  receiver handle.
+- `tests/test_task_broadcasting.py` - prove the second call fails while
+  the original receive remains usable.
+- `docs/guide/streaming.rst` - document per-task receiver ownership and
+  explicit lag behavior.
+
+## Human edits
+
+The human selected PR #490 as the next landing target, rebased it onto
+current `main`, directed the agent to continue the landing pass, and
+required a faster, tightly scoped execution after the review expanded.
+The human then confirmed the per-receiver guard and required the public
+method, type, and surrounding guide to emphasize lossless per-consumer
+semantics. The human also required the regression's checkpoint loop to
+explain its synchronization and the busy error to report the active
+consumer task. The agent applied the source, test, and documentation
+changes; no direct manual edits to these files were identified.

ai/prompt-io/opencode/20260828T014247Z_d06ee34b_prompt_io.raw.md

@@ -0,0 +1,56 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-28T01:42:47Z
+git_ref: wkt/fix_broadcast_lag_count
+diff_cmd: git diff
+---
+
+## Prompt
+
+Get PR #490 primed and landed. Continue after rebasing, but keep the
+review and fix pass focused.
+
+## Response
+
+The mutable `{}` default is safe because `msgspec.Struct` creates a
+fresh dictionary per instance. A separate concurrency defect remains:
+two tasks can call `receive()` through one `BroadcastReceiver`, and the
+second call overwrites the only `_wait_scope`. Closing that receiver can
+then leave the first task blocked indefinitely.
+
+> `git diff -- tractor/trionics/_broadcast.py tests/test_task_broadcasting.py`
+
+Added a per-receiver in-progress guard that raises
+`trio.BusyResourceError` for an overlapping receive. Added a regression
+test that blocks the first receive, verifies the second call is
+rejected, and releases the source to prove the first call still works.
+The complete broadcast test file passes: 28 tests in 4.57 seconds.
+
+## Follow-up prompt
+
+Clarify that the guard is per receiver, preserves lossless consumer
+semantics, and document the contract on the public method, type, and
+surrounding guide.
+
+## Follow-up response
+
+> `git diff -- tractor/trionics/_broadcast.py docs/guide/streaming.rst`
+
+Documented that each `BroadcastReceiver` owns one logical cursor, each
+concurrent consumer needs its own subscribed receiver, overlapping
+calls on one handle raise `BusyResourceError`, and strict lag handling
+never skips values silently.
+
+## Second follow-up prompt
+
+Explain that the polling loop waits until the background receive is
+blocked before making the concurrent call, and include the first
+consumer task's runtime information in the busy exception.
+
+## Second follow-up response
+
+Replaced the boolean guard with the active `trio.lowlevel.Task`, added
+its name and representation to `BusyResourceError`, named the fixture
+task for a deterministic assertion, and documented the checkpoint-loop
+interleaving directly above the poll.

docs/guide/streaming.rst

@@ -171,6 +171,14 @@ 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.
 
+Each ``BroadcastReceiver`` is one logical subscription cursor, so
+give every concurrent consumer task its own receiver. Overlapping
+``receive()`` calls on the same handle raise
+``trio.BusyResourceError`` instead of racing that cursor. In strict
+mode values are never skipped silently: the consumer either reads
+each retained value in sequence or receives an explicit ``Lagged``
+error after exceeding the buffer window.
+
 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

tests/test_task_broadcasting.py

@@ -1018,6 +1018,52 @@ def test_closing_non_owner_preserves_source_wait() -> None:
     trio.run(main)
 
 
+def test_concurrent_receive_raises_busy() -> None:
+    '''
+    Reject concurrent receives on one broadcast handle.
+
+    A receiver stores one private peer-wait cancellation scope. If two
+    tasks receive through the same handle, the second task can replace
+    that scope and prevent `BroadcastReceiver.aclose()` from waking the
+    first task. Block one task in the shared source receive, then prove
+    a second call raises `BusyResourceError` before it can mutate any
+    per-receiver wait state. Releasing the source proves the original
+    receive remains usable.
+
+    '''
+    async def main() -> None:
+        tx, rx = trio.open_memory_channel(1)
+        brx = broadcast_receiver(rx, 1)
+        values: list[int] = []
+
+        async def receive() -> None:
+            values.append(await brx.receive())
+
+        async with trio.open_nursery() as nursery:
+            nursery.start_soon(
+                receive,
+                name='first broadcast consumer',
+            )
+
+            # Synchronize with the background task after it blocks in
+            # the shared source `.receive()`, ensuring the next call is
+            # concurrent with an already-active receive on this handle.
+            while brx._state.recv_ready is None:
+                await trio.lowlevel.checkpoint()
+
+            with pytest.raises(
+                trio.BusyResourceError,
+                match='first broadcast consumer',
+            ):
+                await brx.receive()
+
+            await tx.send(1)
+
+        assert values == [1]
+
+    trio.run(main)
+
+
 @pytest.mark.parametrize(
     'first_outcome',
     [

tractor/trionics/_broadcast.py

@@ -184,12 +184,18 @@ class BroadcastState(Struct):
 
 class BroadcastReceiver(ReceiveChannel):
     '''
-    A memory receive channel broadcaster which is non-lossy for
+    One logical subscriber to a shared receive-channel broadcast.
-    the fastest consumer.
 
-    Additional consumer tasks can receive all produced values by
+    Each instance owns one sequence cursor. Additional consumer tasks
-    registering with ``.subscribe()`` and receiving from the new
+    must call `.subscribe()` and receive through the new instance it
-    instance it delivers.
+    yields. Overlapping `.receive()` calls on the same instance raise
+    `trio.BusyResourceError` rather than racing that cursor or the
+    receiver's close-cancellation state.
+
+    A strict subscriber reads each retained value in sequence. Falling
+    behind the retention window raises `Lagged` instead of silently
+    losing values; `raise_on_lag=False` explicitly opts into dropping
+    displaced values.
 
     '''
     def __init__(
@@ -219,6 +225,7 @@ class BroadcastReceiver(ReceiveChannel):
         self._closed: bool = False
         self._raise_on_lag = raise_on_lag
         self._wait_scope: trio.CancelScope|None = None
+        self._receive_task: trio.lowlevel.Task|None = None
 
     def receive_nowait(
         self,
@@ -434,6 +441,30 @@ class BroadcastReceiver(ReceiveChannel):
             state.recv_scope = None
 
     async def receive(self) -> ReceiveType:
+        '''
+        Receive the next value for this subscriber's sequence cursor.
+
+        Only one task may receive through this instance at a time. Use
+        `.subscribe()` to give each concurrent consumer its own cursor
+        and loss/lag policy. `trio.BusyResourceError` identifies the
+        task which owns an already-active receive.
+
+        '''
+        if receive_task := self._receive_task:
+            raise trio.BusyResourceError(
+                'another task is already receiving from this '
+                '`BroadcastReceiver`\n'
+                f'active receive task: {receive_task.name!r}\n'
+                f'{receive_task!r}'
+            )
+
+        self._receive_task = trio.lowlevel.current_task()
+        try:
+            return await self._receive()
+        finally:
+            self._receive_task = None
+
+    async def _receive(self) -> ReceiveType:
         key = self.key
         state = self._state
 
@@ -518,11 +549,12 @@ class BroadcastReceiver(ReceiveChannel):
 
     ) -> AsyncIterator[BroadcastReceiver]:
         '''
-        Subscribe for values from this broadcast receiver.
+        Create a receiver with its own logical subscription cursor.
 
-        Returns a new ``BroadCastReceiver`` which is registered for and
+        The new `BroadcastReceiver` is registered against the shared
-        pulls data from a clone of the original
+        source and receives every retained value in sequence. Give each
-        ``trio.abc.ReceiveChannel`` provided at creation.
+        concurrent consumer task its own receiver instead of sharing
+        one instance across overlapping `.receive()` calls.
 
         '''
         if self._closed: