diff --git a/ai/prompt-io/opencode/20260812T012324Z_06c4af17_prompt_io.md b/ai/prompt-io/opencode/20260812T012324Z_06c4af17_prompt_io.md new file mode 100644 index 00000000..c6af20c1 --- /dev/null +++ b/ai/prompt-io/opencode/20260812T012324Z_06c4af17_prompt_io.md @@ -0,0 +1,31 @@ +--- +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. diff --git a/ai/prompt-io/opencode/20260812T012324Z_06c4af17_prompt_io.raw.md b/ai/prompt-io/opencode/20260812T012324Z_06c4af17_prompt_io.raw.md new file mode 100644 index 00000000..f2da978c --- /dev/null +++ b/ai/prompt-io/opencode/20260812T012324Z_06c4af17_prompt_io.raw.md @@ -0,0 +1,37 @@ +--- +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. diff --git a/tests/test_task_broadcasting.py b/tests/test_task_broadcasting.py index 20965cbe..b5d0d8aa 100644 --- a/tests/test_task_broadcasting.py +++ b/tests/test_task_broadcasting.py @@ -9,6 +9,7 @@ from functools import partial from itertools import cycle import time from typing import Optional +import warnings import pytest import trio @@ -349,6 +350,61 @@ 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_ensure_slow_consumers_lag_out( reg_addr, start_method, diff --git a/tractor/trionics/_broadcast.py b/tractor/trionics/_broadcast.py index 4bc2e3b8..24aeb24f 100644 --- a/tractor/trionics/_broadcast.py +++ b/tractor/trionics/_broadcast.py @@ -142,13 +142,20 @@ class BroadcastState(Struct): qlens: dict[int, int] = {} for tid, sz in subs.items(): - qlens[tid] = sz if sz != -1 else 0 + qlens[tid] = min( + sz + 1, + len(self.queue), + ) return { 'open_consumers': len(subs), 'queued_len_by_task': qlens, 'max_buffer_size': self.maxlen, - 'tasks_waiting': ev.statistics().tasks_waiting if ev else 0, + 'tasks_waiting': ( + ev.statistics().tasks_waiting + if ev is not None + else 0 + ), 'tasks_cancelled': self.cancelled, 'next_value_receiver_id': key, }