KalshiBackpressureError in recv loop breaks the loop but leaves the WS open and _running=True

[TexasCoding]

Problem

On KalshiBackpressureError (or KalshiSubscriptionError) the recv loop fans sentinels to every active subscription and breaks. The recv task is now .done(), but self._running is still True, the websocket is still open, and self._sub_mgr._subscriptions still holds every entry (sentinels close the per-sub queues but the manager bookkeeping stays — see the companion zombie-sub finding).

If the user catches the error on the affected iterator and calls another subscribe_* on the same KalshiWebSocket instance (a documented recovery path — the session re-entry guard only fires when _connection is None or not _running, neither of which is true), _ensure_recv_loop() happily creates a brand-new recv task on top of the already-orphaned server-side subscriptions. The next frames the server emits land on queues that are closed (silent drop) or get routed to brand-new sids without local snapshots. The current implicit contract — "one bad consumer kills the whole session" — is in the comment, but the implementation does not enforce it.

Evidence

• kalshi/ws/client.py:338-344 — fatal-error tear-down does NOT close the connection or flip _running:

except (KalshiBackpressureError, KalshiSubscriptionError):
    logger.error("Fatal WS error in recv loop", exc_info=True)
    await self._broadcast_sentinels()
    break

• kalshi/ws/client.py:242-243 — _ensure_recv_loop revives a finished task:

if self._recv_task is None or self._recv_task.done():
    self._recv_task = asyncio.create_task(self._recv_loop())

• kalshi/ws/backpressure.py:78 — closed queues silently drop subsequent puts:

async def put(self, item: T) -> None:
    if self._closed:
        return

Suggested fix

On the KalshiBackpressureError / KalshiSubscriptionError tear-down inside _recv_loop, flip self._running = False and await self._connection.close() (or route through the existing _stop()-equivalent cleanup so all server-side subs are unsubscribed cleanly and _subscriptions is emptied). Either way, the next subscribe_* call should fail fast with a clear error rather than resurrecting a dead session on top of orphaned server state.

Source

Round-3 independent audit (reviewer: websocket).