Pass `Bindspace` FD to `trio` children

Thread an optional live `Bindspace` through actor spawn APIs and
give each Trio exec child an inherited namespace descriptor before
runtime bootstrap.

Deats,
- duplicate the namespace FD without changing parent ownership
- preserve caller `pass_fds` and process options
- send the `(fd, inode)` pair through the `_child` CLI
- close the parent duplicate on success, failure and cancellation
- reject MP backends until descriptor reduction is implemented
- avoid cleanup before child publication
- exercise real E2E namespace relay through handshake and RPC
- verify parent FD-table cleanup after successful child spawn

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

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

.claude/skills/run-tests/SKILL.md

@@ -1,632 +0,0 @@
----
-name: run-tests
-description: >
-  Run tractor test suite (or subsets). Use when the user wants
-  to run tests, verify changes, or check for regressions.
-argument-hint: "[test-path-or-pattern] [--opts]"
-allowed-tools:
-  - Bash(python -m pytest *)
-  - Bash(python -c *)
-  - Bash(python --version *)
-  - Bash(UV_PROJECT_ENVIRONMENT=py* uv run python *)
-  - Bash(UV_PROJECT_ENVIRONMENT=py* uv run pytest *)
-  - Bash(UV_PROJECT_ENVIRONMENT=py* uv sync *)
-  - Bash(UV_PROJECT_ENVIRONMENT=py* uv pip show *)
-  - Bash(git rev-parse *)
-  - Bash(ls *)
-  - Bash(cat *)
-  - Bash(jq * .pytest_cache/*)
-  # process inspection + SIGINT-first cleanup ladder (see
-  # the zombie-actor pre-flight / teardown steps below).
-  - Bash(ss *)
-  - Bash(pgrep *)
-  - Bash(pkill *)
-  - Bash(sleep *)
-  - Bash(rm -f /tmp/registry@*.sock)
-  - Read
-  - Grep
-  - Glob
-  - Task
-  - AskUserQuestion
----
-
-Run the `tractor` test suite using `pytest`. Follow this
-process:
-
-## 1. Parse user intent
-
-From the user's message and any arguments, determine:
-
-- **scope**: full suite, specific file(s), specific
-  test(s), or a keyword pattern (`-k`).
-- **transport**: which IPC transport protocol to test
-  against (default: `tcp`, also: `uds`).
-- **options**: any extra pytest flags the user wants
-  (e.g. `--ll debug`, `--tpdb`, `-x`, `-v`).
-
-If the user provides a bare path or pattern as argument,
-treat it as the test target. Examples:
-
-- `/run-tests` → full suite
-- `/run-tests test_local.py` → single file
-- `/run-tests test_registrar -v` → file + verbose
-- `/run-tests -k cancel` → keyword filter
-- `/run-tests tests/ipc/ --tpt-proto uds` → subdir + UDS
-
-## 2. Construct the pytest command
-
-Base command:
-```
-python -m pytest
-```
-
-### Default flags (always include unless user overrides):
-- `-x` (stop on first failure)
-- `--tb=short` (concise tracebacks)
-- `--no-header` (reduce noise)
-
-### Path resolution:
-- If the user gives a bare filename like `test_local.py`,
-  resolve it under `tests/`.
-- If the user gives a subdirectory like `ipc/`, resolve
-  under `tests/ipc/`.
-- Glob if needed: `tests/**/test_*<pattern>*.py`
-
-### Key pytest options for this project:
-
-| Flag | Purpose |
-|---|---|
-| `--ll <level>` | Set tractor log level (e.g. `debug`, `info`, `runtime`) |
-| `--tpdb` / `--debug-mode` | Enable tractor's multi-proc debugger |
-| `--tpt-proto <key>` | IPC transport: `tcp` (default) or `uds` |
-| `--spawn-backend <be>` | Spawn method: `trio` (default), `mp_spawn`, `mp_forkserver` |
-| `-k <expr>` | pytest keyword filter |
-| `-v` / `-vv` | Verbosity |
-| `-s` | No output capture (useful with `--tpdb`) |
-
-### Common combos:
-```sh
-# quick smoke test of core modules
-python -m pytest tests/test_local.py tests/test_rpc.py -x --tb=short --no-header
-
-# full suite, stop on first failure
-python -m pytest tests/ -x --tb=short --no-header
-
-# specific test with debug
-python -m pytest tests/discovery/test_registrar.py::test_reg_then_unreg -x -s --tpdb --ll debug
-
-# run with UDS transport
-python -m pytest tests/ -x --tb=short --no-header --tpt-proto uds
-
-# keyword filter
-python -m pytest tests/ -x --tb=short --no-header -k "cancel and not slow"
-```
-
-## 3. Pre-flight: venv detection (MANDATORY)
-
-**Always verify a `uv` venv is active before running
-`python` or `pytest`.** This project uses
-`UV_PROJECT_ENVIRONMENT=py<MINOR>` naming (e.g.
-`py313`) — never `.venv`.
-
-### Step 1: detect active venv
-
-Run this check first:
-
-```sh
-python -c "
-import sys, os
-venv = os.environ.get('VIRTUAL_ENV', '')
-prefix = sys.prefix
-print(f'VIRTUAL_ENV={venv}')
-print(f'sys.prefix={prefix}')
-print(f'executable={sys.executable}')
-"
-```
-
-### Step 2: interpret results
-
-**Case A — venv is active** (`VIRTUAL_ENV` is set
-and points to a `py<MINOR>/` dir under the project
-root or worktree):
-
-Use bare `python` / `python -m pytest` for all
-commands. This is the normal, fast path.
-
-**Case B — no venv active** (`VIRTUAL_ENV` is empty
-or `sys.prefix` points to a system Python):
-
-Use `AskUserQuestion` to ask the user:
-
-> "No uv venv is active. Should I activate one
-> via `UV_PROJECT_ENVIRONMENT=py<MINOR> uv sync`,
-> or would you prefer to activate your shell venv
-> first?"
-
-Options:
-1. **"Create/sync venv"** — run
-   `UV_PROJECT_ENVIRONMENT=py<MINOR> uv sync` where
-   `<MINOR>` is detected from `python --version`
-   (e.g. `313` for 3.13). Then use
-   `py<MINOR>/bin/python` for all subsequent
-   commands in this session.
-2. **"I'll activate it myself"** — stop and let the
-   user `source py<MINOR>/bin/activate` or similar.
-
-**Case C — inside a git worktree** (`git rev-parse
---git-common-dir` differs from `--git-dir`):
-
-Verify Python resolves from the **worktree's own
-venv**, not the main repo's:
-
-```sh
-python -c "import tractor; print(tractor.__file__)"
-```
-
-If the path points outside the worktree, create a
-worktree-local venv:
-
-```sh
-UV_PROJECT_ENVIRONMENT=py<MINOR> uv sync
-```
-
-Then use `py<MINOR>/bin/python` for all commands.
-
-**Why this matters**: without the correct venv,
-subprocesses spawned by tractor resolve modules
-from the wrong editable install, causing spurious
-`AttributeError` / `ModuleNotFoundError`.
-
-### Fallback: `uv run`
-
-If the user can't or won't activate a venv, all
-`python` and `pytest` commands can be prefixed
-with `UV_PROJECT_ENVIRONMENT=py<MINOR> uv run`:
-
-```sh
-# instead of: python -m pytest tests/ -x
-UV_PROJECT_ENVIRONMENT=py313 uv run pytest tests/ -x
-
-# instead of: python -c 'import tractor'
-UV_PROJECT_ENVIRONMENT=py313 uv run python -c 'import tractor'
-```
-
-`uv run` auto-discovers the project and venv,
-but is slower than a pre-activated venv due to
-lock-file resolution on each invocation. Prefer
-activating the venv when possible.
-
-### Step 3: import + collection checks
-
-After venv is confirmed, always run these
-(especially after refactors or module moves):
-
-```sh
-# 1. package import smoke check
-python -c 'import tractor; print(tractor)'
-
-# 2. verify all tests collect (no import errors)
-python -m pytest tests/ -x -q --co 2>&1 | tail -5
-```
-
-If either fails, fix the import error before running
-any actual tests.
-
-### Step 4: zombie-actor / stale-registry check (MANDATORY)
-
-The tractor runtime's default registry address is
-**`127.0.0.1:1616`** (TCP) / `/tmp/registry@1616.sock`
-(UDS). Whenever any prior test run — especially one
-using a fork-based backend like `subint_forkserver` —
-leaks a child actor process, that zombie keeps the
-registry port bound and **every subsequent test
-session fails to bind**, often presenting as 50+
-unrelated failures ("all tests broken"!) across
-backends.
-
-**This has to be checked before the first run AND
-after any cancelled/SIGINT'd run** — signal failures
-in the middle of a test can leave orphan children.
-
-```sh
-# 1. TCP registry — any listener on :1616? (primary signal)
-ss -tlnp 2>/dev/null | grep ':1616' || echo 'TCP :1616 free'
-
-# 2. leftover actor/forkserver procs — scoped to THIS
-#    repo's python path, so we don't false-flag legit
-#    long-running tractor-using apps (e.g. `piker`,
-#    downstream projects that embed tractor).
-pgrep -af "$(pwd)/py[0-9]*/bin/python.*_actor_child_main|subint-forkserv" \
-  | grep -v 'grep\|pgrep' \
-  || echo 'no leaked actor procs from this repo'
-
-# 3. stale UDS registry sockets
-ls -la /tmp/registry@*.sock 2>/dev/null \
-  || echo 'no leaked UDS registry sockets'
-```
-
-**Interpretation:**
-
-- **TCP :1616 free AND no stale sockets** → clean,
-  proceed. The actor-procs probe is secondary — false
-  positives are common (piker, any other tractor-
-  embedding app); only cleanup if `:1616` is bound or
-  sockets linger.
-- **TCP :1616 bound OR stale sockets present** →
-  surface PIDs + cmdlines to the user, offer cleanup:
-
-  ```sh
-  # 1. GRACEFUL FIRST (tractor is structured concurrent — it
-  #    catches SIGINT as an OS-cancel in `_trio_main` and
-  #    cascades Portal.cancel_actor via IPC to every descendant.
-  #    So always try SIGINT first with a bounded timeout; only
-  #    escalate to SIGKILL if graceful cleanup doesn't complete).
-  pkill -INT -f "$(pwd)/py[0-9]*/bin/python.*_actor_child_main|subint-forkserv"
-
-  # 2. bounded wait for graceful teardown (usually sub-second).
-  #    Loop until the processes exit, or timeout. Keep the
-  #    bound tight — hung/abrupt-killed descendants usually
-  #    hang forever, so don't wait more than a few seconds.
-  for i in $(seq 1 10); do
-    pgrep -f "$(pwd)/py[0-9]*/bin/python.*_actor_child_main|subint-forkserv" >/dev/null || break
-    sleep 0.3
-  done
-
-  # 3. ESCALATE TO SIGKILL only if graceful didn't finish.
-  if pgrep -f "$(pwd)/py[0-9]*/bin/python.*_actor_child_main|subint-forkserv" >/dev/null; then
-    echo 'graceful teardown timed out — escalating to SIGKILL'
-    pkill -9 -f "$(pwd)/py[0-9]*/bin/python.*_actor_child_main|subint-forkserv"
-  fi
-
-  # 4. if a test zombie holds :1616 specifically and doesn't
-  #    match the above pattern, find its PID the hard way:
-  ss -tlnp 2>/dev/null | grep ':1616'   # prints `users:(("<name>",pid=NNNN,...))`
-  # then (same SIGINT-first ladder):
-  # kill -INT <NNNN>; sleep 1; kill -9 <NNNN> 2>/dev/null
-
-  # 5. remove stale UDS sockets
-  rm -f /tmp/registry@*.sock
-
-  # 6. re-verify
-  ss -tlnp 2>/dev/null | grep ':1616' || echo 'TCP :1616 now free'
-  ```
-
-**Never ignore stale registry state.** If you see the
-"all tests failing" pattern — especially
-`trio.TooSlowError` / connection refused / address in
-use on many unrelated tests — check registry **before**
-spelunking into test code. The failure signature will
-be identical across backends because they're all
-fighting for the same port.
-
-**False-positive warning for step 2:** a plain
-`pgrep -af '_actor_child_main'` will also match
-legit long-running tractor-embedding apps (e.g.
-`piker` at `~/repos/piker/py*/bin/python3 -m
-tractor._child ...`). Always scope to the current
-repo's python path, or only use step 1 (`:1616`) as
-the authoritative signal.
-
-## 4. Run and report
-
-- Run the constructed command.
-- Use a timeout of **600000ms** (10min) for full suite
-  runs, **120000ms** (2min) for single-file runs.
-- If the suite is large (full `tests/`), consider running
-  in the background and checking output when done.
-- Use `--lf` (last-failed) to re-run only previously
-  failing tests when iterating on a fix.
-
-### On failure:
-- Show the failing test name(s) and short traceback.
-- If the failure looks related to recent changes, point
-  out the likely cause and suggest a fix.
-- **Check the known-flaky list** (section 8) before
-  investigating — don't waste time on pre-existing
-  timeout issues.
-- **NEVER auto-commit fixes.** If you apply a code fix
-  during test iteration, leave it unstaged. Tell the
-  user what changed and suggest they review the
-  worktree state, stage files manually, and use
-  `/commit-msg` (inline or in a separate session) to
-  generate the commit message. The human drives all
-  `git add` and `git commit` operations.
-
-### On success:
-- Report the pass/fail/skip counts concisely.
-
-## 5. Test directory layout (reference)
-
-```
-tests/
-├── conftest.py          # root fixtures, daemon, signals
-├── devx/                # debugger/tooling tests
-├── ipc/                 # transport protocol tests
-├── msg/                 # messaging layer tests
-├── discovery/           # discovery subsystem tests
-│   ├── test_multiaddr.py  # multiaddr construction
-│   └── test_registrar.py  # registry/discovery protocol
-├── test_local.py        # registrar + local actor basics
-├── test_rpc.py          # RPC error handling
-├── test_spawning.py     # subprocess spawning
-├── test_multi_program.py  # multi-process tree tests
-├── test_cancellation.py # cancellation semantics
-├── test_context_stream_semantics.py  # ctx streaming
-├── test_inter_peer_cancellation.py   # peer cancel
-├── test_infected_asyncio.py  # trio-in-asyncio
-└── ...
-```
-
-## 6. Change-type → test mapping
-
-After modifying specific modules, run the corresponding
-test subset first for fast feedback:
-
-| Changed module(s) | Run these tests first |
-|---|---|
-| `runtime/_runtime.py`, `runtime/_state.py` | `test_local.py test_rpc.py test_spawning.py test_root_runtime.py` |
-| `discovery/` (`_registry`, `_discovery`, `_addr`) | `tests/discovery/ test_multi_program.py test_local.py` |
-| `_context.py`, `_streaming.py` | `test_context_stream_semantics.py test_advanced_streaming.py` |
-| `ipc/` (`_chan`, `_server`, `_transport`) | `tests/ipc/ test_2way.py` |
-| `runtime/_portal.py`, `runtime/_rpc.py` | `test_rpc.py test_cancellation.py` |
-| `spawn/` (`_spawn`, `_entry`) | `test_spawning.py test_multi_program.py` |
-| `devx/debug/` | `tests/devx/test_debugger.py` (slow!) |
-| `to_asyncio.py` | `test_infected_asyncio.py test_root_infect_asyncio.py` |
-| `msg/` | `tests/msg/` |
-| `_exceptions.py` | `test_remote_exc_relay.py test_inter_peer_cancellation.py` |
-| `runtime/_supervise.py` | `test_cancellation.py test_spawning.py` |
-
-## 7. Quick-check shortcuts
-
-### After refactors (fastest first-pass):
-```sh
-# import + collect check
-python -c 'import tractor' && python -m pytest tests/ -x -q --co 2>&1 | tail -3
-
-# core subset (~10s)
-python -m pytest tests/test_local.py tests/test_rpc.py tests/test_spawning.py tests/discovery/test_registrar.py -x --tb=short --no-header
-```
-
-### Inspect last failures (without re-running):
-
-When the user asks "what failed?", "show failures",
-or wants to check the last-failed set before
-re-running — read the pytest cache directly. This
-is instant and avoids test collection overhead.
-
-```sh
-python -c "
-import json, pathlib, sys
-p = pathlib.Path('.pytest_cache/v/cache/lastfailed')
-if not p.exists():
-    print('No lastfailed cache found.'); sys.exit()
-data = json.loads(p.read_text())
-# filter to real test node IDs (ignore junk
-# entries that can accumulate from system paths)
-tests = sorted(k for k in data if k.startswith('tests/'))
-if not tests:
-    print('No failures recorded.')
-else:
-    print(f'{len(tests)} last-failed test(s):')
-    for t in tests:
-        print(f'  {t}')
-"
-```
-
-**Why not `--cache-show` or `--co --lf`?**
-
-- `pytest --cache-show 'cache/lastfailed'` works
-  but dumps raw dict repr including junk entries
-  (stale system paths that leak into the cache).
-- `pytest --co --lf` actually *collects* tests which
-  triggers import resolution and is slow (~0.5s+).
-  Worse, when cached node IDs don't exactly match
-  current parametrize IDs (e.g. param names changed
-  between runs), pytest falls back to collecting
-  the *entire file*, giving false positives.
-- Reading the JSON directly is instant, filterable
-  to `tests/`-prefixed entries, and shows exactly
-  what pytest recorded — no interpretation.
-
-**After inspecting**, re-run the failures:
-```sh
-python -m pytest --lf -x --tb=short --no-header
-```
-
-### Full suite in background:
-When core tests pass and you want full coverage while
-continuing other work, run in background:
-```sh
-python -m pytest tests/ -x --tb=short --no-header -q
-```
-(use `run_in_background=true` on the Bash tool)
-
-## 8. Known flaky tests
-
-These tests have **pre-existing** timing/environment
-sensitivity. If they fail with `TooSlowError` or
-pexpect `TIMEOUT`, they are almost certainly NOT caused
-by your changes — note them and move on.
-
-| Test | Typical error | Notes |
-|---|---|---|
-| `devx/test_debugger.py::test_multi_nested_subactors_error_through_nurseries` | pexpect TIMEOUT | Debugger pexpect timing |
-| `test_cancellation.py::test_cancel_via_SIGINT_other_task` | TooSlowError | Signal handling race |
-| `test_inter_peer_cancellation.py::test_peer_spawns_and_cancels_service_subactor` | TooSlowError | Async timing (both param variants) |
-| `test_docs_examples.py::test_example[we_are_processes.py]` | `assert None == 0` | `__main__` missing `__file__` in subproc |
-
-**Rule of thumb**: if a test fails with `TooSlowError`,
-`trio.TooSlowError`, or `pexpect.TIMEOUT` and you didn't
-touch the relevant code path, it's flaky — skip it.
-
-## 9. The pytest-capture hang pattern (CHECK THIS FIRST)
-
-**Symptom:** a tractor test hangs indefinitely under
-default `pytest` but passes instantly when you add
-`-s` (`--capture=no`).
-
-**Cause:** tractor subactors (especially under fork-
-based backends) inherit pytest's stdout/stderr
-capture pipes via fds 1,2. Under high-volume error
-logging (e.g. multi-level cancel cascade, nested
-`run_in_actor` failures, anything triggering
-`RemoteActorError` + `ExceptionGroup` traceback
-spew), the **64KB Linux pipe buffer fills** faster
-than pytest drains it. Subactor writes block → can't
-finish exit → parent's `waitpid`/pidfd wait blocks →
-deadlock cascades up the tree.
-
-**Pre-existing guards in the tractor harness** that
-encode this same knowledge — grep these FIRST
-before spelunking:
-
-- `tests/conftest.py:258-260` (in the `daemon`
-  fixture): `# XXX: too much logging will lock up
-  the subproc (smh)` — downgrades `trace`/`debug`
-  loglevel to `info` to prevent the hang.
-- `tests/conftest.py:316`: `# can lock up on the
-  _io.BufferedReader and hang..` — noted on the
-  `proc.stderr.read()` post-SIGINT.
-
-**Debug recipe (in priority order):**
-
-1. **Try `-s` first.** If the hang disappears with
-   `pytest -s`, you've confirmed it's capture-pipe
-   fill. Skip spelunking.
-2. **Lower the loglevel.** Default `--ll=error` on
-   this project; if you've bumped it to `debug` /
-   `info`, try dropping back. Each log level
-   multiplies pipe-pressure under fault cascades.
-3. **If you MUST use default capture + high log
-   volume**, redirect subactor stdout/stderr in the
-   child prelude (e.g.
-   `tractor.spawn._subint_forkserver._child_target`
-   post-`_close_inherited_fds`) to `/dev/null` or a
-   file.
-
-**Signature tells you it's THIS bug (vs. a real
-code hang):**
-
-- Multi-actor test under fork-based backend
-  (`subint_forkserver`, eventually `trio_proc` too
-  under enough log volume).
-- Multiple `RemoteActorError` / `ExceptionGroup`
-  tracebacks in the error path.
-- Test passes with `-s` in the 5-10s range, hangs
-  past pytest-timeout (usually 30+ s) without `-s`.
-- Subactor processes visible via `pgrep -af
-  subint-forkserv` or similar after the hang —
-  they're alive but blocked on `write()` to an
-  inherited stdout fd.
-
-**Historical reference:** this deadlock cost a
-multi-session investigation (4 genuine cascade
-fixes landed along the way) that only surfaced the
-capture-pipe issue AFTER the deeper fixes let the
-tree actually tear down enough to produce pipe-
-filling log volume. Full post-mortem in
-`ai/conc-anal/subint_forkserver_test_cancellation_leak_issue.md`.
-Lesson codified here so future-me grep-finds the
-workaround before digging.
-
-## 10. Reaping zombie subactors (`tractor-reap`)
-
-**Symptom:** after a `pytest` run crashes, times out,
-or is `Ctrl+C`'d, subactor forks (esp. under
-`subint_forkserver`) can be reparented to `init`
-(PPid==1) and linger. They hold onto ports, inherit
-pytest's capture-pipe fds, and flakify later
-sessions.
-
-**Two layers of defense:**
-
-### a) Session-scoped auto-fixture (always on)
-
-`tractor/_testing/pytest.py::_reap_orphaned_subactors`
-runs at pytest session teardown. It walks `/proc` for
-direct descendants of the pytest pid, SIGINTs them,
-waits up to 3s, then SIGKILLs survivors. SC-polite:
-gives the subactor runtime a chance to run its trio
-cancel shield + IPC teardown before escalation.
-
-This is *autouse* and session-scoped — you don't need
-to do anything. It just runs.
-
-### b) `scripts/tractor-reap` CLI (manual reap)
-
-For the **pytest-died-mid-session** case (Ctrl+C, OOM
-kill, hung process you had to `kill -9`), the fixture
-never ran. Reach for the CLI:
-
-```sh
-# default: orphans (PPid==1, cwd==repo, cmd contains python)
-scripts/tractor-reap
-
-# descendant-mode: from a still-live supervisor
-scripts/tractor-reap --parent <pytest-pid>
-
-# see what would be reaped, don't signal
-scripts/tractor-reap -n
-
-# tune the SIGINT → SIGKILL grace window
-scripts/tractor-reap --grace 5
-```
-
-Exit code: `0` if everyone exited on SIGINT, `1` if
-SIGKILL had to escalate — so you can chain it in CI
-health-checks (`scripts/tractor-reap || <alert>`).
-
-**What it matches** (orphan-mode):
-- `PPid == 1` (reparented to init → definitely
-  orphaned, not just a currently-running child)
-- `cwd == <repo-root>` (keeps the sweep scoped; won't
-  touch unrelated init-children elsewhere)
-- `python` in cmdline
-
-**What it does not do:** kill anything whose PPid is
-still a live tractor parent. If the parent is alive
-it's not an orphan; use `--parent <pid>` if you need
-to force-reap under a still-live supervisor.
-
-**When NOT to run it:** while a pytest session is
-active in another terminal. It's safe (won't touch
-that session's live children in orphan-mode) but can
-race if the target session is mid-teardown.
-
-### c) `--shm` / `--shm-only`: orphan-segment sweep
-
-Because `tractor.ipc._mp_bs.disable_mantracker()`
-turns off `mp.resource_tracker` (see
-`ai/conc-anal/subint_forkserver_mp_shared_memory_issue.md`),
-a hard-crashing actor can leave `/dev/shm/<key>`
-segments behind that nothing else GCs.
-
-```sh
-# process reap THEN shm sweep
-scripts/tractor-reap --shm
-
-# shm sweep only (skip process phase)
-scripts/tractor-reap --shm-only
-
-# dry-run: list candidates, don't unlink
-scripts/tractor-reap --shm -n
-```
-
-**Match criteria** (very conservative — this is a
-shared-system path, can't be wrong):
-- segment is a regular file under `/dev/shm`,
-- owned by the **current uid** (`stat.st_uid`),
-- AND **no live process holds it open** —
-  enumerated by walking every readable
-  `/proc/<pid>/maps` (post-mmap mappings) AND
-  `/proc/<pid>/fd/*` (pre-mmap shm-opened fds).
-
-The "nobody has it open" check is the
-kernel-canonical "is this leaked?" test — same
-answer `lsof /dev/shm/<key>` would give. No
-reliance on tractor-specific naming, so it works
-for any tractor app. Critically, it WILL NOT touch
-segments held by other apps you have running
-(e.g. `piker`, `lttng-ust-*`, `aja-shm-*` —
-verified locally with 81 in-use segments correctly
-preserved).

.claude/skills/run-tests/test-harness-reference.md

@@ -0,0 +1,255 @@
+# Tractor Test Harness Reference
+
+This repository-local file supplements the canonical [`/run-tests` skill][1]
+from the [`ai.skillz` repository][2]. Its deployer links the shared `SKILL.md`
+into both
+`.claude/skills/run-tests/` and `.opencode/skills/run-tests/` while preserving
+this project-owned reference:
+
+```text
+bash /path/to/ai.skillz/scripts/deploy.sh run-tests /path/to/tractor --provider all --method symlink
+```
+
+Keep shared environment permission, process-signal safety, target selection,
+failure inspection, and result reporting policy in the deployed `SKILL.md`.
+
+[1]: https://github.com/baudco/ai.skillz/blob/2d4896ca7e38fe2cb3090cdefc7245be4241a6d2/skills/run-tests/SKILL.md
+[2]: https://github.com/baudco/ai.skillz
+
+## Project And Environment
+
+- Project/import: `tractor`
+- Test root: `tests/`
+- Supported Python: `>=3.13,<3.15`
+- Runner: pytest `>=9.0.3`
+- Test dependencies: the `dev` group includes the `testing` group
+- CI uses uv's default `.venv`; the Nix flake uses `py313`.
+- Run from the repository root so pytest loads `pyproject.toml`.
+- Do not use `default.nix` as current test-environment authority; it still
+  selects unsupported Python 3.12.
+
+Environment directory naming is not a harness invariant. Use an already
+verified active project environment when available. Otherwise, use an
+existing uv environment without syncing it:
+
+```text
+uv run --frozen --no-sync python -c 'import pathlib, sys, tractor; root = pathlib.Path.cwd().resolve(); mod = pathlib.Path(tractor.__file__).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
+```
+
+After module moves or collection failures, check collection with:
+
+```text
+uv run --frozen --no-sync pytest --collect-only -q tests/
+```
+
+Collection is not a mandatory precursor to every narrow run. Ask before
+provisioning or changing an environment.
+
+Before trusting CLI-selected runtime settings, inspect
+`TRACTOR_SPAWN_METHOD` and `TRACTOR_LOGLEVEL`. They override the spawn method
+and runtime log level passed by callers, so report active values with test
+results rather than claiming the CLI flags alone selected the runtime.
+
+## Pytest Configuration And Commands
+
+`pyproject.toml` configures:
+
+- `testpaths = ["tests"]` and `--rootdir=./tests`;
+- importlib import mode;
+- the `tractor._testing.pytest` plugin;
+- xonsh plugin disablement;
+- `--show-capture=no` and `--capture=fd`.
+
+Do not silently add `-x`, `--tb=short`, or `--no-header`; those are not
+project defaults. In a verified active environment, replace `uv run
+--frozen --no-sync pytest` below with `python -m pytest`.
+
+```text
+# Full suite
+uv run --frozen --no-sync pytest tests/
+
+# Narrow file
+uv run --frozen --no-sync pytest tests/test_local.py
+
+# Exact node
+uv run --frozen --no-sync pytest tests/discovery/test_registrar.py::test_reg_then_unreg
+
+# Keyword selection
+uv run --frozen --no-sync pytest tests/ -k 'cancel and not slow'
+
+# Previous failures
+uv run --frozen --no-sync pytest --lf
+```
+
+After verifying that the no-sync environment is current, these pytest
+arguments match the Linux TCP CI row:
+
+```text
+CI=1 uv run --frozen --no-sync pytest tests/ -rsx --spawn-backend=trio --tpt-proto=tcp --capture=fd
+```
+
+## Plugin Options And Matrices
+
+Supported spawn backends:
+
+- `trio` (default)
+- `mp_spawn`
+- `mp_forkserver`
+
+Do not advertise `subint`, `subint_forkserver`, or
+`main_thread_forkserver` as runnable backends. Supported transports are
+`tcp` (default) and `uds`. Run one transport per pytest session.
+`mp_forkserver` and UDS are POSIX-only.
+
+Other Tractor plugin options include:
+
+- `--tpdb` / `--debug-mode`
+- `--ll` / `--loglevel`
+- `--tl` / `--tractor-loglevel`
+- `--enable-stackscope`
+
+Examples:
+
+```text
+uv run --frozen --no-sync pytest tests/ipc/ --tpt-proto=uds
+uv run --frozen --no-sync pytest tests/test_spawning.py --spawn-backend=mp_spawn
+uv run --frozen --no-sync pytest tests/test_spawning.py --spawn-backend=mp_forkserver --capture=sys
+```
+
+CI currently exercises Python 3.13 with the `trio` backend: TCP and UDS on
+Linux and macOS, plus an informational TCP row on Windows whose pytest step
+uses `continue-on-error`.
+
+## Registry And Transport Isolation
+
+Tests requesting the `reg_addr` fixture use addresses randomized per session:
+an unreserved unprivileged loopback port for TCP or a unique socket name under
+the platform runtime directory for UDS. A TCP collision remains possible.
+
+The runtime fallback remains `127.0.0.1:1616` or `registry@1616.sock`.
+Inspect that fallback only when the selected test intentionally uses runtime
+defaults or a failure identifies that address. Do not perform a mandatory
+`:1616` preflight or assume UDS sockets live under `/tmp`.
+
+## Capture And Hang Diagnosis
+
+Normal capture is `fd`. Use `--capture=sys` with `mp_forkserver`; some tests
+switch to `capsys`, but the harness does not enforce that suite-wide.
+
+For a suspected capture interaction, compare only the exact node:
+
+```text
+uv run --frozen --no-sync pytest <node> --capture=sys
+uv run --frozen --no-sync pytest <node> -s
+```
+
+Treat `-s` as a diagnostic comparison, not a pass-equivalent workaround. Do
+not use it to reinterpret an ordinary captured pass. Interactive `--tpdb` or
+`tractor.pause()` sessions are different: they require a real TTY and disabled
+capture, normally `-s`.
+
+Do not add a global pytest timeout. `fail_after_w_trace` is Trio-cooperative;
+`afk_alarm_w_trace` is a POSIX main-thread `SIGALRM` hard backstop and can
+raise asynchronously. Use the latter only as a last resort, not as a generally
+Trio-safe timeout replacement.
+
+For live task-tree diagnosis:
+
+```text
+uv run --frozen --no-sync python -c 'import stackscope'
+uv run --frozen --no-sync pytest <node> --enable-stackscope --capture=sys
+kill -USR1 <pytest-pid>
+```
+
+The import check and pytest command must use the same environment. Do not send
+SIGUSR1 if the import fails or setup warns that stackscope or SIGUSR1 is
+unavailable: without the installed handler, SIGUSR1 normally terminates the
+target process. Signal a subactor only after separately confirming that it
+installed the same handler.
+
+Stackscope appends dumps to `/tmp/tractor-stackscope-<pid>.log`, including when
+pytest capture hides terminal output. SIGUSR1 stackscope is unavailable on
+Windows and degrades to a no-op there.
+
+When a trace guard actually fires and snapshot capture succeeds, it writes
+under `$XDG_CACHE_HOME/tractor/hung-dumps/`, falling back beneath
+`~/.cache/tractor/hung-dumps/`, and prints an end-of-session index. A normal
+non-timeout run creates no snapshot.
+
+## Cleanup And `tractor-reap`
+
+On Linux, normal pytest teardown discovers surviving descendants through
+`/proc`, sends SIGINT, waits three seconds, then escalates survivors to
+SIGKILL. It does not sweep shared memory and cannot run if pytest never reaches
+fixture teardown.
+
+Process discovery is a no-op off Linux. UDS PID liveness also depends on
+`/proc`; on macOS, recognized PID-named sockets can therefore be classified as
+dead without proof. The session-scoped autouse fixture currently passes those
+candidates directly to `reap_uds()` at teardown. Do not treat its non-Linux
+classification as proof of orphanhood or run concurrent live Tractor sessions
+against the same UDS bindspace.
+
+Use the CLI in inspection-only mode first:
+
+`--shm` and `--shm-only` are Linux/FreeBSD-only and raise
+`NotImplementedError` elsewhere. On other platforms, use the UDS-only command.
+
+```text
+uv run --frozen --no-sync scripts/tractor-reap -n
+uv run --frozen --no-sync scripts/tractor-reap --parent <pytest-pid> -n
+uv run --frozen --no-sync scripts/tractor-reap --shm --uds -n
+uv run --frozen --no-sync scripts/tractor-reap --uds-only -n
+```
+
+Direct `scripts/tractor-reap` execution is acceptable only after verifying its
+`python3` shebang resolves the intended project environment.
+
+Review every candidate before requesting a mutating run:
+
+- default orphan mode is not repository-scoped;
+- `--parent` trusts the supplied PID and can include non-Tractor children;
+- `--shm` scans all current-user candidate files, not just Tractor-named
+  files;
+- `--uds` treats `registry@1616.sock` as removable even if a live default UDS
+  registrar uses it.
+
+Dry-run output prints only the initially matched root PIDs. A mutating run can
+recursively expand those roots to additional descendants when `psutil` is
+available. Inspect the descendant process tree separately; `-n` is not exact
+signal-set parity and does not by itself authorize signaling unseen children.
+
+The canonical skill owns signaling and unlinking authorization.
+
+## Test Layout And Change Mapping
+
+| Changed area | Run first |
+|---|---|
+| `tractor/runtime/_runtime.py`, `_state.py`, `tractor/_root.py` | `tests/test_local.py`, `tests/test_root_runtime.py`, `tests/test_runtime.py`, `tests/test_rpc.py` |
+| `tractor/runtime/_portal.py`, `_rpc.py` | `tests/test_rpc.py`, `tests/test_cancellation.py` |
+| `tractor/runtime/_supervise.py` | `tests/test_cancellation.py`, `tests/test_spawning.py` |
+| `tractor/discovery/` | `tests/discovery/`, `tests/test_local.py` |
+| `tractor/ipc/` | `tests/ipc/`, `tests/test_2way.py`, `tests/test_shm.py` as relevant |
+| `tractor/spawn/` | `tests/test_spawning.py`, `tests/discovery/test_multi_program.py`, `tests/test_cancellation.py` |
+| `tractor/_context.py`, `_streaming.py` | `tests/test_context_stream_semantics.py`, `tests/test_advanced_streaming.py`, `tests/test_legacy_one_way_streaming.py` |
+| `tractor/to_asyncio.py` | `tests/test_infected_asyncio.py`, `tests/test_root_infect_asyncio.py` |
+| `tractor/msg/` | `tests/msg/` |
+| `tractor/devx/` | `tests/devx/`; debugger tests use pexpect and are comparatively slow |
+| `tractor/_exceptions.py` | `tests/test_remote_exc_relay.py`, `tests/test_reg_err_types.py`, `tests/test_inter_peer_cancellation.py`, `tests/test_cancellation.py`, `tests/msg/` |
+
+Current subdirectories include `discovery/`, `ipc/`, `msg/`, `devx/`, and
+`trionics/`. There is no `tests/spawn/` directory.
+
+## Expected Outcomes
+
+Do not maintain a blanket known-flaky exemption list. Classify only current
+explicit skip or xfail marks and exact expected signatures. Notable tracked
+outcomes include:
+
+- duplicate-name `n_dups=4` and `n_dups=8` variants in
+  `tests/discovery/test_multi_program.py` are non-strict xfails;
+- `tests/test_ringbuf.py` is module-skipped;
+- some documentation examples have explicit macOS-CI skips.
+
+A generic `TooSlowError` or `pexpect.TIMEOUT` is not enough to classify a
+failure as pre-existing.

.gitignore

@@ -168,3 +168,267 @@ gh/
 
 # LLM conversations that should remain private
 docs/conversations/
+
+# BEGIN ai.skillz: direct:symlink:claude:run-tests
+/.claude/skills/run-tests/SKILL.md
+# END ai.skillz: direct:symlink:claude:run-tests
+
+# BEGIN ai.skillz: direct:symlink:opencode:run-tests
+/.opencode/skills/run-tests/SKILL.md
+# END ai.skillz: direct:symlink:opencode:run-tests
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:run-tests
+/.opencode/commands/run-tests.md
+# END ai.skillz: direct:symlink:opencode:command:run-tests
+
+# BEGIN ai.skillz: direct:symlink:claude:gish
+/.claude/skills/gish
+# END ai.skillz: direct:symlink:claude:gish
+
+# BEGIN ai.skillz: direct:symlink:opencode:gish
+/.opencode/skills/gish
+# END ai.skillz: direct:symlink:opencode:gish
+
+# BEGIN ai.skillz: direct:symlink:claude:resolve-conflicts
+/.claude/skills/resolve-conflicts
+# END ai.skillz: direct:symlink:claude:resolve-conflicts
+
+# BEGIN ai.skillz: direct:symlink:opencode:resolve-conflicts
+/.opencode/skills/resolve-conflicts
+# END ai.skillz: direct:symlink:opencode:resolve-conflicts
+
+# BEGIN ai.skillz: direct:symlink:claude:git-mgmt
+/.claude/skills/git-mgmt
+# END ai.skillz: direct:symlink:claude:git-mgmt
+
+# BEGIN ai.skillz: direct:symlink:opencode:git-mgmt
+/.opencode/skills/git-mgmt
+# END ai.skillz: direct:symlink:opencode:git-mgmt
+
+# BEGIN ai.skillz: runtime:open-wkt
+/wkts/
+# END ai.skillz: runtime:open-wkt
+
+# BEGIN ai.skillz: direct:symlink:claude:open-wkt
+/.claude/skills/open-wkt
+# END ai.skillz: direct:symlink:claude:open-wkt
+
+# BEGIN ai.skillz: direct:symlink:opencode:open-wkt
+/.opencode/skills/open-wkt
+# END ai.skillz: direct:symlink:opencode:open-wkt
+
+# BEGIN ai.skillz: direct:symlink:claude:close-wkt
+/.claude/skills/close-wkt
+# END ai.skillz: direct:symlink:claude:close-wkt
+
+# BEGIN ai.skillz: direct:symlink:opencode:close-wkt
+/.opencode/skills/close-wkt
+# END ai.skillz: direct:symlink:opencode:close-wkt
+
+# BEGIN ai.skillz: runtime:code-review
+.ai/code-review/reports/
+# END ai.skillz: runtime:code-review
+
+# BEGIN ai.skillz: direct:symlink:claude:code-review
+/.claude/skills/code-review
+# END ai.skillz: direct:symlink:claude:code-review
+
+# BEGIN ai.skillz: direct:symlink:opencode:code-review
+/.opencode/skills/code-review
+# END ai.skillz: direct:symlink:opencode:code-review
+
+# BEGIN ai.skillz: direct:symlink:claude:code-nav-refs
+/.claude/skills/code-nav-refs
+# END ai.skillz: direct:symlink:claude:code-nav-refs
+
+# BEGIN ai.skillz: direct:symlink:opencode:code-nav-refs
+/.opencode/skills/code-nav-refs
+# END ai.skillz: direct:symlink:opencode:code-nav-refs
+
+# BEGIN ai.skillz: runtime:code-review-changes
+.claude/review_context.md
+.claude/review_regression.md
+.claude/review_replies/
+# END ai.skillz: runtime:code-review-changes
+
+# BEGIN ai.skillz: direct:symlink:claude:code-review-changes
+/.claude/skills/code-review-changes
+# END ai.skillz: direct:symlink:claude:code-review-changes
+
+# BEGIN ai.skillz: direct:symlink:opencode:code-review-changes
+/.opencode/skills/code-review-changes
+# END ai.skillz: direct:symlink:opencode:code-review-changes
+
+# BEGIN ai.skillz: runtime:commit-msg
+.claude/skills/commit-msg/msgs/
+.claude/git_commit_msg_LATEST.md
+# END ai.skillz: runtime:commit-msg
+
+# BEGIN ai.skillz: direct:symlink:claude:commit-msg
+/.claude/skills/commit-msg/SKILL.md
+# END ai.skillz: direct:symlink:claude:commit-msg
+
+# BEGIN ai.skillz: direct:symlink:opencode:commit-msg
+/.opencode/skills/commit-msg/SKILL.md
+# END ai.skillz: direct:symlink:opencode:commit-msg
+
+# BEGIN ai.skillz: direct:symlink:claude:commit-plan
+/.claude/skills/commit-plan
+# END ai.skillz: direct:symlink:claude:commit-plan
+
+# BEGIN ai.skillz: direct:symlink:opencode:commit-plan
+/.opencode/skills/commit-plan
+# END ai.skillz: direct:symlink:opencode:commit-plan
+
+# BEGIN ai.skillz: direct:symlink:claude:dep-supersede-scan
+/.claude/skills/dep-supersede-scan
+# END ai.skillz: direct:symlink:claude:dep-supersede-scan
+
+# BEGIN ai.skillz: direct:symlink:opencode:dep-supersede-scan
+/.opencode/skills/dep-supersede-scan
+# END ai.skillz: direct:symlink:opencode:dep-supersede-scan
+
+# BEGIN ai.skillz: direct:symlink:claude:harness-perf
+/.claude/skills/harness-perf
+# END ai.skillz: direct:symlink:claude:harness-perf
+
+# BEGIN ai.skillz: direct:symlink:opencode:harness-perf
+/.opencode/skills/harness-perf
+# END ai.skillz: direct:symlink:opencode:harness-perf
+
+# BEGIN ai.skillz: direct:symlink:claude:inter-skill-review
+/.claude/skills/inter-skill-review
+# END ai.skillz: direct:symlink:claude:inter-skill-review
+
+# BEGIN ai.skillz: direct:symlink:opencode:inter-skill-review
+/.opencode/skills/inter-skill-review
+# END ai.skillz: direct:symlink:opencode:inter-skill-review
+
+# BEGIN ai.skillz: direct:symlink:claude:opencode-cleaning
+/.claude/skills/opencode-cleaning
+# END ai.skillz: direct:symlink:claude:opencode-cleaning
+
+# BEGIN ai.skillz: direct:symlink:opencode:opencode-cleaning
+/.opencode/skills/opencode-cleaning
+# END ai.skillz: direct:symlink:opencode:opencode-cleaning
+
+# BEGIN ai.skillz: direct:symlink:claude:plan-io
+/.claude/skills/plan-io
+# END ai.skillz: direct:symlink:claude:plan-io
+
+# BEGIN ai.skillz: direct:symlink:opencode:plan-io
+/.opencode/skills/plan-io
+# END ai.skillz: direct:symlink:opencode:plan-io
+
+# BEGIN ai.skillz: runtime:pr-msg
+.claude/skills/pr-msg/msgs/
+.claude/skills/pr-msg/pr_msg_LATEST.md
+# END ai.skillz: runtime:pr-msg
+
+# BEGIN ai.skillz: direct:symlink:claude:pr-msg
+/.claude/skills/pr-msg/SKILL.md
+/.claude/skills/pr-msg/references
+/.claude/skills/pr-msg/scripts
+# END ai.skillz: direct:symlink:claude:pr-msg
+
+# BEGIN ai.skillz: direct:symlink:opencode:pr-msg
+/.opencode/skills/pr-msg/SKILL.md
+/.opencode/skills/pr-msg/references
+/.opencode/skills/pr-msg/scripts
+# END ai.skillz: direct:symlink:opencode:pr-msg
+
+# BEGIN ai.skillz: direct:symlink:claude:prompt-io
+/.claude/skills/prompt-io
+# END ai.skillz: direct:symlink:claude:prompt-io
+
+# BEGIN ai.skillz: direct:symlink:opencode:prompt-io
+/.opencode/skills/prompt-io
+# END ai.skillz: direct:symlink:opencode:prompt-io
+
+# BEGIN ai.skillz: direct:symlink:claude:py-codestyle
+/.claude/skills/py-codestyle
+# END ai.skillz: direct:symlink:claude:py-codestyle
+
+# BEGIN ai.skillz: direct:symlink:opencode:py-codestyle
+/.opencode/skills/py-codestyle
+# END ai.skillz: direct:symlink:opencode:py-codestyle
+
+# BEGIN ai.skillz: runtime:taken-export
+.ai/taken/exports/
+# END ai.skillz: runtime:taken-export
+
+# BEGIN ai.skillz: direct:symlink:claude:taken-export
+/.claude/skills/taken-export
+# END ai.skillz: direct:symlink:claude:taken-export
+
+# BEGIN ai.skillz: direct:symlink:opencode:taken-export
+/.opencode/skills/taken-export
+# END ai.skillz: direct:symlink:opencode:taken-export
+
+# BEGIN ai.skillz: direct:symlink:claude:yt-url-lookup
+/.claude/skills/yt-url-lookup
+# END ai.skillz: direct:symlink:claude:yt-url-lookup
+
+# BEGIN ai.skillz: direct:symlink:opencode:yt-url-lookup
+/.opencode/skills/yt-url-lookup
+# END ai.skillz: direct:symlink:opencode:yt-url-lookup
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:gish
+/.opencode/commands/gish.md
+# END ai.skillz: direct:symlink:opencode:command:gish
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:resolve-conflicts
+/.opencode/commands/resolve-conflicts.md
+# END ai.skillz: direct:symlink:opencode:command:resolve-conflicts
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:git-mgmt
+/.opencode/commands/git-mgmt.md
+# END ai.skillz: direct:symlink:opencode:command:git-mgmt
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:open-wkt
+/.opencode/commands/open-wkt.md
+# END ai.skillz: direct:symlink:opencode:command:open-wkt
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:close-wkt
+/.opencode/commands/close-wkt.md
+# END ai.skillz: direct:symlink:opencode:command:close-wkt
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:code-review
+/.opencode/commands/code-review.md
+# END ai.skillz: direct:symlink:opencode:command:code-review
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:code-review-changes
+/.opencode/commands/code-review-changes.md
+# END ai.skillz: direct:symlink:opencode:command:code-review-changes
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:commit-msg
+/.opencode/commands/commit-msg.md
+# END ai.skillz: direct:symlink:opencode:command:commit-msg
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:commit-plan
+/.opencode/commands/commit-plan.md
+# END ai.skillz: direct:symlink:opencode:command:commit-plan
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:dep-supersede-scan
+/.opencode/commands/dep-supersede-scan.md
+# END ai.skillz: direct:symlink:opencode:command:dep-supersede-scan
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:harness-perf
+/.opencode/commands/harness-perf.md
+# END ai.skillz: direct:symlink:opencode:command:harness-perf
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:opencode-cleaning
+/.opencode/commands/opencode-cleaning.md
+# END ai.skillz: direct:symlink:opencode:command:opencode-cleaning
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:pr-msg
+/.opencode/commands/pr-msg.md
+# END ai.skillz: direct:symlink:opencode:command:pr-msg
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:taken-export
+/.opencode/commands/taken-export.md
+# END ai.skillz: direct:symlink:opencode:command:taken-export
+
+# BEGIN ai.skillz: direct:symlink:opencode:command:yt-url-lookup
+/.opencode/commands/yt-url-lookup.md
+# END ai.skillz: direct:symlink:opencode:command:yt-url-lookup

ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.md

@@ -0,0 +1,30 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: ses_0799212ebffe42arY96czXn89F
+timestamp: 2026-08-11T23:38:33Z
+git_ref: 7cbd64ee
+scope: code
+substantive: true
+raw_file: 20260811T233833Z_7cbd64ee_prompt_io.raw.md
+---
+
+## Prompt
+
+Open a new isolated worktree in the local Tractor repository and draft a fix
+for `BroadcastReceiver` reporting that a lagged one-slot consumer dropped
+zero values when one value had actually been displaced.
+
+## Response summary
+
+Corrected the off-by-one lag count while preserving cursor recovery and added
+deterministic narrow- and wider-window regressions for exact loss reporting.
+
+## Files changed
+
+- `tractor/trionics/_broadcast.py` - exact broadcast overrun count.
+- `tests/test_task_broadcasting.py` - lag count and recovery regression.
+
+## Human edits
+
+None - generated output follows the user's diagnosed edge case.

ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.raw.md

@@ -0,0 +1,50 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-11T23:38:33Z
+git_ref: 7cbd64ee
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+The user asked for a Tractor fix in a new isolated worktree after a live piker
+failure reported:
+
+```text
+tractor.trionics._broadcast.Lagged:
+Task `piker.brokers.ib.broker.handle_order_requests` overrun and
+dropped `0` values
+```
+
+Inspection showed the lag exception was valid but its count was off by one.
+`BroadcastReceiver.receive_nowait()` treats `seq` as a deque index. With a
+one-entry queue, index zero is the only retained value and `seq == 1` already
+means one value was displaced. The old `seq - maxlen` calculation therefore
+reported zero instead of one.
+
+> `git diff HEAD~1..HEAD -- tractor/trionics/_broadcast.py`
+
+Adjusted the lag count to `seq - maxlen + 1` and documented why the first
+invalid deque index must be included. The existing Tokio-style cursor reset
+remains unchanged.
+
+> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py`
+
+Added a deterministic parameterized regression covering a one-slot queue
+with one dropped value and a three-slot queue with two dropped values. The
+test keeps the root receiver idle while a child subscriber synchronously
+drains each produced value, asserts the exact `Lagged` message, and proves the
+next receive resumes at the oldest retained item.
+
+Verification output:
+
+```text
+..                                                                       [100%]
+2 passed in 0.04s
+
+..........                                                               [100%]
+10 passed in 4.65s
+```
+
+The targeted import resolved to the new Tractor worktree. Python compilation
+and `git diff --check` passed. Adversarial review found no actionable issues;
+zero-capacity channels remain a pre-existing untested edge outside this fix.

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.

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.

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

@@ -0,0 +1,33 @@
+---
+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

@@ -0,0 +1,49 @@
+---
+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.

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

@@ -0,0 +1,33 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: ses_0799212ebffe42arY96czXn89F
+timestamp: 2026-08-12T15:00:27Z
+git_ref: c2a6ccef
+scope: code
+substantive: true
+raw_file: 20260812T150027Z_c2a6ccef_prompt_io.raw.md
+---
+
+## Prompt
+
+For Tractor PR 490 item three, prevent subscriber closure from waking another
+receiver's shared event or stranding peers. Preserve close-safe source-read
+ownership handoff, then review, verify and return a complete commit plan.
+
+## Response summary
+
+Introduced receiver-local wait/source cancellation, owner-specific handoff,
+and close precedence over shielded source values/errors/EOC. Clarified that
+private scope cancellation means explicit close while outer task cancellation
+remains `trio.Cancelled` for both source-read and peer-wait scopes, with
+deterministic receiver-close regressions.
+
+## Files changed
+
+- `tractor/trionics/_broadcast.py` - receiver-local close and ownership scopes.
+- `tests/test_task_broadcasting.py` - peer close and owner handoff regressions.
+
+## Human edits
+
+None - generated output follows the requested third iterative item.

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

@@ -0,0 +1,51 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-12T15:00:27Z
+git_ref: c2a6ccef
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+The user requested the third iterative refinement for Tractor PR 490: closing
+one broadcast subscriber must not set another receiver owner's shared event
+and create a runnable hot loop, then stop for a complete commit plan.
+
+> `git diff HEAD~1..HEAD -- tractor/trionics/_broadcast.py`
+
+Added receiver-local wait cancellation and source-read ownership scopes.
+Closing a non-owner waiting behind another source reader cancels only that
+receiver's private wait and maps it to `ClosedResourceError`; the shared event
+remains untouched. Closing the active source owner cancels only its private
+source-read scope, wakes peers after cleanup, and lets one peer take ownership.
+
+Source outcomes are captured inside the owner scope and classified only after
+checking close/cancel state. A cancellation-shielding source therefore cannot
+publish a returned value, ordinary error, or EOC after its owner was closed.
+The private scope's `cancel_called` bit is asserted to imply receiver closure;
+outer task cancellation remains `trio.Cancelled` and is not translated into
+`ClosedResourceError`. Owner-key comments document that only the receiver
+identified by `recv_ready[0]` may cancel the shared source-read scope. The
+same explicit-close invariant is enforced symmetrically for private peer-wait
+scope cancellation.
+
+> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py`
+
+Added deterministic bounded regressions for both close positions. The
+non-owner test places two peers behind an active source read, closes one and
+proves only that peer exits while the shared event stays unset. The owner test
+closes a source owner whose receive shields cancellation and parameterizes a
+returned value, `RuntimeError`, and `EndOfChannel`; each discarded outcome
+hands the next source receive to the waiting root without terminal-state or
+EOC publication.
+
+Verification output:
+
+```text
+.................                                                        [100%]
+17 passed in 5.84s
+```
+
+Python compilation and `git diff --check` passed. Iterative adversarial review
+caught a waiting non-owner hang, cancellation-shielded source returns, and
+shielded source exceptions. All were fixed. Final review found no actionable
+issues.

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.

ai/prompt-io/opencode/20260813T181901Z_a2e0df4b_prompt_io.md

@@ -0,0 +1,36 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: unavailable
+timestamp: 2026-08-13T18:19:01Z
+git_ref: a2e0df4b
+scope: code
+substantive: true
+raw_file: 20260813T181901Z_a2e0df4b_prompt_io.raw.md
+---
+
+## Prompt
+
+Continue Tractor PR 490 after the paired piker EMS consumer commit. Clean
+cancelled-task diagnostics, define or reject zero-buffer broadcast behavior,
+review and verify the change, then stop at a complete commit plan.
+
+## Response summary
+
+Bound cancellation diagnostics to receiver progress, terminal state and
+resource lifetime; made EOC durable across peers; released wrapper-owned root
+broadcasters without breaking graceful EOC or subclass overrides; and rejected
+non-positive fan-out retention capacity.
+
+## Files changed
+
+- `tractor/trionics/_broadcast.py` - diagnostic lifecycle, durable EOC and
+  buffer validation.
+- `tractor/_streaming.py` - safe `MsgStream` root broadcaster cleanup.
+- `tractor/to_asyncio.py` - linked-channel root broadcaster cleanup.
+- `tests/test_task_broadcasting.py` - cancellation, EOC, wrapper and capacity
+  regressions.
+
+## Human edits
+
+None - generated output follows the requested fifth iterative item.

ai/prompt-io/opencode/20260813T181901Z_a2e0df4b_prompt_io.raw.md

@@ -0,0 +1,58 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-13T18:19:01Z
+git_ref: a2e0df4b
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+The user asked to continue after committing the paired piker EMS consumer
+fix. The next isolated Tractor PR 490 item was to clean cancelled-task
+diagnostics and define zero-buffer broadcast behavior, then review, test and
+stop at a complete commit plan.
+
+> `git diff HEAD~1..HEAD -- tractor/trionics/_broadcast.py`
+
+Made `BroadcastState.cancelled` transient: receiver progress and close clear
+that receiver's diagnostic, terminal EOC and shared receive failure clear all
+stale cancelled tasks, and durable EOC prevents peers from re-entering the
+closed source. `broadcast_receiver()` now rejects non-positive retention
+capacity before creating an unusable zero-length deque.
+
+> `git diff HEAD~1..HEAD -- tractor/_streaming.py`
+
+Made explicit `MsgStream.aclose()` release its internally allocated root
+broadcaster while preserving graceful receive-internal EOC teardown. Used a
+task-local marker so the public zero-argument `aclose()` signature and valid
+subclass overrides remain compatible.
+
+> `git diff HEAD~1..HEAD -- tractor/to_asyncio.py`
+
+Made `LinkedTaskChannel.aclose()` release its internally allocated root
+broadcaster before closing the underlying Trio receive channel.
+
+> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py`
+
+Added synchronized regressions for transient child cancellation diagnostics,
+cross-receiver terminal cleanup, durable EOC peer wakeups, root broadcaster
+cleanup through both public wrappers, `MsgStream.aclose()` subclass
+compatibility, and zero-buffer rejection.
+
+Verification output:
+
+```text
+...........................                                              [100%]
+27 passed in 5.89s
+
+.                                                                        [100%]
+1 passed in 1.22s
+
+.                                                                        [100%]
+1 passed in 0.88s
+```
+
+The integration runs cover real `MsgStream` actor fan-out and infected-asyncio
+`LinkedTaskChannel` fan-out. Ruff, Python compilation and `git diff --check`
+passed. Repeated adversarial review found and resolved root close re-entrancy,
+cross-receiver terminal retention, durable-EOC and subclass-compatibility
+issues; final review reported no findings.

ai/prompt-io/opencode/20260822T022526Z_5562fd9a_prompt_io.md

@@ -0,0 +1,49 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: 76c5d31c-5a2f-4503-9b16-410ee7f4fab3
+timestamp: 2026-08-22T02:25:26Z
+git_ref: eb3c99c9
+scope: config
+substantive: true
+raw_file: 20260822T022526Z_5562fd9a_prompt_io.raw.md
+---
+
+## Prompt
+
+Perform a full Tractor repository scan for related `ai.skillz` work,
+then correct the run-tests landing branch, prune unrelated `.gitignore`
+additions, preserve only the focused migration, and provide canonical
+deployment commands.
+
+## Response summary
+
+Audited all local branches, worktrees, affected-path history, deployment
+state, canonical skill dependencies, and current Tractor harness behavior.
+Corrected the local test reference where it overstated cleanup safety or
+omitted current environment, platform, debugger, timeout, and CI details.
+Narrowed the correction commit to three managed `run-tests` deployment
+blocks. A later dedicated commit records the complete generated `ai.skillz`
+deployment state.
+
+## Files changed
+
+- `.claude/skills/run-tests/test-harness-reference.md` - correct the
+  project-specific test and cleanup contract.
+- `.gitignore` - narrow the correction commit before the later dedicated
+  deployment-state expansion.
+- `ai/prompt-io/opencode/20260822T022526Z_5562fd9a_prompt_io.md` - record
+  the migration review provenance.
+- `ai/prompt-io/opencode/20260822T022526Z_5562fd9a_prompt_io.raw.md` -
+  preserve the unedited response record.
+
+## Human edits
+
+The human required an existing-work scan after duplicate implementation
+was discovered, approved correcting the landing branch during PR #481
+review, and directed removal or reconciliation of unrelated ignore rules.
+During PR #510 review, the human required the stackscope and shared-memory
+safety clarifications and immutable provenance pointers before landing. No
+direct source-line edits were made by the human. Copilot review then prompted
+the human to require explicit canonical deployment instructions and clarify
+the later `.gitignore` expansion.

ai/prompt-io/opencode/20260822T022526Z_5562fd9a_prompt_io.raw.md

@@ -0,0 +1,24 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-22T02:25:26Z
+git_ref: eb3c99c9
+diff_cmd: git diff eb3c99c9^..eb3c99c9
+---
+
+Audit Tractor's repository-wide `ai.skillz` state and correct the
+`wkt/ai_skillz_run_tests_landing` migration before landing it.
+
+> `git diff eb3c99c9^..eb3c99c9 -- .claude/skills/run-tests/test-harness-reference.md`
+
+The harness reference is corrected against current Tractor behavior:
+environment overrides, CI/platform matrices, randomized registry
+addresses, interactive debugger capture, timeout mechanisms, stackscope
+artifacts, reaper platform limits, dry-run expansion, and exact
+change-to-test mappings.
+
+> `git diff eb3c99c9^..eb3c99c9 -- .gitignore`
+
+Unrelated Taken and broad OpenCode command ignore additions are removed;
+only the managed canonical `run-tests` link and command blocks remain in
+this migration.

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/api/trionics.rst

@@ -40,6 +40,9 @@ 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
@@ -48,6 +51,13 @@ 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
 ----------------------
 

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,20 @@ 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
+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,14 +8,18 @@ from contextlib import (
 from functools import partial
 from itertools import cycle
 import time
+from types import SimpleNamespace
 from typing import Optional
+import warnings
 
 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,
     Lagged,
     collapse_eg,
 )
@@ -307,6 +311,847 @@ def test_subscribe_errors_after_close():
     trio.run(main)
 
 
+@pytest.mark.parametrize(
+    ('size', 'sent', 'dropped'),
+    [
+        (1, 2, 1),
+        (3, 5, 2),
+    ],
+)
+def test_lagged_reports_exact_drop_count(
+    size: int,
+    sent: int,
+    dropped: int,
+) -> None:
+    '''
+    `Lagged` must report every value outside the retained window.
+
+    `BroadcastReceiver.receive_nowait()` previously subtracted the
+    queue length from an already-invalid deque index without counting
+    that first displaced value. A one-slot queue therefore claimed it
+    dropped zero values after two sends. Keep one root receiver idle
+    while a child subscriber drains every produced value, then prove
+    the lag error reports the exact overrun and positions the root at
+    the oldest value still retained by `BroadcastState.queue`.
+
+    '''
+    async def main() -> None:
+        tx, rx = trio.open_memory_channel(size)
+        brx = broadcast_receiver(rx, size)
+
+        async with brx.subscribe() as fast:
+            for value in range(sent):
+                await tx.send(value)
+                assert await fast.receive() == value
+
+            match = rf'dropped `{dropped}` values'
+            with pytest.raises(Lagged, match=match):
+                await brx.receive()
+
+            assert await brx.receive() == sent - size
+
+    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
+            state.recv_ready = None
+
+    trio.run(main)
+
+
+def test_cancelled_reader_diagnostics_are_transient() -> None:
+    '''
+    Cancelled-reader diagnostics must not retain stale `Task`s.
+
+    `BroadcastState.cancelled` previously accumulated every source
+    owner cancelled during `BroadcastReceiver.receive()`. Even after
+    that receiver successfully read again or its subscription closed,
+    `BroadcastState.statistics()` retained the old `Task`, reporting
+    stale state and keeping the completed task alive.
+
+    Cancel one child's source read under a receiver-local scope and
+    verify its task is reported. Reuse that same receiver for one
+    successful read to prove progress clears the entry. Cancel it once
+    more, then leave the subscription and prove close also removes the
+    diagnostic while the root receiver remains registered.
+
+    '''
+    async def main() -> None:
+        tx, rx = trio.open_memory_channel(1)
+        brx = broadcast_receiver(rx, 1)
+        cancel_scope = trio.CancelScope()
+        child_key: int
+        child_task = None
+
+        async with brx.subscribe() as child:
+            child_key = child.key
+
+            async def cancel_source_read() -> None:
+                nonlocal child_task
+                child_task = current_task()
+                with cancel_scope:
+                    await child.receive()
+                assert cancel_scope.cancelled_caught
+
+            async with trio.open_nursery() as nursery:
+                nursery.start_soon(cancel_source_read)
+                while brx._state.recv_ready is None:
+                    await trio.lowlevel.checkpoint()
+                cancel_scope.cancel()
+
+            stats = brx._state.statistics()
+            assert child_task is not None
+            assert stats['tasks_cancelled'] == {
+                child_key: child_task,
+            }
+
+            await tx.send(1)
+            assert await child.receive() == 1
+            assert not brx._state.cancelled
+
+            cancel_scope = trio.CancelScope()
+            async with trio.open_nursery() as nursery:
+                nursery.start_soon(cancel_source_read)
+                while brx._state.recv_ready is None:
+                    await trio.lowlevel.checkpoint()
+                cancel_scope.cancel()
+
+            assert child_key in brx._state.cancelled
+
+        assert child_key not in brx._state.cancelled
+        assert brx.key in brx._state.subs
+
+    trio.run(main)
+
+
+@pytest.mark.parametrize(
+    'terminal_exc',
+    [
+        trio.EndOfChannel(),
+        RuntimeError('terminal source failure'),
+    ],
+    ids=['end-of-channel', 'receive-error'],
+)
+def test_terminal_broadcast_clears_cancelled_tasks(
+    terminal_exc: Exception,
+) -> None:
+    '''
+    Terminal broadcast state must release every cancelled `Task`.
+
+    A receiver which owned and cancelled a source read can leave its
+    task in `BroadcastState.cancelled`. If another receiver later gets
+    EOC or a terminal source failure, no subscriber can make source
+    progress to clear that stale diagnostic. Clearing only the terminal
+    owner's key therefore retained the first receiver's completed task.
+
+    Cancel a child during the first controlled source read, then let
+    the root own a second read which raises EOC or `RuntimeError`.
+    Prove each terminal path clears the other receiver's diagnostic
+    before propagating its exact source outcome.
+
+    '''
+    class TerminalReceiver:
+        '''
+        Block one cancellable read, then raise a terminal outcome.
+
+        '''
+        def __init__(self) -> None:
+            self.calls = 0
+            self.first_started = trio.Event()
+
+        async def receive(self) -> None:
+            '''
+            Drive cancellation followed by terminal source state.
+
+            '''
+            self.calls += 1
+            if self.calls == 1:
+                self.first_started.set()
+                await trio.sleep_forever()
+
+            raise terminal_exc
+
+    async def main() -> None:
+        source = TerminalReceiver()
+        brx = broadcast_receiver(source, 1)
+        cancel_scope = trio.CancelScope()
+
+        async with brx.subscribe() as child:
+            async def cancel_child_read() -> None:
+                with cancel_scope:
+                    await child.receive()
+                assert cancel_scope.cancelled_caught
+
+            async with trio.open_nursery() as nursery:
+                nursery.start_soon(cancel_child_read)
+                await source.first_started.wait()
+                cancel_scope.cancel()
+
+            assert child.key in brx._state.cancelled
+            with pytest.raises(type(terminal_exc)) as exc_info:
+                await brx.receive()
+            assert exc_info.value is terminal_exc
+            assert not brx._state.cancelled
+
+    trio.run(main)
+
+
+def test_end_of_channel_is_terminal_for_waiting_peer() -> None:
+    '''
+    EOC must not let an awakened peer re-enter the closed source.
+
+    `BroadcastState.eoc` was set when one source owner received EOC,
+    but neither receive path consulted it. A peer waiting behind that
+    owner therefore woke, saw no queued value, and started a second
+    source read. Cancellation at that checkpoint could repopulate
+    `BroadcastState.cancelled` after the broadcast became terminal.
+
+    Block one child in the sole source read while the root waits on its
+    event, then release EOC. Both receivers must terminate from that
+    one source call, and the root's later receive must replay EOC
+    immediately without retaining cancellation diagnostics.
+
+    '''
+    class EOCReceiver:
+        '''
+        Publish one controlled EOC and reject any second source read.
+
+        '''
+        def __init__(self) -> None:
+            self.calls = 0
+            self.started = trio.Event()
+            self.release = trio.Event()
+
+        async def receive(self) -> None:
+            '''
+            Block the only valid source read until EOC release.
+
+            '''
+            self.calls += 1
+            assert self.calls == 1
+            self.started.set()
+            await self.release.wait()
+            raise trio.EndOfChannel
+
+    async def main() -> None:
+        source = EOCReceiver()
+        brx = broadcast_receiver(source, 1)
+        outcomes: list[str] = []
+
+        async with brx.subscribe() as child:
+            async def receive_eoc(
+                receiver,
+                name: str,
+            ) -> None:
+                with pytest.raises(trio.EndOfChannel):
+                    await receiver.receive()
+                outcomes.append(name)
+
+            async with trio.open_nursery() as nursery:
+                nursery.start_soon(receive_eoc, child, 'child')
+                await source.started.wait()
+                nursery.start_soon(receive_eoc, brx, 'root')
+
+                _, event = brx._state.recv_ready
+                while not event.statistics().tasks_waiting:
+                    await trio.lowlevel.checkpoint()
+                source.release.set()
+
+            with pytest.raises(trio.EndOfChannel):
+                await brx.receive()
+
+        assert sorted(outcomes) == ['child', 'root']
+        assert source.calls == 1
+        assert not brx._state.cancelled
+
+    trio.run(main)
+
+
+def test_msgstream_eoc_close_preserves_aclose_override() -> None:
+    '''
+    Internal EOC cleanup must preserve the public `aclose()` contract.
+
+    Passing a new private keyword from `MsgStream.receive()` to
+    `self.aclose()` broke subclasses whose compatible override kept
+    the original zero-argument signature. Use a minimal subclass which
+    records virtual dispatch and delegates to the base implementation.
+    Drive graceful EOC through the real root broadcaster and prove the
+    override runs without closing that active root re-entrantly.
+
+    '''
+    class Stream(tractor.MsgStream):
+        '''
+        Record public close dispatch with the established signature.
+
+        '''
+        close_calls = 0
+
+        async def aclose(self):
+            '''
+            Delegate closure without accepting private arguments.
+
+            '''
+            self.close_calls += 1
+            return await super().aclose()
+
+    class PldRx:
+        '''
+        Delegate source receive and terminate the close drain.
+
+        '''
+        def __init__(self, rx) -> None:
+            self._rx = rx
+
+        async def recv_pld(self, **kwargs):
+            '''
+            Receive directly from the test source channel.
+
+            '''
+            return await self._rx.receive()
+
+        def recv_msg_nowait(self, **kwargs):
+            '''
+            Report EOC to finish `MsgStream.aclose()` draining.
+
+            '''
+            raise trio.EndOfChannel
+
+    async def main() -> None:
+        tx, rx = trio.open_memory_channel(1)
+        ctx = SimpleNamespace(
+            cid='test-context',
+            _pld_rx=PldRx(rx),
+            send_stop=lambda: trio.lowlevel.checkpoint(),
+            side='caller',
+            peer_side='callee',
+            maybe_raise=lambda **kwargs: None,
+        )
+        stream = Stream(ctx, rx)
+
+        async with stream.subscribe():
+            await tx.aclose()
+            with pytest.raises(trio.EndOfChannel):
+                await stream.receive()
+            assert stream.close_calls == 1
+            assert not stream._broadcaster._closed
+
+    trio.run(main)
+
+
+@pytest.mark.parametrize(
+    'close_wrapper',
+    [
+        tractor.MsgStream.aclose,
+        LinkedTaskChannel.aclose,
+    ],
+    ids=['msg-stream', 'linked-task-channel'],
+)
+def test_wrapper_close_clears_root_cancelled_task(
+    close_wrapper,
+) -> None:
+    '''
+    Public stream close must release root cancellation diagnostics.
+
+    Root broadcasters allocated by `MsgStream.subscribe()` and
+    `LinkedTaskChannel.subscribe()` are private implementation state.
+    If their source receive was cancelled, callers had no public way
+    to close the root, so wrapper teardown retained the completed
+    `Task` in `BroadcastState.cancelled` indefinitely.
+
+    Cancel a root source read, attach that broadcaster to a minimal
+    public wrapper, and close it through each real `aclose()` method.
+    The root receiver and its task diagnostic must both be removed;
+    for `MsgStream`, pre-close the source to cover its idempotent early
+    return path.
+
+    '''
+    async def main() -> None:
+        _, rx = trio.open_memory_channel(1)
+        brx = broadcast_receiver(rx, 1)
+        cancel_scope = trio.CancelScope()
+
+        async def cancel_source_read() -> None:
+            with cancel_scope:
+                await brx.receive()
+            assert cancel_scope.cancelled_caught
+
+        async with trio.open_nursery() as nursery:
+            nursery.start_soon(cancel_source_read)
+            while brx._state.recv_ready is None:
+                await trio.lowlevel.checkpoint()
+            cancel_scope.cancel()
+
+        assert brx.key in brx._state.cancelled
+
+        if close_wrapper is tractor.MsgStream.aclose:
+            ctx = SimpleNamespace(cid='test-context')
+            wrapper = tractor.MsgStream(ctx, rx)
+            wrapper._broadcaster = brx
+            await rx.aclose()
+        else:
+            wrapper = SimpleNamespace(
+                _broadcaster=brx,
+                _from_aio=rx,
+            )
+
+        await close_wrapper(wrapper)
+        assert brx.key not in brx._state.subs
+        assert brx.key not in brx._state.cancelled
+
+    trio.run(main)
+
+
+def test_broadcast_rejects_zero_buffer_size() -> None:
+    '''
+    A broadcaster must retain at least one value for peer fan-out.
+
+    `collections.deque(maxlen=0)` silently discards every appended
+    value, so `broadcast_receiver(..., 0)` allowed the source owner to
+    receive while peer cursors advanced into an always-empty queue.
+    Their lag recovery then reset to index `-1` and recursively retried
+    without any retained value to consume.
+
+    Construct a rendezvous memory channel and prove broadcaster setup
+    rejects its zero capacity synchronously with a clear public error,
+    before any receiver is registered or source receive can begin.
+
+    '''
+    _, rx = trio.open_memory_channel(0)
+    with pytest.raises(
+        ValueError,
+        match='`max_buffer_size` must be greater than zero',
+    ):
+        broadcast_receiver(rx, 0)
+
+
+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_closing_non_owner_preserves_source_wait() -> None:
+    '''
+    Closing one subscriber must not wake another receiver's peers.
+
+    `BroadcastReceiver.aclose()` previously set the one shared
+    `BroadcastState.recv_ready` event even when a different receiver
+    owned the source read. Waiting peers then repeatedly awaited an
+    already-set event until the source produced another value,
+    creating a runnable hot loop on idle streams.
+
+    Block one child in the source receive, then place both the root
+    and a closing child behind its event. Close only that waiting
+    child and prove it gets `ClosedResourceError` without setting the
+    shared event. Both remaining receivers must still get the same
+    value after the source is released.
+
+    '''
+    async def main() -> None:
+        tx, rx = trio.open_memory_channel(1)
+        brx = broadcast_receiver(rx, 3)
+        owner_value: list[int] = []
+        root_value: list[int] = []
+        closing_closed = trio.Event()
+
+        async with (
+            brx.subscribe() as owner,
+            brx.subscribe() as closing,
+        ):
+            async def receive_owner() -> None:
+                owner_value.append(await owner.receive())
+
+            async def receive_root() -> None:
+                root_value.append(await brx.receive())
+
+            async def receive_closing() -> None:
+                with pytest.raises(trio.ClosedResourceError):
+                    await closing.receive()
+                closing_closed.set()
+
+            with trio.fail_after(1):
+                async with trio.open_nursery() as nursery:
+                    nursery.start_soon(receive_owner)
+                    while brx._state.recv_ready is None:
+                        await trio.lowlevel.checkpoint()
+
+                    nursery.start_soon(receive_root)
+                    nursery.start_soon(receive_closing)
+                    _, event = brx._state.recv_ready
+                    while event.statistics().tasks_waiting < 2:
+                        await trio.lowlevel.checkpoint()
+
+                    await closing.aclose()
+                    await closing_closed.wait()
+                    assert not event.is_set()
+                    await tx.send(1)
+
+        assert owner_value == [1]
+        assert root_value == [1]
+
+    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',
+    [
+        1,
+        RuntimeError('discarded source error'),
+        trio.EndOfChannel(),
+    ],
+)
+def test_closing_source_owner_hands_read_to_peer(
+    first_outcome: int|Exception,
+) -> None:
+    '''
+    Closing the source-read owner must transfer ownership to a peer.
+
+    Merely suppressing the old shared-event wake would leave peers
+    blocked behind an externally closed receiver that still owned an
+    idle source read. Script a first receive which blocks until its
+    private scope is cancelled and a second which returns immediately.
+    Close that owner only after the root is waiting behind it. Cover
+    a shielded value, ordinary error and EOC from the cancelled source
+    read. The owner must always get `ClosedResourceError`, while the
+    awakened root takes the second source read without publishing the
+    discarded source outcome.
+
+    '''
+    class HandoffReceiver:
+        '''
+        Block the first source read and satisfy the second.
+
+        '''
+        def __init__(self) -> None:
+            self.calls: int = 0
+            self.first_started = trio.Event()
+            self.release_first = trio.Event()
+
+        async def receive(self) -> int:
+            '''
+            Drive one cancelled read followed by one value.
+
+            '''
+            self.calls += 1
+            if self.calls == 1:
+                self.first_started.set()
+                with trio.CancelScope(shield=True):
+                    await self.release_first.wait()
+                if isinstance(first_outcome, BaseException):
+                    raise first_outcome
+                return first_outcome
+
+            return 2
+
+    async def main() -> None:
+        source = HandoffReceiver()
+        brx = broadcast_receiver(source, 3)
+        owner_closed = trio.Event()
+        root_value: list[int] = []
+
+        async with brx.subscribe() as owner:
+            async def receive_owner() -> None:
+                with pytest.raises(trio.ClosedResourceError):
+                    await owner.receive()
+                owner_closed.set()
+
+            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_owner)
+                    await source.first_started.wait()
+                    nursery.start_soon(receive_root)
+
+                    _, event = brx._state.recv_ready
+                    while not event.statistics().tasks_waiting:
+                        await trio.lowlevel.checkpoint()
+
+                    await owner.aclose()
+                    source.release_first.set()
+                    await owner_closed.wait()
+
+        assert source.calls == 2
+        assert root_value == [2]
+        assert brx._state.receive_exc is None
+        assert not brx._state.eoc
+
+    trio.run(main)
+
+
 def test_ensure_slow_consumers_lag_out(
     reg_addr,
     start_method,
@@ -448,6 +1293,7 @@ def test_first_recver_is_cancelled():
                     async with brx.subscribe() as bc:
                         async for value in bc:
                             print(value)
+                assert cs.cancelled_caught
 
             async def cancel_and_send():
                 await trio.sleep(0.2)
@@ -519,3 +1365,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

@@ -103,6 +103,12 @@ class MsgStream(trio.abc.Channel):
         self._eoc: bool|trio.EndOfChannel = False
         self._closed: bool|trio.ClosedResourceError = False
 
+        # `MsgStream.receive()` sets this while it calls
+        # `MsgStream.aclose()` after source EOC. That close is
+        # re-entrant from the root `BroadcastReceiver._recv`, so it
+        # must not cancel the same receiver before EOC propagates.
+        self._eoc_close_task: trio.lowlevel.Task|None = None
+
     @property
     def ctx(self) -> Context:
         '''
@@ -256,7 +262,16 @@ class MsgStream(trio.abc.Channel):
 
         # when the send is closed we assume the stream has
         # terminated and signal this local iterator to stop
-        drained: list[Exception|dict] = await self.aclose()
+        #
+        # Preserve virtual dispatch through the public zero-argument
+        # `MsgStream.aclose()` API. The task marker lets the base
+        # implementation distinguish this receive-internal close from
+        # an explicit caller or `MsgStream.__aexit__()` close.
+        self._eoc_close_task = trio.lowlevel.current_task()
+        try:
+            drained: list[Exception|dict] = await self.aclose()
+        finally:
+            self._eoc_close_task = None
         if drained:
         #  ^^^^^^^^TODO? pass these to the `._ctx._drained_msgs:
         #  deque` and then iterate them as part of any
@@ -335,6 +350,20 @@ class MsgStream(trio.abc.Channel):
         # `.__aexit__()` as well!!!
         # => SO ENSURE WE CATCH ALL TERMINATION STATES in this
         # block including the EoC..
+
+        # `MsgStream.subscribe()` stores its hidden root broadcaster
+        # on `self._broadcaster`. Explicit teardown owns that root and
+        # must close it to release its subscriber and cancelled-task
+        # diagnostic. Skip only the receive-internal EOC close above:
+        # cancelling the active root's source-read scope there would
+        # turn graceful EOC into `trio.ClosedResourceError`.
+        if (
+            trio.lowlevel.current_task() is not self._eoc_close_task
+            and
+            (broadcaster := self._broadcaster) is not None
+        ):
+            await broadcaster.aclose()
+
         if self.closed:
             # this stream has already been closed so silently succeed as
             # per ``trio.AsyncResource`` semantics.
@@ -512,6 +541,7 @@ class MsgStream(trio.abc.Channel):
     @acm
     async def subscribe(
         self,
+        raise_on_lag: bool = True,
 
     ) -> AsyncIterator[BroadcastReceiver]:
         '''
@@ -526,6 +556,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 +576,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 +588,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

@@ -213,6 +213,14 @@ class LinkedTaskChannel(
     _broadcaster: BroadcastReceiver|None = None
 
     async def aclose(self) -> None:
+        # `LinkedTaskChannel.subscribe()` lazily allocates and retains
+        # this root receiver. Close it first so its receiver-local
+        # source-read scope and cancellation diagnostics are released
+        # before `self._from_aio` becomes inaccessible; child
+        # subscriptions retain their own independent close lifetimes.
+        if (broadcaster := self._broadcaster) is not None:
+            await broadcaster.aclose()
+
         await self._from_aio.aclose()
 
     # ?TODO? async version of this?
@@ -324,6 +332,7 @@ class LinkedTaskChannel(
     @acm
     async def subscribe(
         self,
+        raise_on_lag: bool = True,
 
     ) -> AsyncIterator[BroadcastReceiver]:
         '''
@@ -335,6 +344,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 +357,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

tractor/trionics/__init__.py

@@ -26,6 +26,7 @@ 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,6 +100,20 @@ class Lagged(trio.TooSlowError):
     '''
 
 
+class BroadcastReceiveError(Exception):
+    '''
+    A shared underlying receiver failed in another subscriber task.
+
+    '''
+
+
+class _BroadcastReceiverClosed(Exception):
+    '''
+    An active receiver was closed while owning the source read.
+
+    '''
+
+
 class BroadcastState(Struct):
     '''
     Common state to all receivers of a broadcast.
@@ -115,6 +129,7 @@ class BroadcastState(Struct):
     # broadcast event to wake up all sleeping consumer tasks
     # on a newly produced value from the sender.
     recv_ready: tuple[int, trio.Event]|None = None
+    recv_scope: trio.CancelScope|None = None
 
     # if a ``trio.EndOfChannel`` is received on any
     # consumer all consumers should be placed in this state
@@ -122,7 +137,13 @@ class BroadcastState(Struct):
     # For now, this is solely for testing/debugging purposes.
     eoc: bool = False
 
-    # If the broadcaster was cancelled, we might as well track it
+    # 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
+
+    # Retain the latest interrupted source-reader task until its
+    # receiver next makes progress or closes.
     cancelled: dict[int, Task] = {}
 
     def statistics(self) -> dict[str, Any]:
@@ -142,13 +163,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,
         }
@@ -156,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__(
@@ -190,6 +224,8 @@ class BroadcastReceiver(ReceiveChannel):
         self._recv = receive_afunc or rx_chan.receive
         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,
@@ -237,7 +273,10 @@ class BroadcastReceiver(ReceiveChannel):
                 # https://docs.rs/tokio/1.11.0/tokio/sync/broadcast/index.html#lagging
 
                 mxln = state.maxlen
-                lost = seq - mxln
+                # `seq == mxln` is already one past the final
+                # valid deque index, so include that first
+                # displaced value in the loss count.
+                lost = seq - mxln + 1
 
                 # decrement to the last value and expect
                 # consumer to either handle the ``Lagged`` and come back
@@ -255,8 +294,21 @@ class BroadcastReceiver(ReceiveChannel):
                     return self.receive_nowait(_key, _state)
 
             state.subs[key] -= 1
+            state.cancelled.pop(key, None)
             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
+
+        if state.eoc:
+            raise trio.EndOfChannel
+
         raise trio.WouldBlock
 
     async def _receive_from_underlying(
@@ -270,14 +322,34 @@ class BroadcastReceiver(ReceiveChannel):
             raise trio.ClosedResourceError
 
         event = trio.Event()
+        recv_scope = trio.CancelScope()
         assert state.recv_ready is None
+        assert state.recv_scope is None
         state.recv_ready = key, event
+        state.recv_scope = recv_scope
 
         try:
             # if we're cancelled here it should be
             # fine to bail without affecting any other consumers
             # right?
-            value = await self._recv()
+            receive_exc: BaseException|None = None
+            with recv_scope:
+                try:
+                    value = await self._recv()
+                except BaseException as exc:
+                    receive_exc = exc
+
+            # Only this receiver's `aclose()` cancels its private
+            # source-read scope, and it marks the receiver closed
+            # first without a checkpoint. Outer task cancellation does
+            # not set `recv_scope.cancel_called`; it remains a real
+            # `trio.Cancelled` and follows the handler below.
+            if recv_scope.cancel_called:
+                assert self._closed
+            if self._closed:
+                raise _BroadcastReceiverClosed
+            if receive_exc is not None:
+                raise receive_exc
 
             # items with lower indices are "newer"
             # NOTE: ``collections.deque`` implicitly takes care of
@@ -303,6 +375,8 @@ class BroadcastReceiver(ReceiveChannel):
             ):
                 state.subs[sub_key] += 1
 
+            state.cancelled.pop(key, None)
+
             # NOTE: this should ONLY be set if the above task was *NOT*
             # cancelled on the `._recv()` call.
             event.set()
@@ -312,11 +386,20 @@ class BroadcastReceiver(ReceiveChannel):
             # if any one consumer gets an EOC from the underlying
             # receiver we need to unblock and send that signal to
             # all other consumers.
+            state.cancelled.clear()
             self._state.eoc = True
             if event.statistics().tasks_waiting:
                 event.set()
             raise
 
+        except _BroadcastReceiverClosed:
+            # `aclose()` cancelled this receiver's source-read scope.
+            # Wake peers so one of them can take ownership after this
+            # task clears `recv_ready` in `finally`.
+            if event.statistics().tasks_waiting:
+                event.set()
+            raise trio.ClosedResourceError
+
         except (
             trio.Cancelled,
         ):
@@ -329,14 +412,59 @@ 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.cancelled.clear()
+            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.
+            state.cancelled.pop(key, None)
+            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
             # was cancelled to avoid the next consumer from blocking on
             # an event that won't be set!
             state.recv_ready = None
+            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
 
@@ -362,7 +490,23 @@ class BroadcastReceiver(ReceiveChannel):
                 # seq = state.subs[key]
                 # assert seq == -1  # sanity
                 _, ev = state.recv_ready
-                await ev.wait()
+                wait_scope = trio.CancelScope()
+                self._wait_scope = wait_scope
+                try:
+                    with wait_scope:
+                        await ev.wait()
+
+                    # As with `recv_scope`, only this receiver's
+                    # `aclose()` cancels its private peer-wait scope
+                    # after marking the receiver closed. Outer task
+                    # cancellation remains `trio.Cancelled`.
+                    if wait_scope.cancel_called:
+                        assert self._closed
+                    if self._closed:
+                        raise trio.ClosedResourceError
+                finally:
+                    self._wait_scope = None
+
                 try:
                     return self.receive_nowait(
                         _key=key,
@@ -405,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:
@@ -440,18 +585,35 @@ class BroadcastReceiver(ReceiveChannel):
         if self._closed:
             return
 
-        # if there are sleeping consumers wake
-        # them on closure.
-        rr = self._state.recv_ready
-        if rr:
-            _, event = rr
-            event.set()
-
         # XXX: leaving it like this consumers can still get values
         # up to the last received that still reside in the queue.
-        self._state.subs.pop(self.key)
+        state = self._state
+        state.subs.pop(self.key)
+        state.cancelled.pop(self.key, None)
         self._closed = True
 
+        # A non-owner close must not wake peers waiting behind some
+        # other receiver's source read. If this receiver owns that
+        # read, cancel only its private scope; the owner task wakes
+        # peers after cancellation is delivered and state is ready
+        # for a clean ownership handoff.
+        rr = state.recv_ready
+        if (
+            rr is not None
+
+            # `recv_ready[0]` identifies the receiver which currently
+            # owns the one shared source read. Only that receiver may
+            # cancel `BroadcastState.recv_scope`; closing any other
+            # subscriber must not disturb the owner or its peer tasks.
+            and
+            rr[0] == self.key
+        ):
+            recv_scope = state.recv_scope
+            assert recv_scope is not None
+            recv_scope.cancel()
+        elif (wait_scope := self._wait_scope) is not None:
+            wait_scope.cancel()
+
 
 def broadcast_receiver(
 
@@ -462,6 +624,11 @@ def broadcast_receiver(
 
 ) -> BroadcastReceiver:
 
+    if max_buffer_size < 1:
+        raise ValueError(
+            '`max_buffer_size` must be greater than zero'
+        )
+
     return BroadcastReceiver(
         recv_chan,
         state=BroadcastState(