Fix reconnect startup in `_web_bs`

- restart handled handshake failures before touching cancel state
- propagate `trio.Cancelled` without shielded retry throttling
- bound initial connection waits and expose transport failures
- cover real reconnect task paths in `test_websocket.py`

Prompt-IO: ai/prompt-io/opencode/20260901T024235Z_e09e306a_prompt_io.md

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

ai/prompt-io/opencode/20260901T024235Z_e09e306a_prompt_io.md

@@ -0,0 +1,42 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: c3a1b1ba-65c6-4c8e-bd12-e596031b0eb0
+timestamp: 2026-09-01T02:42:35Z
+git_ref: e09e306a
+scope: code
+substantive: true
+raw_file: 20260901T024235Z_e09e306a_prompt_io.raw.md
+---
+
+## Prompt
+
+Address the Tuicr review by separating generic websocket reconnect
+changes from Binance-specific work and testing the real reconnect task
+hierarchy rather than a reimplementation.
+
+## Response summary
+
+Preserves structured cancellation and safe first-handshake retries in
+`_web_bs`. Adds three focused regressions in a dedicated websocket test
+module and packages them as an independent commit boundary.
+
+## Files changed
+
+- `piker/data/_web_bs.py` - restart handled handshake failures and let
+  cancellation bypass the reconnect throttle.
+- `tests/test_websocket.py` - exercise real reconnect startup, retry,
+  and cancellation task paths with controlled transport boundaries.
+- `ai/prompt-io/opencode/20260901T024235Z_e09e306a_prompt_io.raw.md`
+  - retain the review prompt, generated-code summary, and verification.
+- `ai/prompt-io/opencode/20260901T024235Z_e09e306a_prompt_io.md` - record
+  structured provenance for the generic reconnect boundary.
+
+## Human edits
+
+The user identified that these changes were generic rather than
+Binance-specific, required a separate commit and test module, requested
+coverage of the real reconnect hierarchy, and corrected code prose to
+refer to `trio` consistently. A follow-up review required the production
+code to explain why `trio.Cancelled` bypasses the broad reconnect-error
+handler and cite the regression which audits that teardown invariant.

ai/prompt-io/opencode/20260901T024235Z_e09e306a_prompt_io.raw.md

@@ -0,0 +1,33 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-09-01T02:42:35Z
+git_ref: e09e306a
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+## Prompt
+
+The user requested that generic `_web_bs` reconnect fixes and tests be
+separated from the Binance backend patch. The review also required tests
+to exercise the real reconnect task hierarchy, cite the previous faulty
+branches, and consistently name `trio` in code prose.
+
+## Generated code
+
+> `git diff HEAD~1..HEAD -- piker/data/_web_bs.py`
+
+Ensures handled handshake failures immediately restart the reconnect
+loop and lets structured cancellation bypass shielded error throttling.
+
+> `git diff HEAD~1..HEAD -- tests/test_websocket.py`
+
+Adds deterministic scheduler-real regressions for handshake startup
+timeouts, immediate rejection retries, and cancellation propagation.
+
+## Verification output
+
+- Generic and Binance targeted tests: `9 passed in 0.06s`.
+- Ruff passes for all four changed Python files.
+- `git diff --check` passes.
+- No live network probe was run.

piker/data/_web_bs.py

@@ -304,6 +304,15 @@ async def _reconnect_forever(
         ):
             log.exception('Retrying connection')
             await trio.sleep(0.5)  # throttle
+            continue
+
+        # ``trio.Cancelled`` controls task-tree teardown; it is not a
+        # reconnect failure. Let it escape before ``BaseException``
+        # so nursery cancellation never enters the shielded 200ms
+        # error throttle. Audited by
+        # ``test_reconnect_cancellation_bypasses_error_throttle()``.
+        except trio.Cancelled:
+            raise
 
         except BaseException as _berr:
             berr = _berr
@@ -417,6 +426,7 @@ async def open_autorecon_ws(
             f'Entire ws-channel disconnect due to,\n'
             f'con_err: {con_err!r}\n'
         )
+        raise
 
 
 '''

tests/test_websocket.py

@@ -0,0 +1,175 @@
+'''
+Generic websocket connection and teardown regressions.
+
+'''
+from contextlib import asynccontextmanager as acm
+
+import pytest
+import trio
+from trio_websocket import ConnectionRejected
+from wsproto.utilities import LocalProtocolError
+
+from piker.data import _web_bs
+
+
+def test_websocket_connection_startup_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    ``open_autorecon_ws()`` previously had no bound while the real
+    ``_reconnect_forever()`` task awaited its first websocket
+    handshake, so provider startup inherited a 616-second deadline.
+
+    Replace only ``_web_bs.open_websocket_url()`` with a context
+    entry that never completes. The URL-bearing ``TimeoutError``
+    proves the wrapper bounds the real reconnect task and cancels its
+    nursery instead of waiting indefinitely for
+    ``NoBsWs._connected``.
+
+    '''
+    @acm
+    async def never_connect(url: str):
+        await trio.sleep_forever()
+        yield
+
+    monkeypatch.setattr(
+        _web_bs,
+        'open_websocket_url',
+        never_connect,
+    )
+
+    async def main() -> None:
+        with pytest.raises(
+            TimeoutError,
+            match='wss://silent.invalid/ws',
+        ):
+            async with _web_bs.open_autorecon_ws(
+                url='wss://silent.invalid/ws',
+                connect_timeout=0.01,
+            ):
+                pytest.fail(
+                    'silent websocket unexpectedly connected'
+                )
+
+    trio.run(main)
+
+
+def test_immediate_websocket_failure_retries(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    ``_reconnect_forever()`` previously fell through after a handled
+    first-handshake rejection and read ``cs`` before any websocket
+    nursery had assigned it, masking the transport failure with an
+    unbound-local error.
+
+    Make the real reconnect loop receive ``ConnectionRejected`` on
+    its first context entry and ``LocalProtocolError`` on its second.
+    Removing retry sleeps fixes their order. Two attempts and the
+    propagated second error prove the first rejection restarts the
+    loop without touching connection-local cancellation state.
+
+    '''
+    attempts: int = 0
+
+    @acm
+    async def reject_url(url: str):
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise ConnectionRejected(503, [], b'unavailable')
+
+        raise LocalProtocolError('invalid websocket response')
+        yield
+
+    async def no_throttle(delay: float) -> None:
+        await trio.lowlevel.checkpoint()
+
+    monkeypatch.setattr(
+        _web_bs,
+        'open_websocket_url',
+        reject_url,
+    )
+    monkeypatch.setattr(
+        _web_bs.trio,
+        'sleep',
+        no_throttle,
+    )
+
+    async def main() -> None:
+        with pytest.raises(LocalProtocolError):
+            async with _web_bs.open_autorecon_ws(
+                url='wss://reject.invalid/ws',
+                connect_timeout=0.1,
+            ):
+                pytest.fail(
+                    'failed websocket unexpectedly connected'
+                )
+
+    trio.run(main)
+    assert attempts == 2
+
+
+def test_reconnect_cancellation_bypasses_error_throttle(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    ``_reconnect_forever()`` previously caught ``trio`` cancellation
+    in its broad ``BaseException`` handler and entered a shielded
+    200ms error throttle, delaying websocket and feed-task teardown.
+
+    Block the real reconnect task in the patched websocket context,
+    wait until that cancellation point is active, and cancel its
+    nursery. The tracked ``trio.sleep()`` replacement records any
+    retry throttle without wall time. No calls after nursery exit
+    proves structured cancellation bypasses normal error throttling.
+
+    '''
+    entered = trio.Event()
+    sleep_calls: list[float] = []
+
+    @acm
+    async def blocked_url(url: str):
+        entered.set()
+        await trio.sleep_forever()
+        yield
+
+    async def record_sleep(delay: float) -> None:
+        sleep_calls.append(delay)
+        await trio.lowlevel.checkpoint()
+
+    monkeypatch.setattr(
+        _web_bs,
+        'open_websocket_url',
+        blocked_url,
+    )
+    monkeypatch.setattr(
+        _web_bs.trio,
+        'sleep',
+        record_sleep,
+    )
+
+    async def main() -> None:
+        send, recv = trio.open_memory_channel(1)
+        nobsws = _web_bs.NoBsWs(
+            'wss://cancel.invalid/ws',
+            recv,
+            16,
+        )
+        async with (
+            send,
+            recv,
+            trio.open_nursery() as tn,
+        ):
+            await tn.start(
+                _web_bs._reconnect_forever,
+                'wss://cancel.invalid/ws',
+                send,
+                nobsws,
+                3,
+            )
+            await entered.wait()
+            tn.cancel_scope.cancel()
+
+    trio.run(main)
+    assert sleep_calls == []