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

@@ -1,31 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-session: ses_0799212ebffe42arY96czXn89F
-timestamp: 2026-08-12T01:23:24Z
-git_ref: 06c4af17
-scope: code
-substantive: true
-raw_file: 20260812T012324Z_06c4af17_prompt_io.raw.md
----
-
-## Prompt
-
-Iteratively refine Tractor PR 490. For item one, correct broadcast statistics
-queue counts and Trio event checks, verify and review the exact change, then
-return a complete commit plan before proceeding.
-
-## Response summary
-
-Converted subscriber cursor indexes into clamped retained queue counts,
-removed deprecated event truthiness, and added deterministic state and
-deprecation regressions.
-
-## Files changed
-
-- `tractor/trionics/_broadcast.py` - accurate queue and waiter statistics.
-- `tests/test_task_broadcasting.py` - retained-count and event regression.
-
-## Human edits
-
-None - generated output follows the requested first iterative item.

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

@@ -1,37 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-timestamp: 2026-08-12T01:23:24Z
-git_ref: 06c4af17
-diff_cmd: git diff HEAD~1..HEAD
----
-
-After opening draft Tractor PR 490, the user requested an iterative pass over
-additional broadcast subsystem findings. The first item was to correct
-`BroadcastState.statistics()` queued counts and its deprecated Trio event
-truthiness check, then stop for a complete commit plan.
-
-> `git diff HEAD~1..HEAD -- tractor/trionics/_broadcast.py`
-
-Changed `queued_len_by_task` from raw deque cursor indexes to retained,
-receivable counts. Caught-up `-1` reports zero, valid indexes report index plus
-one, and lagged cursors clamp to the current retained queue length. Replaced
-`trio.Event` truthiness with an explicit `is not None` branch.
-
-> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py`
-
-Added a deterministic statistics regression using actual sends and receives.
-It verifies caught-up and one-queued states, drives a root receiver beyond a
-three-slot retention window to prove clamping, and installs a real
-`trio.Event` while treating deprecations as errors.
-
-Verification output:
-
-```text
-...........                                                              [100%]
-11 passed in 5.76s
-```
-
-Python compilation and `git diff --check` passed. Initial adversarial review
-caught unclamped lagged cursors and an ineffective event test; both were
-fixed. Final review found no actionable issues.

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

@@ -1,33 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-session: ses_0799212ebffe42arY96czXn89F
-timestamp: 2026-08-12T03:06:08Z
-git_ref: 1095e7f7
-scope: code
-substantive: true
-raw_file: 20260812T030608Z_1095e7f7_prompt_io.raw.md
----
-
-## Prompt
-
-For Tractor PR 490 item two, make shared underlying receive failures wake and
-terminate every broadcast subscriber without losing retained values. Review,
-verify and return a complete commit plan before continuing.
-
-## Response summary
-
-Published ordinary receive failures as terminal broadcast state, introduced
-a public chained peer exception, kept control-flow exits transient while
-waking peers, documented the contract, and added deterministic regressions.
-
-## Files changed
-
-- `tractor/trionics/_broadcast.py` - terminal failure and peer wake protocol.
-- `tractor/trionics/__init__.py` - public peer exception export.
-- `docs/api/trionics.rst` - failure-delivery API contract.
-- `tests/test_task_broadcasting.py` - terminal and transient failure tests.
-
-## Human edits
-
-None - generated output follows the requested second iterative item.

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

@@ -1,49 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-timestamp: 2026-08-12T03:06:08Z
-git_ref: 1095e7f7
-diff_cmd: git diff HEAD~1..HEAD
----
-
-The user requested the second iterative refinement for Tractor PR 490: ensure
-non-EOC failures from a shared underlying broadcast receiver do not leave peer
-subscribers blocked forever, then stop for a complete commit plan.
-
-> `git diff HEAD~1..HEAD -- tractor/trionics/_broadcast.py`
-
-Added shared terminal failure publication for ordinary `Exception` values.
-The receive owner gets the original exception; peers may drain retained
-values and then get a fresh `BroadcastReceiveError` chained from the original.
-Late subscribers observe the same terminal state without retrying the failed
-underlying receiver. Process-control and cancellation-like `BaseException`
-values wake peers but are re-raised without becoming durable channel state.
-
-> `git diff HEAD~1..HEAD -- tractor/trionics/__init__.py`
-
-Exported `BroadcastReceiveError` as the public peer-delivery exception.
-
-> `git diff HEAD~1..HEAD -- docs/api/trionics.rst`
-
-Documented `BroadcastReceiveError` and the owner-versus-peer delivery
-contract, including retained-value draining and late subscribers.
-
-> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py`
-
-Added deterministic bounded regressions. One scripts a successful receive
-followed by `RuntimeError`, proving the root drains retained data, all current
-and late receivers observe terminal failure, and the source is not retried.
-The second scripts a custom `BaseException`, proving peers wake and take over
-the next source receive without retaining control-flow state.
-
-Verification output:
-
-```text
-.............                                                            [100%]
-13 passed in 5.62s
-```
-
-Python compilation and `git diff --check` passed. Iterative adversarial review
-drove independent peer exception wrappers, ordinary-versus-control-flow
-classification, bounded test completion, public docs, and the final catch-all
-peer wake. Final review found no issues.

docs/api/trionics.rst

@@ -40,9 +40,6 @@ Broadcast fan-out
 .. autoexception:: Lagged
    :show-inheritance:
 
-.. autoexception:: BroadcastReceiveError
-   :show-inheritance:
-
 A single-producer, many-consumer broadcast layer over any
 ``trio``-style receive channel: non-lossy for the *fastest*
 consumer while slower consumers raise :class:`Lagged` (a
@@ -51,13 +48,6 @@ internal ring. This is exactly the machinery behind
 :meth:`tractor.MsgStream.subscribe` — see
 ``examples/streaming_broadcast_fanout.py``.
 
-If the shared underlying receiver raises an ordinary exception, the
-subscriber which owned that receive gets the original failure.
-Waiting peers drain their retained values and then raise
-:class:`BroadcastReceiveError`, with the original failure available
-as ``__cause__``. Later subscribers observe the same terminal state
-without retrying the failed underlying receiver.
-
 ExceptionGroup helpers
 ----------------------
 

tests/test_task_broadcasting.py

@@ -9,7 +9,6 @@ from functools import partial
 from itertools import cycle
 import time
 from typing import Optional
-import warnings
 
 import pytest
 import trio
@@ -17,7 +16,6 @@ from trio.lowlevel import current_task
 import tractor
 from tractor.trionics import (
     broadcast_receiver,
-    BroadcastReceiveError,
     Lagged,
     collapse_eg,
 )
@@ -351,243 +349,6 @@ def test_lagged_reports_exact_drop_count(
     trio.run(main)
 
 
-def test_broadcast_statistics_report_queued_counts() -> None:
-    '''
-    `BroadcastState.statistics()` must report counts, not indexes.
-
-    Each `BroadcastState.subs` value is the deque index of a
-    receiver's next unread value, with `-1` meaning caught up. The
-    statistics API returned these indexes directly, so one queued
-    value appeared as zero and every positive count was one short.
-    Keep one root receiver idle while a child synchronously receives
-    four produced values. Prove the root count advances through one
-    and three retained values, then remains clamped to the three-slot
-    retention window after lagging.
-
-    Finally install an actual unwaited `trio.Event` in
-    `BroadcastState.recv_ready` while treating deprecations as errors.
-    This proves statistics checks `None` explicitly instead of using
-    deprecated `trio.Event` truthiness.
-
-    '''
-    async def main() -> None:
-        tx, rx = trio.open_memory_channel(3)
-        brx = broadcast_receiver(rx, 3)
-
-        async with brx.subscribe() as child:
-            state = brx._state
-            assert state.statistics()['queued_len_by_task'] == {
-                brx.key: 0,
-                child.key: 0,
-            }
-
-            await tx.send(0)
-            assert await child.receive() == 0
-            assert state.statistics()['queued_len_by_task'] == {
-                brx.key: 1,
-                child.key: 0,
-            }
-
-            for value in range(1, 4):
-                await tx.send(value)
-                assert await child.receive() == value
-
-            state.recv_ready = (child.key, trio.Event())
-            with warnings.catch_warnings():
-                warnings.simplefilter('error', DeprecationWarning)
-                stats = state.statistics()
-
-            assert stats['queued_len_by_task'] == {
-                brx.key: 3,
-                child.key: 0,
-            }
-            assert stats['tasks_waiting'] == 0
-
-    trio.run(main)
-
-
-def test_underlying_receive_failure_wakes_all_subscribers() -> None:
-    '''
-    A shared receive failure must terminate every broadcast receiver.
-
-    Previously, only `EndOfChannel` and receiver cancellation woke
-    peer tasks waiting on `BroadcastState.recv_ready`. If the shared
-    underlying receiver raised another error, its owner propagated
-    the failure and cleared the event while every peer remained
-    blocked forever.
-
-    Script one successful receive followed by a controlled
-    `RuntimeError`. Let a fast child own both underlying receives
-    while the root first drains its retained value and then waits on
-    the child's second receive. Release the failure only after both
-    tasks have reached those positions. Both exact errors prove the
-    peer was awakened without losing buffered data. A later
-    subscriber proves the terminal failure remains published for new
-    receivers instead of retrying the failed underlying channel.
-
-    '''
-    class FailingReceiver:
-        '''
-        Return one value, then fail after deterministic release.
-
-        '''
-        def __init__(self) -> None:
-            self.calls: int = 0
-            self.failure_started = trio.Event()
-            self.release_failure = trio.Event()
-
-        async def receive(self) -> int:
-            '''
-            Drive the scripted success-then-failure sequence.
-
-            '''
-            self.calls += 1
-            if self.calls == 1:
-                return 1
-
-            self.failure_started.set()
-            await self.release_failure.wait()
-            raise RuntimeError('underlying receive failed')
-
-    async def main() -> None:
-        source = FailingReceiver()
-        brx = broadcast_receiver(source, 3)
-        child_error: list[RuntimeError] = []
-        root_error: list[BroadcastReceiveError] = []
-        late_error: list[BroadcastReceiveError] = []
-        root_drained = trio.Event()
-
-        async def receive_child() -> None:
-            async with brx.subscribe() as child:
-                assert await child.receive() == 1
-                try:
-                    await child.receive()
-                except RuntimeError as exc:
-                    child_error.append(exc)
-
-        async def receive_root() -> None:
-            assert await brx.receive() == 1
-            root_drained.set()
-            try:
-                await brx.receive()
-            except BroadcastReceiveError as exc:
-                root_error.append(exc)
-
-        with trio.fail_after(1):
-            async with trio.open_nursery() as nursery:
-                nursery.start_soon(receive_child)
-                await source.failure_started.wait()
-
-                nursery.start_soon(receive_root)
-                await root_drained.wait()
-
-                source.release_failure.set()
-
-        assert source.calls == 2
-        assert [str(exc) for exc in child_error] == [
-            'underlying receive failed',
-        ]
-        assert [str(exc) for exc in root_error] == [
-            'Shared broadcast receiver failed',
-        ]
-        assert child_error[0] is not root_error[0]
-        assert root_error[0].__cause__ is child_error[0]
-
-        async with brx.subscribe() as late:
-            with pytest.raises(
-                BroadcastReceiveError,
-                match='Shared broadcast receiver failed',
-            ) as exc_info:
-                await late.receive()
-            late_error.append(exc_info.value)
-        assert late_error[0] is not child_error[0]
-        assert late_error[0] is not root_error[0]
-        assert late_error[0].__cause__ is child_error[0]
-        assert source.calls == 2
-
-    trio.run(main)
-
-
-def test_control_flow_exit_wakes_broadcast_peer() -> None:
-    '''
-    Non-terminal control flow must wake peers without being retained.
-
-    Process-control and cancellation-like `BaseException` values
-    should remain local to the task which receives them, but the old
-    owner still has to wake subscribers blocked on its shared event.
-    Make one child own a controlled `BaseException` receive while the
-    root waits behind it. After release, prove the child gets that
-    exact exit and the root takes ownership of the next underlying
-    receive instead of hanging or replaying the control-flow event.
-
-    '''
-    class ReceiveExit(BaseException):
-        '''
-        Model a non-terminal process-control receive exit.
-
-        '''
-
-    class ControlFlowReceiver:
-        '''
-        Raise one controlled exit, then return a value.
-
-        '''
-        def __init__(self) -> None:
-            self.calls: int = 0
-            self.exit_started = trio.Event()
-            self.release_exit = trio.Event()
-
-        async def receive(self) -> int:
-            '''
-            Drive the scripted control-flow-then-value sequence.
-
-            '''
-            self.calls += 1
-            if self.calls == 1:
-                self.exit_started.set()
-                await self.release_exit.wait()
-                raise ReceiveExit
-
-            return 2
-
-    async def main() -> None:
-        source = ControlFlowReceiver()
-        brx = broadcast_receiver(source, 3)
-        child_exit: list[ReceiveExit] = []
-        root_value: list[int] = []
-
-        async def receive_child() -> None:
-            async with brx.subscribe() as child:
-                try:
-                    await child.receive()
-                except ReceiveExit as exc:
-                    child_exit.append(exc)
-
-        async def receive_root() -> None:
-            root_value.append(await brx.receive())
-
-        with trio.fail_after(1):
-            async with trio.open_nursery() as nursery:
-                nursery.start_soon(receive_child)
-                await source.exit_started.wait()
-                nursery.start_soon(receive_root)
-
-                while True:
-                    _, event = brx._state.recv_ready
-                    if event.statistics().tasks_waiting:
-                        break
-                    await trio.lowlevel.checkpoint()
-
-                source.release_exit.set()
-
-        assert len(child_exit) == 1
-        assert root_value == [2]
-        assert source.calls == 2
-        assert brx._state.receive_exc is None
-
-    trio.run(main)
-
-
 def test_ensure_slow_consumers_lag_out(
     reg_addr,
     start_method,

tractor/trionics/__init__.py

@@ -26,7 +26,6 @@ from ._mngrs import (
 from ._broadcast import (
     AsyncReceiver as AsyncReceiver,
     broadcast_receiver as broadcast_receiver,
-    BroadcastReceiveError as BroadcastReceiveError,
     BroadcastReceiver as BroadcastReceiver,
     Lagged as Lagged,
 )

tractor/trionics/_broadcast.py

@@ -100,13 +100,6 @@ class Lagged(trio.TooSlowError):
     '''
 
 
-class BroadcastReceiveError(Exception):
-    '''
-    A shared underlying receiver failed in another subscriber task.
-
-    '''
-
-
 class BroadcastState(Struct):
     '''
     Common state to all receivers of a broadcast.
@@ -129,11 +122,6 @@ class BroadcastState(Struct):
     # For now, this is solely for testing/debugging purposes.
     eoc: bool = False
 
-    # Any non-EOC failure from the shared underlying receiver is
-    # terminal for every subscriber. Retained values remain readable
-    # before this failure is replayed at each receiver's boundary.
-    receive_exc: Exception | None = None
-
     # If the broadcaster was cancelled, we might as well track it
     cancelled: dict[int, Task] = {}
 
@@ -154,20 +142,13 @@ class BroadcastState(Struct):
 
         qlens: dict[int, int] = {}
         for tid, sz in subs.items():
-            qlens[tid] = min(
+            qlens[tid] = sz if sz != -1 else 0
-                sz + 1,
-                len(self.queue),
-            )
 
         return {
             'open_consumers': len(subs),
             'queued_len_by_task': qlens,
             'max_buffer_size': self.maxlen,
-            'tasks_waiting': (
+            'tasks_waiting': ev.statistics().tasks_waiting if ev else 0,
-                ev.statistics().tasks_waiting
-                if ev is not None
-                else 0
-            ),
             'tasks_cancelled': self.cancelled,
             'next_value_receiver_id': key,
         }
@@ -279,15 +260,6 @@ class BroadcastReceiver(ReceiveChannel):
             state.subs[key] -= 1
             return value
 
-        receive_exc = state.receive_exc
-        if receive_exc is not None:
-            # Re-raising one shared exception mutates its traceback on
-            # every delivery. Give each receiver a stable wrapper while
-            # retaining the original failure as its cause.
-            raise BroadcastReceiveError(
-                'Shared broadcast receiver failed'
-            ) from receive_exc
-
         raise trio.WouldBlock
 
     async def _receive_from_underlying(
@@ -360,24 +332,6 @@ class BroadcastReceiver(ReceiveChannel):
                 event.set()
             raise
 
-        except Exception as receive_exc:
-            # The underlying receiver is shared by every subscriber,
-            # so any non-EOC failure terminates the entire broadcast.
-            # Publish it before waking peers so they can drain their
-            # retained values and then observe the same failure.
-            state.receive_exc = receive_exc
-            if event.statistics().tasks_waiting:
-                event.set()
-            raise
-
-        except BaseException:
-            # Process-control and cancellation-like exceptions must
-            # not become durable broadcast state, but peers still
-            # need waking before `recv_ready` is cleared.
-            if event.statistics().tasks_waiting:
-                event.set()
-            raise
-
         finally:
             # Reset receiver waiter task event for next blocking condition.
             # this MUST be reset even if the above ``.recv()`` call