fix: reconnect after WebSocket close 1000/1001

[AlexanderZ-Band]

Why

A live agent stays offline for good after a received WebSocket close 1000 or 1001: a customer agent was down for 23h41m, and rooms it joined during the gap went unanswered until a restart. The SDK inherited the phoenix client's default ReconnectPolicy, which reads those codes as an intentional stop. On Band neither code is a stop signal. The platform's "disconnect" broadcast closes with 1001 ("reconnect expected"), and the platform answers any close the client or a proxy starts with 1000. The stop signals are a supersede push and an upgrade rejection.

What

• WebSocketClient._build_phx_client passes RECONNECT_POLICY = ReconnectPolicy(reconnect_on_normal_close=True), so a live agent reconnects after 1000/1001 and reconciles its rooms.
• A terminal supersede ends the transport:
  • record_terminal_disconnect calls end_transport(), which starts PHXChannelsClient.shutdown instead of only clearing auto_reconnect. The transport reads that flag once per close, before its backoff, so clearing it alone can lose to a reconnect.
  • WebSocketClient.run_forever waits on that shutdown task with report_owned, so a host cancelling run_forever never aborts it, and returns once it settles, including when it is entered after the supersede.
  • BandLink._on_supersede acts on the client the push arrived on, so a supersede handled while connect() is still joining also ends the transport, and connect() then leaves the link disconnected.
  • disconnect() skips leaving agent_control once the transport has ended, since the shutdown already left every topic.
• band.testing gains close_transport(link) (a client-started normal close) and FakePhoenixServer.stalled() (a slow peer). The fake server and transport helper docstrings that stated the old rule are rewritten.
• Tests:
  • Real-wire, tests/platform/linkclose/:
    • reconnect after server 1000, server 1001 and a client-started close;
    • no reconnect after a local disconnect();
    • a terminal supersede ends the session, including when it is handled while connecting or during the backoff (that test's precondition on the pinned client is guarded separately);
    • disconnecting a superseded session logs no warning;
    • run_forever entered after a supersede returns;
    • a cancelled run leaves the link able to disconnect and reconnect.
  • Model-based, test_lifecycle_model.py: a Hypothesis test generates sequences of closes, control signals (prompt and slow), supersedes and host cancellations. It checks that a superseded session never reconnects, that disconnect() always succeeds, that no hook outlives it, that signals apply in order, and that the link can connect again. It fails on each defect earlier review cycles found by hand.
  • Live E2E: the reconnect smoke gains a normal-close test beside the abort one, and a new smoke has a newer same-key session end the running one for good while it keeps replying. The manifest is regenerated.
  • hypothesis is added to the dev extras.

Behavior changes: agents now reconnect after 1000/1001. An agent whose key is disabled retries a refused (403) reconnect with backoff until the SDK adopts the phoenix-channels-python-client release that stops on a refused handshake (it used to stop silently). On-prem platforms older than 0.9.1 send no supersede push, so same-key duplicates there would keep displacing each other; those installs should upgrade.

Scope

band-sdk-python only, on the existing phoenix-channels-python-client>=0.2.5,<0.3 and band-sdk-core==2.8.0 pins. Builds on #870 (control signals off the topic processor, merged); a test-structure PR (#871) follows.

Not included:

• terminal handling of a refused reconnect: INT-1855 in the phoenix client, picked up by INT-1859;
• Session-owned reconnect and the residual window for a supersede handled after the whole backoff: INT-1355;
• adopting band-sdk-core 5.x so RECONNECT_POLICY takes its close-code rule from core's classify_close: INT-1860;
• the TypeScript SDK: INT-1858.

Supersedes draft #751 (INT-1697).

Linear: INT-1808

🤖 Generated with Claude Code

https://claude.ai/code/session_01PBt2TLCc2YD1AXxmmo61Hx

[linear-code]

INT-1808

INT-1860

[AlexanderZ-Band]

Blocking — would be REQUEST_CHANGES on a PR by someone else.

my-code-review, cycle 1: 4 issue:, 1 suggestion:, 3 nit:.

Two issues are proven defects in the new reconnect path: a supersede dropped when it waits behind a slow agent.control, and run_forever raising after a supersede. Each was reproduced against the PR head and compared with the pre-change behavior.

[Copilot]

🔵 Needs a closer look

An unresolved startup supersede race and deferred terminal-reconnect handling require maintainer review.

1 open issue

• Initial join supersede bypasses shutdown and allows stale reconnect · New

What changed in this PR

Restores automatic reconnection after normal WebSocket closes so Band agents can resume receiving room events.

Changes:

• Enables reconnect after close codes 1000 and 1001.
• Schedules transport shutdown for terminal supersedes.
• Adds normal-close helpers and regression coverage.

File	Description
tests/​websocket/​test_client.py	Updates the mock reconnect flag.
tests/​testing/​test_transport.py	Groups transport-helper tests.
tests/​testing/​test_phoenix_server.py	Removes obsolete close expectations and simplifies synchronization.
tests/​platform/​test_link_close_reconnect.py	Adds reconnect and terminal-stop regressions.
tests/​e2e/​baseline/​smoke/​behavior/​test_reconnect.py	Adds live normal-close coverage.
tests/​e2e/​baseline/​expected-cells.json	Records both reconnect scenarios.
src/​band/​testing/​transport.py	Adds a normal-close helper.
src/​band/​testing/​phoenix_server.py	Updates close-behavior docstrings.
src/​band/​testing/​__init__.py	Exports the new helper.
src/​band/​client/​streaming/​client.py	Enables normal-close reconnect and terminal shutdown.

_{Review effort: Balanced}

[AlexanderZ-Band]

Blocking — would be REQUEST_CHANGES on a PR by someone else.

Cycle 2 of /my-code-review --fix: a fresh full-PR review at 3b51c7c. One proven regression: a cancelled run_forever poisons the shared transport shutdown (client.py:1007, real-wire repro attached). There is also an observability regression for failed control hooks, plus test-quality gaps. Other hypotheses were checked and dropped with evidence. For example, a push slipping past a failed leave is unreachable (no routing task while not CONNECTED, and a timed-out leave unregisters the topic). Also dropped: a hook that stops the agent (no production chain), a supersede during the reconnect callback (no observable harm), and a hook that swallows cancellation (it hangs identically on main).

[AlexanderZ-Band]

Blocking — would be REQUEST_CHANGES on a PR by someone else.

my-code-review on cf78d3c8: 6 issue:, 2 suggestion:, 3 nit: (10 inline, 1 below). All 22 earlier threads are resolved; none of these repeats one. Two issues are proven defects, each reproduced against the PR head.

issue: src/band/platform/link.py:407 (outside this PR's diff, which stops at #870; routed here because this PR makes it reachable). _on_supersede resolves the connection through self._ws, which connect() assigns only after join_agent_control_channel returns (link.py:222). A supersede handled before that assignment takes the else branch: it records the reason and queues WebSocketDisconnectedEvent, but never calls record_terminal_disconnect, so end_transport never runs and auto_reconnect stays on. connect() then sets _ws and _is_connected = True. The drain-end 1001 close now reconnects under RECONNECT_POLICY, and the reconnect supersedes the newer same-key instance; before this PR that 1001 stopped the transport, which masked the gap.

An earlier thread on this resolved it as unreachable. The new fact is Python 3.11, which CI runs: when the supersede frame is read in the same burst as the join reply, the topic processor dispatches the queued supersede callback without yielding, and 3.11's asyncio.wait_for resumes the join waiter one loop hop later than 3.12+, so the callback runs first. Reproduced 3 of 3 on 3.11 (reconnected, server connection count 2), 0 of 3 on 3.12 and 0 of 2 on 3.13, where _ws is already set. The repro writes the reply and the push in one socket write; with two separate writes neither version reproduced in this in-process setup. In production the platform registers the control channel in Presence before replying to the join, and an older same-key connection's supersede is already in the channel mailbox when join/3 returns, so both frames are written back to back and can be read in one TCP read. Reaching it needs two same-key instances starting within about one round trip, so the window is narrow.

Root cause: a supersede belongs to the connection that received it, but the handler finds that connection through a link slot that is empty during connect(). Bind the handler to the ws it was joined on (pass ws into the callback registered in _join_agent_control_channel) so the terminal path always runs record_terminal_disconnect on the right client, and stop connect() from overwriting _is_connected after a terminal reason was recorded for that ws.

Repro (Python 3.11)

"""E2: a terminal supersede pushed right behind the agent_control join reply.

Full real stack: BandLink -> WebSocketClient -> PHXChannelsClient against the
fake Phoenix server. The fake server behaves like the platform on an eviction
of a connection that has only just registered its control channel: it replies
to the agent_control join and immediately pushes `supersede`; after the drain
it closes the socket with 1001 (what Phoenix 1.8's "disconnect" broadcast does).

Correct behaviour: the superseded link never reconnects (run_forever returns
once the transport ends). Defect: the link reconnects (ReconnectedEvent) and
would evict the newer same-key instance.
"""

from __future__ import annotations

import asyncio
import sys

from band.client.streaming.client import WireEvent
from band.platform.event import ReconnectedEvent, WebSocketDisconnectedEvent
from band.platform.link import BandLink
from band.testing import fake_phoenix_server
from band_sdk_core import AgentTopicKind
from phoenix_channels_python_client.phx_messages import PHXEvent

AGENT = "agent-123"
TOPIC = AgentTopicKind.Control.topic(AGENT)
SUPERSEDE = {
    "reason": "session.already_connected",
    "message": "Superseded by a newer session",
    "retryable": False,
    "retry_after": 15,
    "target_socket_id": f"agent_socket:{AGENT}",
    "correlation_id": "evict-1",
}


async def main() -> int:
    print("python", sys.version.split()[0])
    if "--default-reconnect-policy" in sys.argv:
        # What the transport did before this PR: a received 1000/1001 is a stop.
        import band.client.streaming.client as streaming
        from phoenix_channels_python_client.client_types import ReconnectPolicy

        streaming.RECONNECT_POLICY = ReconnectPolicy()
        print("(transport reconnect policy: library default)")
    async with fake_phoenix_server() as server:
        handle = server._handle_message

        async def reply_then_supersede(connection, message) -> None:
            if not (message.event == PHXEvent.join and message.topic == TOPIC):
                await handle(connection, message)
                return
            # The platform's transport process writes the join reply and then the
            # queued supersede push back to back; the client reads both in one
            # TCP read. Hold the socket writes so both leave in one write.
            transport = connection.transport
            held: list[bytes] = []
            transport.write = held.append
            try:
                await handle(connection, message)
                await server.push(TOPIC, WireEvent.SUPERSEDE, SUPERSEDE)
            finally:
                del transport.write
                transport.write(b"".join(held))

        server._handle_message = reply_then_supersede

        link = BandLink(
            agent_id=AGENT, api_key="k", ws_url=server.url, rest_url="https://test.invalid"
        )
        seen_ws_none: list[bool] = []
        on_supersede = link._on_supersede

        async def spy(payload) -> None:
            seen_ws_none.append(link._ws is None)  # read-only diagnostic
            await on_supersede(payload)

        link._on_supersede = spy
        await link.connect()

        first = await asyncio.wait_for(anext(link), 10)
        print("BandLink._ws was None when the supersede was handled:", seen_ws_none)
        assert isinstance(first, WebSocketDisconnectedEvent), first
        print("supersede handled; link.last_disconnect_reason set:",
              link.last_disconnect_reason is not None,
              "| link.is_connected:", link.is_connected)

        run = asyncio.create_task(link.run_forever(install_signal_handlers=False))
        await server.close_connection(code=1001)  # drain-end close after the push

        async def next_reconnect() -> None:
            while not isinstance(await anext(link), ReconnectedEvent):
                pass

        reconnect = asyncio.create_task(next_reconnect())
        done, pending = await asyncio.wait(
            {run, reconnect}, timeout=20, return_when=asyncio.FIRST_COMPLETED
        )
        outcome = "run_forever returned" if run in done else (
            "RECONNECTED" if reconnect in done else "neither within 20s"
        )
        print("after the 1001 close:", outcome, "| server connections:", server.connection_count)
        for task in pending:
            task.cancel()
        await link.disconnect()
        if reconnect in done:
            print("DEFECT: superseded agent reconnected after the 1001 close "
                  "(on the real platform this evicts the newer same-key instance)")
            return 1
        print("OK: superseded agent did not reconnect")
        return 0


sys.exit(asyncio.run(main()))

[AlexanderZ-Band]

Re the finding in the review body on link.py:407 (a supersede handled while connect() is still joining skipped ending the transport, so the closing 1001 reconnected):

Fixed in 4e7dd19. BandLink._on_supersede now receives the WebSocketClient the push arrived on instead of reading _ws, which connect() assigns only after the control channel is joined. test_a_supersede_handled_while_connecting_ends_the_session delivers a supersede over the wire inside connect() and requires run_forever to return with one connection; it fails without the change. The earlier repro (Python 3.11, 3 of 3) no longer reproduces.

[AlexanderZ-Band]

Blocking — would be REQUEST_CHANGES on a PR by someone else.

Cycle-1 review of #869 at 7c6c550, after the rebase onto main. Blocking: two tests that can't fail or can't stop. The cancelled-run test survives an unshielded await of the shutdown task, and the model runs for minutes on a regression. Two more issues: real backoff timers drive the reconnect tests, and the scorecard records the reconnect cells under the wrong adapter.

The skeptic dropped five hypotheses, each with a repro:

• Teardown stall after a supersede: no phx_leave is sent once the state is SHUTTING_DOWN, so teardown is instant.
• Supersede during the initial join: connect() always resumes before the handler runs.
• Eviction with no push received: the mechanism is real, but the platform's 30 s supersede cooldown answers the reconnect with 429, so the outcome matches main.
• _on_reconnected before _ws exists: unreachable for the same reason.
• A shutdown that hides a terminal error: not proven.

[AlexanderZ-Band]

Blocking — would be REQUEST_CHANGES on a PR by someone else.

Cycle 2. The cycle-1 fixes hold: every mutation (no RECONNECT_POLICY, flag-only end_transport, no aenter reset, _on_supersede on self._ws, unshielded join) fails at least one test. The skeptic reproduced and dropped three hypotheses: the double-reported shutdown failure (_shutdown cannot raise), the cancelled _on_reconnected (nothing observable) and the connect overwrite as a regression (base has it too).

Outside the diff hunks:

• suggestion: src/band/platform/link.py:236 (C). Once a terminal supersede has ended the transport and run_forever has returned, disconnect() still leaves agent_control, which the shutdown already unsubscribed. Every superseded agent's shutdown then logs WARNING Error unsubscribing from agent_control: Topic agent_control:… not subscribed. Base never reached this, because run_forever didn't return after a supersede. Repro: scratchpad/review869/skeptic2/hb.py. Fix: skip the leave once the transport has ended.
• suggestion: src/band/platform/link.py:226 (C). On Python 3.11 the supersede handler can run before connect() resumes, when the join reply and the push land in one read (6 of 15 runs in skeptic2/ha2.py). connect() then sets _is_connected = True over a CLOSED transport while last_disconnect_reason says superseded. Base has the same overwrite over a live transport, so this isn't a regression, but now the flag is plainly wrong. Fix: self._is_connected = ws.last_disconnect_reason is None, and assert is_connected in test_a_supersede_handled_while_connecting_ends_the_session.

[AlexanderZ-Band]

Review-body items from review 5581156359, both fixed in 26567c6:

• link.py:236: disconnect() now skips leaving agent_control once ws.transport_ended is true. The new test_disconnecting_an_ended_session_logs_no_warning fails when the guard is removed.
• link.py:226: connect() now sets _is_connected = not ws.transport_ended. test_a_supersede_handled_while_connecting_ends_the_session now asserts not link.is_connected and fails when the line is reverted.

Gate: 9019 unit and 33 doc tests pass; ruff and pyrefly are clean. Live E2E against the platform: the supersede smoke, the abort reconnect and the normal-close reconnect all pass.