[TexasCoding]

Closes #297.

What

Re-entering an active KalshiWebSocket session (nested connect() or restart-on-error with a cached instance) silently rebuilt every manager (_connection, _sub_mgr, _seq_tracker, _orderbook_mgr, _dispatcher), orphaning the outer session's subscriptions and recv task with no error.

• _start() now refuses re-entry while _connection or _running is live, with a RuntimeError that points at the two legitimate paths (await the existing context, or build a fresh KalshiWebSocket).
• _stop() now nulls the manager refs and clears the pause/resume events AFTER awaiting connection close, so the same instance CAN be reused for a fresh connect() once the prior session has fully exited.

Tests

• test_double_start_raises — pins the overlap rejection.
• test_reconnect_after_stop_works — pins that legitimate restart still subscribes + receives a frame on the reused instance.
• Updated test_close_sets_state and the integration shutdown test to match the new clear-on-stop contract.

Full tests/ws/ suite green (364 passed).

Part of the post-v2.5 audit closure (Wave W1).

[claude]

Code Review

Overview

Solid fix for a real and well-described bug. The re-entry guard in _start() is the right approach, and nulling the manager refs in _stop() correctly enables instance reuse. Tests cover the happy path and the new failure path.

One correctness issue needs addressing before merge.

---

Bug: partial _start() failure permanently bricks the instance

The guard checks self._connection is not None, but _connection is assigned before the async connect() call that can fail:

async def _start(self) -> None:
    if self._connection is not None or self._running:
        raise RuntimeError("...already started...")

    self._connection = ConnectionManager(...)   # assigned here
    await self._connection.connect()            # can raise (auth error, network, etc.)
    ...
    self._running = True                        # never reached on failure

When connect() raises:

• self._connection is non-None
• self._running is still False
• _WebSocketSession.__aexit__ is not called (Python does not call __aexit__ when __aenter__ raises), so _stop() never runs

Result: every subsequent _start() call raises "session is already started" even though nothing is actually connected. The instance is permanently unusable — directly contradicting the PR's stated goal that "the same instance CAN be reused."

Fix — reset _connection on failure:

async def _start(self) -> None:
    if self._connection is not None or self._running:
        raise RuntimeError("...")
    try:
        self._connection = ConnectionManager(...)
        await self._connection.connect()
        ...
        self._running = True
    except:
        self._connection = None
        raise

Suggested test to pin this:

async def test_failed_connect_allows_retry(self, test_auth) -> None:
    """_start() partial failure must not brick the instance."""
    config = KalshiConfig(ws_base_url="ws://127.0.0.1:1", timeout=1.0)  # unreachable
    ws = KalshiWebSocket(auth=test_auth, config=config)
    with pytest.raises(Exception):
        async with ws.connect():
            pass
    # Instance should NOT be permanently locked
    assert ws._connection is None
    assert ws._running is False

---

Minor: inline comment block exceeds style limit

kalshi/ws/client.py — the 3-line block comment added in _stop() (# #297: clear instance state …) violates CLAUDE.md's "one short line max" rule for inline comments. The ordering constraint ("Done after close so …") is the non-obvious part worth keeping; consider collapsing to one line or moving the rationale to the commit message/PR description instead.

---

What's good

• The guard condition _connection is not None or _running is good defense-in-depth; both flags can diverge in edge cases and checking both is correct.
• Clearing _pause_pending, .clear()-ing the events, and nulling _recv_task in _stop() are all necessary for clean reuse — nothing is missed.
• test_double_start_raises correctly nests async with ws.connect() inside a live session; the match="already started" pin is precise.
• test_reconnect_after_stop_works goes further than just checking state — it actually subscribes and receives a frame on the reused instance, which is the right level of confidence.
• Updating test_close_sets_state and the integration shutdown test to match the new contract is the right call; the old assertions were testing an implementation detail that no longer holds.

---

Summary

One blocking issue: partial _start() failure leaves _connection set with no way to clean it up, permanently locking the instance. Add the try/except reset and a regression test for it. Everything else looks correct.

[TexasCoding]

Round-2 fix-up (288d78b) addresses the bot review:

• Critical: _start() now wraps the body in try/except BaseException that resets _connection, _sub_mgr, _seq_tracker, _orderbook_mgr, _dispatcher, and _running on any failure. Without this, a raise in ConnectionManager.connect() (auth/network) left _connection set while _running stayed False; since Python skips __aexit__ when __aenter__ raises, the next connect() permanently tripped the 'already started' guard.
• Added test_failed_connect_allows_retry (unreachable ws://127.0.0.1:1) to pin the invariant: _connection is None and _running is False after a failed connect.
• Trimmed the 4-line #297 comment in _stop() to a single line preserving the ordering WHY.

Verify: ruff check clean, mypy kalshi/ clean, pytest tests/ws/ 365 passed.

[claude]

Code Review

Note: A previous review on this PR flagged the partial-failure / bricked-instance bug. The current diff already addresses it with the try/except BaseException cleanup block and the test_failed_connect_allows_retry test. The comments below review the current state of the PR.

---

Overview

Clean fix for a real, well-described bug. Re-entry guard + post-close state reset is the right structure. Tests cover all three cases: overlap rejection, legitimate restart, and connect-time failure.

---

What's Good

• Guard logic is correct for asyncio. The check if self._connection is not None or self._running: is safe because there's no await between the check and the first assignment — cooperative scheduling gives no opportunity for interleaving.

• except BaseException is the right catch. asyncio.CancelledError is a BaseException subclass since Python 3.8. Catching only Exception would leave state dirty on task cancellation mid-_start(), which is exactly the scenario you need to protect against.

• _stop() cleanup is correctly sequenced. Clearing refs after await self._connection.close() ensures _broadcast_sentinels() and the recv-task teardown still have manager refs. Good ordering.

• Test coverage. test_double_start_raises, test_reconnect_after_stop_works, and test_failed_connect_allows_retry together give strong confidence in the three paths that matter.

---

Minor Issues

1. Leaked connection on post-connect constructor failure (theoretical)

In the except BaseException block:

except BaseException:
    self._connection = None   # drops reference without closing
    ...
    raise

If _connection.connect() succeeds but a subsequent constructor raises (e.g., SubscriptionManager.__init__), the underlying WebSocket is leaked — self._connection is dropped to None without calling self._connection.close(). In practice the post-connect constructors are pure synchronous object creation and can't raise, so this is theoretical. But a belt-and-suspenders close would make the contract explicit:

except BaseException:
    if self._connection is not None:
        with contextlib.suppress(Exception):
            await self._connection.close()
    self._connection = None
    ...
    raise

2. _pending_callbacks not reset in except block

If the callback registration loop fails midway (again, practically impossible since register_callback is a dict write), callbacks already registered to the now-discarded _dispatcher are silently lost while the remaining ones would be re-applied on the next _start(). Consider adding # _pending_callbacks intentionally not cleared — items retry on next connect or, if partial-registration is a concern, resetting it here too.

Neither of these is a blocker given how unlikely the failure modes are, but item 1 is worth a note in the except block at minimum.

---

Changed Assertions in Existing Tests

The test_close_sets_state and integration shutdown test changes are correct — asserting _connection is None rather than _connection.state == ConnectionState.CLOSED reflects the new contract accurately. No concern there.

---

Verdict

Approve with the above minor notes. The core logic is sound, the guard/cleanup interaction is correct, and test coverage is solid. The resource-leak edge case is too unlikely to block merge but worth acknowledging in a comment.

[TexasCoding]

Round-2 nits addressed in 9795502:

• Defensive close (item 1): added if self._connection is not None: await self._connection.close() (suppress Exception) at the top of the except BaseException block. Makes the post-connect() partial-failure contract explicit instead of relying on GC of the orphaned ConnectionManager.
• _pending_callbacks retry semantics (item 2): added a 3-line comment above the registration loop noting that the list is intentionally NOT cleared on mid-loop failure — the unregistered tail replays on the next _start, while any already-registered entries are discarded along with the failed dispatcher.

Verified: ruff, mypy kalshi/, pytest tests/ws/ (365 passed, including the 26 test_client.py cases).

[claude]

Code Review — PR #307

Overview

This PR fixes a real, subtle bug (#297): calling connect() on an active KalshiWebSocket silently clobbered all five manager refs, orphaning the running session with no error. The fix is well-scoped — guard on entry, clean up on partial failure, null managers on exit — and the tests pin all three invariants cleanly.

---

What's correct

Guard logic (_start()):
The check if self._connection is not None or self._running: covers both dimensions of "already started." In asyncio, this is race-free: self._connection = ConnectionManager(...) is a synchronous assignment (no await), so no other coroutine can interleave between the guard check and the assignment. Good.

except BaseException for partial-failure cleanup:
Using BaseException (not Exception) is the right call — it catches CancelledError and KeyboardInterrupt, ensuring the cleanup runs even if the caller is cancelled mid-connect. The belt-and-suspenders contextlib.suppress(Exception) around the close within the handler is also correct: if connect() itself succeeded but a later step failed, we need to drop the socket explicitly rather than leaking it to GC.

Ordering in _stop():
Clearing refs after await self._connection.close() means _broadcast_sentinels (which walks _sub_mgr) and the recv-task teardown still have live refs during teardown. The comment makes this explicit.

Resetting _pause_pending and the events in _stop():
Easy to miss. Without these, a reused instance could enter _start() with stale pause state from the prior session, causing _ensure_recv_loop to think it's resuming a parked loop rather than starting fresh. This is the kind of subtle state leak that would have been a nightmare to debug.

_pending_callbacks behavior on failure:
The comment is accurate. If some callbacks were registered before the failure, the new dispatcher they were registered on is discarded with the rest. On the next _start(), _pending_callbacks still contains all of them, so they replay cleanly onto the fresh dispatcher. No duplicates because the old dispatcher is gone.

---

Minor notes (non-blocking)

test_failed_connect_allows_retry port choice:
ws://127.0.0.1:1 (port 1) is almost certainly connection-refused on any CI host, but it's worth a brief comment if port 1 is somehow bound somewhere unusual. Consider ws://127.0.0.1:19999 or a similar high-numbered port, or dynamically find a free port, to eliminate even the theoretical flake. Low risk in practice.

pytest.raises(Exception) in the same test:
The # noqa: B017 note explains the intent, but it also means the test would pass if the connection raised a completely unexpected exception type unrelated to the connection failure. The invariant you're testing (state is cleared) would still be verified, so this is acceptable. Just noting it for awareness.

test_close_sets_state docstring:
The docstring now says "managers are cleared so the instance can be reused." The test name test_close_sets_state is slightly stale since the test no longer checks .state at all — it now checks that _connection is None. Not worth a rename in this PR, but worth noting for a future polish pass.

---

Test coverage

All three invariants from the bug are pinned:

• Re-entry while active → RuntimeError with clear message ✓
• Legitimate restart after clean exit → connects, subscribes, receives frame ✓
• Partial failure in _start() → state cleared so retry isn't permanently blocked ✓

The integration test and test_close_sets_state updates correctly reflect the new contract (managers cleared on stop, not just CLOSED state).

---

Verdict

Approve. Correct implementation, correct ordering, excellent test coverage. The fix is surgical — only the lines that needed to change were touched. The non-obvious decisions (BaseException, post-close ordering, event resets) are all called out in comments.