fix(ws): recv-loop overhaul — reconnect races, narrowed exceptions, log scrub, ConnectionManager API

[TexasCoding]

Summary

Wave 3, the largest unit of work in v1.2. Three commits:

SHA	Closes	Scope
39aa3bd	#77, #83	5 reconnect/resubscribe races unified with narrowed recv-loop exception handling
c0f9897	Refs #84 (F-O-09 only)	KalshiError.__str__ no longer leaks request URLs
a4dc672	#88	ConnectionManager.mark_streaming() public API replaces recv-loop's _set_state reach-through

Test delta: 1620 → 1631 passing (+11). 48 skipped. ruff clean, mypy --strict clean.

Unified #77 + #83 fix (39aa3bd)

5 races (#77) — each addressed inside one cohesive recv-loop rewrite:

• F-P-01 resubscribe_all per-sub isolation — each resub wrapped in try/except, failures push sentinel + drop the sub + continue (was: abort whole reconnect on first failure).
• F-P-03 _handle_reconnect holds _subscribe_lock for the entire reconnect+resubscribe sequence; concurrent user subscribe_* can no longer race the sid-remap. _sid_to_client cleared before sends.
• F-P-04 asyncio.shield around the recv→dispatch critical section (_process_frame). Cancel during dispatch no longer drops the in-flight frame.
• F-P-05 _wait_for_response catches websockets.ConnectionClosed and re-raises as KalshiConnectionError.
• F-P-08 unsubscribe pushes sentinel before mapping cleanup.

Narrowed exception handling (#83):

Class	Disposition
KalshiBackpressureError, KalshiSubscriptionError	break + sentinel-broadcast (was: silently logged+continue)
asyncio.CancelledError	propagate cleanly
json.JSONDecodeError, pydantic.ValidationError, KeyError	log+continue with exc_info=True
Everything else	now propagates with traceback (was: eaten by broad except Exception)

⚠️ Deferred to dispatcher-correctness branch (#137)

• Log info leakage: dispatch exc_info=True dumps trade data; KalshiError str() includes URLs #84 F-O-05 (dispatch.py exc_info=True leaking trade data) — lives in kalshi/ws/dispatch.py, owned by the sibling dispatcher-correctness branch. This PR's commit is Refs #84, not Closes #84, so the issue stays open until the sibling branch lands its half. The hardcoded URL leak (F-O-09) IS closed here.
• WS recv loop & dispatcher parse JSON twice + validate orderbook messages twice #86 (double JSON parse + double orderbook validate) — requires the dispatcher to accept a pre-parsed dict (option (b) in the issue) or absorb sequence-tracking + orderbook-apply (option (a)). Both require modifying dispatch.py. The recv loop already has the parsed data dict ready to forward; the sibling branch can take it from here. Issue stays open.

Wave 4 compatibility

• WS backpressure: ERROR-overflow advances last_seq, silently desyncing orderbook #78 (ERROR-overflow desync): the new policy in _recv_loop already surfaces KalshiBackpressureError via break+sentinel — Wave 4 extends this.
• WS seq-gap: only tickers[0] cleared for multi-ticker orderbook_delta subs #79 (multi-ticker seq-gap): seq tracking still happens in _process_frame per sid; the shielded structure is per-frame so Wave 4 can change the gap handler without races.

Wave 3 boundaries

No edits to kalshi/ws/dispatch.py or kalshi/ws/orderbook.py (sibling branches). One narrow exception: the recv loop calls into OrderbookManager.apply_* and the dispatcher — no signature changes, those sibling branches can evolve freely.

Closes #77
Closes #83
Refs #84
Closes #88

Test plan

• +7 tests in tests/ws/test_recv_loop_hardening.py (new) covering all 5 races
• +3 tests in tests/test_client.py + tests/ws/test_connection.py for URL-leakage regression
• +1 test in tests/ws/test_connection.py for mark_streaming() public surface
• Full suite: 1631 passed, 48 skipped
• uv run ruff check . clean
• uv run mypy kalshi/ clean (strict)

[claude]

Code Review — PR #138 (Wave 3 recv-loop overhaul)

Overall this is a well-scoped PR with clear issue traceability, good test coverage, and clean mypy/ruff output. The per-commit breakdown in the PR description makes it easy to follow the intent. A few concerns below, ranging from a correctness issue to minor polish.

---

🔴 Correctness — asyncio.shield creates an orphan task on cancellation

kalshi/ws/client.py, _recv_loop

try:
    await asyncio.shield(self._process_frame(raw))
except asyncio.CancelledError:
    # Frame has been dispatched (shield protected the inner work).
    break

The comment is incorrect. When the outer task is cancelled while awaiting asyncio.shield(...):

• Python raises CancelledError in the outer context, but the inner coroutine continues as a detached asyncio task with no held reference.
• The break exits the loop, so _pause_recv_loop considers the recv task "done" — but _process_frame is still running in the background.

This means after _pause_recv_loop returns, a new subscribe operation can begin concurrently with the still-running dispatch. _process_frame reads subscription state from _sub_mgr (read-only), so the impact is limited in practice, but it's still a race and the stated guarantee ("in-flight frame dispatched before loop exits") isn't actually met.

The standard fix is to save the inner future and await it in the handler:

inner = asyncio.ensure_future(self._process_frame(raw))
try:
    await asyncio.shield(inner)
except asyncio.CancelledError:
    await inner  # block until dispatch actually completes before breaking
    break

The existing test (test_inflight_frame_dispatched_when_recv_task_cancelled) works because it calls allow_dispatch_finish.set() before await pause_task, so the shielded task naturally finishes first. But that ordering is not guaranteed in production.

---

🟡 Observability — reconnect failures lose their traceback

kalshi/ws/client.py, _handle_reconnect

logger.error("Reconnect failed: %s", reconnect_err)

No exc_info=True, so the full traceback is swallowed. A connection drop mid-resubscribe (e.g. TCP timeout inside resubscribe_all) will produce a single line with no stack. Compare with how narrowed exceptions are logged in the recv loop itself (exc_info=True on every case). Suggest:

logger.error("Reconnect failed", exc_info=True)

---

🟡 Error messages lose useful debug context (F-O-09)

kalshi/_base_client.py and kalshi/ws/connection.py

raise KalshiError("Request timed out")
raise KalshiError("HTTP error")
raise KalshiConnectionError("WebSocket connection failed")

Stripping the httpx exception string is the right call to avoid URL/token leakage — but method and path are both in scope at all three call sites and carry no secrets. Including them would preserve useful debugging information:

raise KalshiError(f"Request timed out: {method} {path}")
raise KalshiConnectionError(f"WebSocket connection failed: {ws_path}")

The tests in TestErrorMessageDoesNotLeakUrl could be extended to assert that the method/path does appear in the message, which would document the expected level of context.

---

🟡 Inline import inside a loop body

kalshi/ws/channels.py, resubscribe_all

from kalshi.errors import KalshiSubscriptionError

This import is inside the for client_id, sub in old_subs.items() loop body. KalshiConnectionError was correctly added to the top-of-file imports in this same PR (line 54–57 of the diff) — KalshiSubscriptionError should be moved there for consistency.

---

🟠 _resume_recv_loop is pure indirection

kalshi/ws/client.py

def _resume_recv_loop(self) -> None:
    """Resume the recv loop after subscribe finishes."""
    self._ensure_recv_loop()

This is a one-line wrapper that calls a method of the same class, with no additional logic. The only production call site is one line in _do_subscribe; the only test call is in teardown. Per the CLAUDE.md simplicity guideline ("no abstractions for single-use code"), calling self._ensure_recv_loop() directly would be cleaner and eliminate one indirection. If the intent was to give tests a cleaner hook, a comment at the _ensure_recv_loop() call site would suffice.

---

🟢 Minor — mark_streaming() docstring overstates the precondition

kalshi/ws/connection.py

Public transition from CONNECTED to STREAMING.

The method calls _set_state(ConnectionState.STREAMING) unconditionally, which will succeed from any state. If it's only valid post-reconnect (CONNECTED), a guard or at least a clarifying note would prevent accidental misuse:

async def mark_streaming(self) -> None:
    """Transition to STREAMING after a successful reconnect.

    Replaces the reach-through to _set_state (#88). Valid only when
    the connection is in CONNECTED state.
    """

---

🟢 Minor — timing-based test synchronisation

A few tests use await asyncio.sleep(0.05) / asyncio.sleep(0.1) to let background tasks advance:

• test_inflight_frame_dispatched_when_recv_task_cancelled — sleep(0.05) before releasing dispatch barrier
• test_backpressure_error_breaks_loop_and_sentinels — sleep(0.1) after sending snapshot
• test_handle_reconnect_acquires_subscribe_lock — sleep(0.05) to assert lock is held

These are well-known sources of flakiness on loaded CI runners. The event-barrier approach already used in the F-P-04 test is the right pattern — the sleep in that same test could be replaced by waiting on a second event (e.g. cancel_delivered set inside a monkeypatched _process_frame). Low-risk today but worth tracking.

---

✅ What works well

• F-P-01 per-sub isolation is correct: except Exception in resubscribe_all intentionally isolates individual failures. asyncio.CancelledError is a BaseException subclass in Python 3.8+, so it won't be swallowed by that clause — clean.
• F-P-03 lock scope is right: _subscribe_lock is held for the full reconnect+resubscribe sequence in _handle_reconnect, matching the lock scope in _do_subscribe.
• F-P-08 sentinel-before-cleanup ordering in unsubscribe is correct and the test precisely captures the hang scenario.
• self._running = False on reconnect failure combined with the if not self._running: break guard after _handle_reconnect is a clean way to ensure a dead connection doesn't loop forever.
• The test file structure (one class per fix, asyncio.Event barriers for the F-P-04 race) is solid and the test descriptions are clear.

[TexasCoding]

Five items addressed in latest commit:

• 🔴 asyncio.shield orphan bug — fixed. Hold the inner task in a local; await inner on cancel before breaking so loop exit is strictly post-dispatch. The existing test's accidental-pass behavior (allow_dispatch_finish.set() happening to fire before pause completes) is no longer load-bearing.
• 🟡 Reconnect failure traceback — _handle_reconnect log now uses exc_info=True.
• 🟡 F-O-09 method/path context restored — KalshiError(f"Request timed out: POST /portfolio/orders") etc. Tightened existing leak tests to assert method + relative path DO appear while secrets/hostname still DON'T.
• 🟡 channels.py inline KalshiSubscriptionError imports — 4 sites, all hoisted to module-level.
• 🟠 _resume_recv_loop indirection — dropped; _do_subscribe calls _ensure_recv_loop() directly.

Skipped (justified):

• mark_streaming() docstring guard suggestion would be a behavior change (currently transitions from any state); leaving as-is until a real misuse pops up.
• Three asyncio.sleep(0.05)-in-tests sites flagged as future cleanup; low flake risk today, deferring to a sweep PR.

Verify: 1631 passed, 48 skipped. ruff + mypy --strict clean.

[claude]

Code Review — PR #138: recv-loop overhaul (Wave 3)

Overall: This is a well-scoped, well-documented fix for five real races and two security/observability issues. The asyncio patterns (shield + explicit await inner, lock scope, per-sub isolation) are applied correctly. Commit 4 already resolved the biggest correctness issue (the shield orphan task) and the observability gap on reconnect. A few smaller items remain.

---

🔴 Correctness (low probability, real edge case)

asyncio.shield exception during concurrent cancellation (client.py)

inner = asyncio.ensure_future(self._process_frame(raw))
try:
    await asyncio.shield(inner)
except asyncio.CancelledError:
    await inner   # <-- re-raises any exception from _process_frame
    break

If _process_frame raises json.JSONDecodeError or ValidationError in the same scheduler turn that the outer task is cancelled, asyncio.shield delivers CancelledError first; await inner then re-raises the parse error inside the except asyncio.CancelledError block, which has no handler. The recv loop crashes with a traceback instead of the intended break. Timing window is very narrow (malformed frame + concurrent pause during dispatch), so low probability — but genuine.

Fix:

except asyncio.CancelledError:
    try:
        await inner
    except Exception:
        pass  # best-effort; loop is exiting anyway
    break

---

🟡 Observability

resubscribe_all swallows unexpected exception tracebacks (channels.py:174)

except Exception as e:
    logger.warning(
        "Resubscribe failed for client_id=%d channel=%s: %s",
        client_id, sub.channel, e,
    )

The %s, e format shows the exception message but not the traceback. For the expected KalshiSubscriptionError (bad server response) that's fine. But for an unexpected AttributeError or RuntimeError caused by a programming mistake, the traceback is silently dropped, making debugging very difficult. Compare with _handle_reconnect, which correctly uses exc_info=True.

Suggested:

logger.warning(
    "Resubscribe failed for client_id=%d channel=%s",
    client_id, sub.channel,
    exc_info=True,
)

---

🟢 Minor style / doc nits

mark_streaming docstring uses wrong term (connection.py:420):

"Replaces the previous reach-through to the name-mangled _set_state"

Python name mangling applies only to __double_underscore attributes (transforms them to _ClassName__attr). _set_state (single underscore) is private-by-convention but not name-mangled. Suggest: "private _set_state".

Extra blank line in resubscribe_all (channels.py:157):

if data.get("type") == "error":
                                  # <-- extra blank line
    error_msg = data.get("msg", {})

Minor inconsistency with surrounding style.

---

🟢 Testing (acknowledged, low urgency)

TestPauseDoesNotDropFrames and TestReconnectHoldsSubscribeLock both use await asyncio.sleep(0.05) as a timing sync. As noted in commit 4, these are low CI-flake risk today but worth converting to event barriers eventually for consistency with the file's stated design goal.

---

✅ Confirmed correct

• F-P-04 shield + await inner — the orphan-task race is properly resolved; the loop cannot break until _process_frame finishes.
• F-P-03 lock scope — _subscribe_lock covers the full reconnect+resubscribe sequence, including _sid_to_client.clear() before any sends.
• F-P-01 per-sub isolation — sentinel is pushed and sub is dropped before continuing; CancelledError (not an Exception) correctly bypasses the broad handler.
• F-P-08 sentinel ordering — sentinel is pushed before mapping removal, so no iterator can hang between the two.
• F-P-05 ConnectionClosed surfacing — correctly wraps websockets.ConnectionClosed → KalshiConnectionError at the right layer.
• F-O-09 URL leak — both _base_client.py and connection.py now include method + path (no host, no query string) as debug context without leaking secrets.
• mark_streaming() encapsulation — clean single-purpose public wrapper; _recv_loop no longer reaches into ConnectionManager's state machine directly.

The shield edge case (🔴) is the only item worth addressing before merge. Everything else is polish.

[TexasCoding]

Three items addressed:

• 🔴 await inner re-raising dispatch exception inside cancel handler — wrapped in best-effort try/except Exception with DEBUG-level traceback log. Recv loop now exits cleanly even when a malformed-frame parse error races a concurrent pause.
• 🟡 resubscribe_all swallowed traceback on unexpected exceptions — switched to exc_info=True. Matches the pattern just added to _handle_reconnect.
• 🟢 mark_streaming() docstring — "name-mangled" → "private". Single-underscore isn't actually mangled in Python.

Skipped (justified):

• Extra blank line — pure cosmetic.
• 3 asyncio.sleep(0.05) test sync points — already flagged as a future sweep in the previous commit; bot called this low urgency.

Verify: tests/ws/ 215 passed; ruff + mypy --strict clean.

[claude]

Code Review — PR #138: recv-loop overhaul

Overall this is a high-quality, well-scoped PR. The races are each fixed with the right primitive (lock, shield, sentinel, isolation), the exception-narrowing is long overdue, and the test coverage is solid. A few things worth discussing before merging.

---

Issues

1. Unexpected exceptions from dispatch() leave consumers hanging (no sentinel)

_process_frame is only partially covered by the catch ladder in _recv_loop. If self._dispatcher.dispatch(raw) raises an exception that isn't KalshiBackpressureError, KalshiSubscriptionError, json.JSONDecodeError, ValidationError, or KeyError — for example an AttributeError or TypeError from a user-supplied callback — the exception propagates out of the while loop, the recv task ends in a failed state, and no sentinel is broadcast. Consumers then hang indefinitely on their queue.

The old except Exception: continue was wrong, but the fix landed halfway: fatal SDK errors get sentinels (KalshiBackpressureError/KalshiSubscriptionError branch), reconnect failures get sentinels (_handle_reconnect), but the "truly unexpected" path does not.

Suggested fix — add a finally or a broad catch-all at the bottom of the loop body, after the specific handlers:

# after the existing except blocks:
except Exception:
    logger.error("Unexpected error in recv loop", exc_info=True)
    if self._sub_mgr:
        for sub in self._sub_mgr.active_subscriptions.values():
            await sub.queue.put_sentinel()
    raise  # still propagate so the task failure is visible

This preserves the "propagate with traceback" intent from #83 while guaranteeing consumers are not silently stranded.

---

2. Orphaned blank lines inside if blocks in channels.py

Three spots where late imports were removed left blank lines as the first statement inside an if block:

# line ~92 in subscribe():
if data.get("type") == "error":
                             ← orphaned blank line
    error_msg = data.get("msg", {})

# line ~109 in update_subscription():
if not sub or sub.server_sid is None:
                             ← orphaned blank line
    raise KalshiSubscriptionError(...)

# line ~158 in resubscribe_all():
if data.get("type") == "error":
                             ← orphaned blank line
    error_msg = data.get("msg", {})

Valid Python but ruff and most style tools flag blank lines at the start of a block. Since this PR is cleaning up late imports anyway, worth fixing these three spots.

---

3. F-P-04 and backpressure tests use time.sleep despite claiming barrier-based determinism

The PR description says:

Race scenarios use asyncio.Event barriers, not sleeps, so they're deterministic.

But test_inflight_frame_dispatched_when_recv_task_cancelled has await asyncio.sleep(0.05) to give the cancel "a chance to propagate", and test_backpressure_error_breaks_loop_and_sentinels has await asyncio.sleep(0.1) to yield after the snapshot. These work in practice but are timing-based, not deterministic. They are also fine to merge as-is — just worth updating the description or adding a comment explaining why a sleep is needed there versus a barrier.

---

Suggestions (non-blocking)

_process_frame assert vs. runtime guard

assert self._dispatcher is not None

assert is compiled away with -O. Per project conventions this is fine (it's an "impossible scenario"), but given this is called in the hot path of a long-running loop, a # type: ignore[union-attr] or a proper if self._dispatcher is None: return reads cleaner and won't silently misfire under optimizer flags. Judgment call.

_handle_reconnect sets self._running = False on failure
Correct and the if not self._running: break check after the call handles it. Just noting that this makes the WS permanently dead after a reconnect failure — a user would need to reconstruct the KalshiWebSocket. That's probably the right behavior but could be called out in a docstring or the CHANGELOG.

---

What's done well

• asyncio.shield pattern is implemented correctly — holding the inner task in a local before shielding is the right way to prevent a detached background task from racing a subsequent subscribe.
• _subscribe_lock for reconnect — holding the lock across the entire reconnect+resubscribe+state-transition sequence is exactly the right scope.
• Sentinel before mapping cleanup in unsubscribe (F-P-08) — correct ordering.
• ConnectionClosed in _wait_for_response (F-P-05) — surfacing as KalshiConnectionError is the right abstraction boundary.
• mark_streaming() public API — cleaner than the previous _set_state reach-through; one-liner wrapper is the right amount of abstraction.
• URL leak fix — applied symmetrically across sync transport, async transport, and WS connection; from e preserves the cause chain.
• Tests — barrier-based where possible, structural lock test is a good pattern, URL-leak tests pin exact expected substrings.

---

Summary

The one blocking concern is the sentinel gap on unexpected exceptions from dispatch(). The orphaned blank lines are cosmetic and could be fixed in a follow-up. Everything else is solid.

Reviewed by Claude Sonnet 4.6

[TexasCoding]

Three items addressed:

• 🔴 Sentinel gap on unexpected dispatch() exceptions — fixed. Added a final except Exception: after the explicit ladder that broadcasts sentinels to every active subscription's queue, then re-raises so the task failure stays visible. New regression test (test_unexpected_exception_broadcasts_sentinels_before_raising) monkey-patches dispatcher.dispatch to raise AttributeError, asserts iterator sees StopAsyncIteration AND recv_task.exception() is the AttributeError.
• 🟦 Orphaned blank lines at start of if blocks — fixed in 4 sites (_wait_for_response, subscribe, update_subscription, resubscribe_all). Artifacts of the inline-import removal last pass.

Skipped (justified):

• assert self._dispatcher is not None vs runtime guard — bot flagged as judgment call; assert matches file's pattern.
• "Reconnect failure permanently kills the WS" — true and intentional via _running=False; will doc in CHANGELOG at release cut.
• F-P-04 / backpressure tests still use asyncio.sleep — already acknowledged as a future sweep.

Verify: tests/ws/test_recv_loop_hardening.py 8 passed (+1). ruff + mypy --strict clean.

[claude]

Code Review — PR #138: recv-loop overhaul

Overall this is a solid, well-scoped PR. The issue decomposition is clear, the commit boundaries are logical, and the test coverage is strong. A few things worth addressing before merge.

---

What's working well

• _process_frame + _handle_reconnect extraction is the right call. The old monolithic loop was impossible to reason about; these two methods make the control flow reviewable.
• Exception taxonomy in _recv_loop is correct and well-documented: CancelledError propagates, fatal infra errors break+sentinel, per-message parse errors log+continue, unexpected errors sentinel+re-raise.
• F-P-01 per-sub isolation — wrapping each resubscribe in its own try/except and pushing a sentinel on failure is the right model.
• F-P-08 sentinel ordering — pushing the sentinel before deleting _subscriptions[client_id] / _sid_to_client[sid] is the correct fix.
• URL leak fix (F-O-09) — using {method.upper()} {path} everywhere instead of interpolating the raw httpx exception string is thorough: sync transport, async transport, and WS connection all covered, tests verify the negative case.
• mark_streaming() public API — clean replacement for the _set_state reach-through.

---

Issues

1. asyncio.ensure_future — consistency + soft deprecation

kalshi/ws/client.py line 176:

inner = asyncio.ensure_future(self._process_frame(raw))

The rest of the file uses asyncio.create_task (line 131). ensure_future has been soft-deprecated since Python 3.10 in favour of create_task. In this context it wraps a coroutine so they're equivalent, but the inconsistency is a readability trap — a future reader might wonder why one was chosen over the other.

# Suggested fix
inner = asyncio.create_task(self._process_frame(raw))

---

2. Sentinel broadcast gap in the CancelledError cancel-cleanup path

kalshi/ws/client.py _recv_loop, CancelledError handler:

except asyncio.CancelledError:
    try:
        await inner
    except Exception:        # ← swallows everything, including KalshiBackpressureError
        logger.debug("Shielded dispatch raised during cancel cleanup", exc_info=True)
    break

If _process_frame raises KalshiBackpressureError while the outer task is being cancelled, the except Exception swallows it without broadcasting sentinels. Consumers on that subscription will hang. The non-cancellation path handles it correctly (break + sentinel broadcast). The cancellation path should do the same:

except asyncio.CancelledError:
    try:
        await inner
    except (KalshiBackpressureError, KalshiSubscriptionError):
        if self._sub_mgr:
            for sub in self._sub_mgr.active_subscriptions.values():
                await sub.queue.put_sentinel()
    except Exception:
        logger.debug("Shielded dispatch raised during cancel cleanup", exc_info=True)
    break

This is an edge case (cancellation firing at the exact moment of a fatal queue event), but the sentinel-before-exit invariant should hold unconditionally.

---

3. test_reconnect_holds_subscribe_lock — timing-sensitive assertion

tests/ws/test_recv_loop_hardening.py lines 238–240:

reconnect_task = asyncio.create_task(session._handle_reconnect())
await asyncio.sleep(0.05)
assert not reconnect_task.done(), "F-P-03 regression: ..."

The test file header says "Race scenarios use asyncio.Event barriers, not sleeps." This test uses a sleep to assert that reconnect_task hasn't finished. On a heavily loaded CI runner a 50 ms sleep may not be enough for the task to even start, let alone be proven blocked — so the assertion could pass vacuously (task not done because it never ran). The other tests in this file correctly use asyncio.Event for synchronisation.

A more deterministic approach: patch self._connection.reconnect to await an event, set the event after verifying the task is blocked on the lock, then release:

reconnect_started = asyncio.Event()

async def patched_reconnect():
    reconnect_started.set()
    await original_reconnect()

session._connection.reconnect = patched_reconnect
reconnect_task = asyncio.create_task(session._handle_reconnect())
await asyncio.wait_for(reconnect_started.wait(), timeout=2.0)
# At this point the task is inside reconnect, past lock acquisition — 
# or, before the lock if we instrument differently.

Alternatively, instrument the lock acquisition itself.

---

Minor notes (non-blocking)

• asyncio.get_event_loop().time() in channels.py:_wait_for_response (lines 82, 84) is pre-existing but worth flagging for a future pass: asyncio.get_running_loop().time() is the correct async-context API since Python 3.10.
• The assert self._dispatcher is not None / assert self._connection is not None pattern is consistent with the existing codebase style, so no objection — just noting that assertions are stripped by -O.
• The resubscribe_all except Exception: correctly excludes asyncio.CancelledError (since Python 3.8+ CancelledError is BaseException, not Exception). This is correct and worth calling out explicitly in the comment for the next reader.

---

Test coverage verdict

+11 tests is the right delta for this PR. Coverage of all 5 races is present. The F-P-04 barrier test is especially well-constructed. The one determinism concern is noted above (#3). Everything else is solid.

---

Summary: Fix #2 (sentinel gap in cancellation path) before merge — it's a correctness issue, not just cleanup. Fix #1 (ensure_future) is trivial. Fix #3 (sleep-based assertion) is a test reliability issue worth addressing but lower urgency.

[TexasCoding]

Three items addressed:

• 🔴 Sentinel gap in CancelledError cleanup — fixed. The previous best-effort except Exception swallowed KalshiBackpressureError / KalshiSubscriptionError during cancel cleanup. Split into two except clauses: fatal SDK errors still broadcast sentinels even on cancel; only generic Exception is best-effort logged. The sentinel-before-exit invariant now holds unconditionally.
• 🟦 ensure_future → create_task — for consistency with line 131 of the same file. create_task is the preferred form for coroutine-→-task since 3.10.
• 🟦 F-P-03 sleep-based assertion — replaced await asyncio.sleep(0.05) with a deterministic poll on session._subscribe_lock._waiters, bounded by asyncio.wait_for(2.0). A real regression fails loudly within 2s instead of passing vacuously on a loaded runner.

Skipped (justified):

• asyncio.get_event_loop().time() in _wait_for_response — pre-existing, not introduced by this PR. Future sweep.
• assert vs runtime guard — matches file convention.

Verify: tests/ws/test_recv_loop_hardening.py 8 passed. ruff + mypy --strict clean.

[claude]

Code Review — PR #138: recv-loop overhaul

This is a large, high-impact PR fixing real concurrency hazards. The overall engineering quality is high: the races are correctly identified, the fixes are targeted, and the tests are genuinely deterministic (events/barriers rather than sleeps in most cases). Below is specific feedback by area.

---

kalshi/ws/client.py — asyncio.shield pattern

The F-P-04 fix is correct, but the implementation has a subtle issue worth discussing:

inner = asyncio.create_task(self._process_frame(raw))
try:
    await asyncio.shield(inner)
except asyncio.CancelledError:
    try:
        await inner
    except (KalshiBackpressureError, KalshiSubscriptionError):
        if self._sub_mgr:
            for sub in self._sub_mgr.active_subscriptions.values():
                await sub.queue.put_sentinel()
    except Exception:
        logger.debug("Shielded dispatch raised during cancel cleanup", exc_info=True)
    break

This correctly handles the cancel path. However, the logic for non-cancel KalshiBackpressureError/KalshiSubscriptionError (line 328 in the diff) is a separate except block at the outer level — meaning these two code paths exist:

1. Normal raise: _process_frame → raises → propagates through asyncio.shield → caught by except (KalshiBackpressureError, KalshiSubscriptionError) ✅
2. Raise during cancel: _process_frame finishes after cancel → await inner raises → caught inside cancel handler ✅

The sentinel broadcast for path 1 is correctly separated. The logic is right, just the structure is dense enough that a future reader may not see both paths. A brief inline note at the outer except (KalshiBackpressureError, KalshiSubscriptionError) block pointing to the cancel handler would reduce confusion — or extract sentinel broadcast to a helper to avoid duplication. Not a bug, just a readability concern.

---

kalshi/ws/client.py — _handle_reconnect double-logs the error

logger.error("Reconnect failed: %s", reconnect_err, exc_info=True)

exc_info=True already includes the exception string in the traceback output. The %s interpolation additionally prefixes the message with the same string, resulting in it appearing twice in Sentry/log aggregators. Prefer one or the other:

logger.error("Reconnect failed", exc_info=True)

---

kalshi/ws/channels.py — resubscribe_all broad except

except Exception:
    logger.warning("Resubscribe failed for client_id=%d ...", ..., exc_info=True)
    await sub.queue.put_sentinel()
    self._subscriptions.pop(client_id, None)

The broad except Exception is intentional for isolation (F-P-01), and the PR description explains it well. One concern: asyncio.CancelledError is not a subclass of Exception in Python 3.8+ (it inherits from BaseException), so it already propagates cleanly. This is correct and no change needed — just confirming the behavior is as intended.

---

kalshi/ws/channels.py — unsubscribe ordering

await sub.queue.put_sentinel()   # F-P-08
self._sid_to_client.pop(sub.server_sid, None)
del self._subscriptions[client_id]

Sentinel before mapping cleanup is correct per F-P-08. One edge case: put_sentinel() is async and yields control. Between the yield and the mapping cleanup, an incoming frame with the old server_sid could still be routed to this subscription's queue. The frame would land after the sentinel, which means the consumer would see the sentinel first and exit — the trailing frame is silently dropped. This is the correct behavior post-unsubscribe, but it may be worth a comment if frame-after-unsubscribe semantics are documented anywhere.

---

kalshi/ws/connection.py — mark_streaming() no state guard

async def mark_streaming(self) -> None:
    await self._set_state(ConnectionState.STREAMING)

_set_state performs no validation, so calling mark_streaming() from any state (e.g., CLOSED) silently succeeds. The usage in _handle_reconnect is safe because it's only called after a successful reconnect(), but the public surface has no guard. Consider adding a pre-condition check or note that the caller is responsible, so if a future caller misuses this it fails loudly:

if self._state != ConnectionState.CONNECTED:
    raise RuntimeError(f"mark_streaming() called in state {self._state}")

---

kalshi/ws/client.py — assert in _process_frame

async def _process_frame(self, raw: str) -> None:
    assert self._dispatcher is not None

assert is stripped with python -O. For an internal invariant that's always true in practice (dispatcher is set before the recv loop starts), this is acceptable — but worth noting that if someone runs with -O the failure mode changes from AssertionError to AttributeError. Alternatively if self._dispatcher is None: return makes the invariant survive optimization flags.

---

Code comment style (CLAUDE.md convention)

Several comments embed issue/label references directly in code:

# F-P-04: shield the recv->dispatch critical section. Once a frame has been read...
# F-P-03: acquire `_subscribe_lock` for the entire reconnect+resubscribe sequence...

The CLAUDE.md convention is: "Don't reference the current task, fix, or callers... since those belong in the PR description and rot as the codebase evolves." The invariant explanations here are genuinely non-obvious and worth keeping — but the F-P-XX tags add noise once the PR merges. Suggestion: keep the why (the invariant, the constraint), drop the task reference:

# Shield the recv→dispatch critical section. Once a frame is read off the socket
# it MUST be dispatched before honoring cancellation — otherwise seq trackers miss
# the gap, ticker frames vanish, etc.

This applies throughout client.py and channels.py.

---

Tests — minor timing concern

In test_backpressure_error_breaks_loop_and_sentinels:

await asyncio.sleep(0.1)

This is the one sleep in an otherwise event-driven test suite. It works, but could be flaky on slow CI. A more robust approach would wait until the queue has an item (or the recv task processes the frame) before sending the overflow frame. Low risk on typical CI, worth noting.

---

What's excellent

• F-P-03 test using _lock._waiters is clever and deterministic. The # type: ignore[attr-defined] + comment honestly documents the tradeoff.
• F-P-04 test using an event barrier to force cancellation mid-dispatch is the right approach — deterministic race testing without sleeps.
• Import consolidation in channels.py (moving deferred from kalshi.errors import ... to top-level) is a clean improvement.
• The _handle_reconnect extraction is exactly the right scope boundary — cohesive, independently testable, lock-holding clearly documented.
• URL-leak tests are thorough: assert the secret is absent and that safe context (method + relative path) is present.

---

Summary

Blocking: None — no correctness bugs found.

Should-fix before merge:

1. Double-logging in _handle_reconnect ("Reconnect failed: %s", reconnect_err, exc_info=True → drop %s arg)
2. F-P-XX labels in code comments violate the project's CLAUDE.md comment convention — keep the invariant, drop the tag

Non-blocking suggestions:
3. mark_streaming() state pre-condition guard
4. assert self._dispatcher → guard that survives -O
5. asyncio.sleep(0.1) in backpressure test — acceptable, low flakiness risk

🤖 Generated with Claude Code