fix(ws): dispatcher correctness — fan-out, server-unsub cleanup, error envelope surface

[TexasCoding]

Summary

Wave 3 — three dispatcher correctness fixes in kalshi/ws/dispatch.py. Three commits, single PR.

#80 — callback + iterator on same channel ⚠️ semantic change

Before: registering a callback for a channel silently suppressed queue delivery; the iterator stopped receiving.
After: messages fan out to both. A WARNING fires at register_callback time if an active subscription already exists, since some users may have relied on the suppress-side-effect.

Test impact: the existing test_callback_mode asserted qsize == 0; updated to qsize == 1 to match fan-out. Callback-only users now retain queued messages until queue overflow (DROP_OLDEST safely drops; ERROR-strategy channels would raise after maxsize=1000).

#81 — server-initiated unsubscribe reaps sid mappings

_sid_to_client previously leaked entries on server-initiated unsubscribe (only client-initiated unsubscribe ran the full cleanup). Subsequent re-subscriptions could collide. Server-side unsubscribe now runs the same teardown path.

#82 — channel-level error envelopes surface via on_error

Error envelopes carrying a sid the dispatcher no longer recognized (post-unsubscribe race) were swallowed. Now routed through the dispatcher's on_error hook, with a fallback WARNING log if no hook is registered.

⚠️ Wave 3 cross-cutting concern

The server-unsubscribe handler should also reset SequenceTracker._last_seq[sid] to fully close the leak. The dispatcher now accepts an optional seq_tracker=None constructor kwarg, but kalshi/ws/client.py (owned by the recv-loop-overhaul branch #77/#83/#84/#86/#88) wires the tracker and still doesn't pass it through. The sid-mapping leak — the primary bug — IS fully fixed here; the seq-tracker reset is a one-line wiring change to apply when integrating with the recv-loop branch (client.py:97-100 area).

I'll handle this wiring during the post-Wave-3 integration when both branches merge.

Wave 3 boundaries

No touches to kalshi/ws/client.py, channels.py, or orderbook.py.

Closes #80
Closes #81
Closes #82

Test plan

• WS suite: 208 → 215 (+7 new tests across the three commits)
• Full suite: 1629 passed, 48 skipped
• uv run ruff check . clean
• uv run mypy kalshi/ clean (strict)

[claude]

Code Review — PR #137: Dispatcher correctness (fan-out, server-unsub, error surface)

Overall this is solid, targeted work. The three bugs are real, the fixes are minimal, and the test coverage is thorough. A few things worth addressing before merge.

---

#80 — Fan-out (remove else:)

The fix itself (await sub.queue.put(parsed) unconditionally) is clean and correct. One concern:

Queue accumulation for callback-only users. Any existing user who registered a callback but never consumed the queue will now silently accumulate up to 1000 messages. The WARNING at register_callback time is the right signal but it fires at registration, not continuously. A user who registered the callback before this SDK version and upgrades won't see any new warning. Consider noting this as a breaking behavioral change in CHANGELOG.md.

---

#81 — Server-initiated unsubscribe cleanup

The logic is correct. One concern and one nit:

if "error" in data: check in dispatch() is too broad — this is also relevant to #82 but surfaces here because of the dispatch ordering. Any future message that legitimately includes "error": null (e.g. an optional error field in a typed envelope) would be routed to _surface_channel_error instead of being parsed normally. Change to:

if data.get("error") is not None:
    await self._surface_channel_error(msg_type, data)
    return

This makes the intent explicit and avoids the false-positive on "error": null.

Layer violation: accessing self._sub_mgr._sid_to_client and self._sub_mgr._subscriptions directly.

# dispatch.py:107-108
client_id = self._sub_mgr._sid_to_client.pop(sid, None)
sub = self._sub_mgr._subscriptions.pop(client_id, None)

The dispatcher is now coupled to SubscriptionManager's internal data layout. Renaming either attribute in channels.py would silently break this with no type-checker warning (they're Any at access time). The PR description acknowledges Wave 3 boundaries, which is fair — but flagging for the integration cleanup: SubscriptionManager should get a remove_by_sid(sid) -> Subscription | None method so this isn't touching privates.

---

#82 — Channel-level error surface

_surface_channel_error silently drops when synthesis fails. When ErrorMessage.model_validate() raises, the function logs a WARNING and returns without calling on_error. A user with an on_error handler registered gets no programmatic signal for that message — they only get a log line:

except Exception:
    logger.warning(
        "Channel-level error envelope (type=%s) could not be "
        "coerced to ErrorMessage: %s",
        msg_type, data,
    )
    return   # ← on_error never called

Since the intent is to surface errors to on_error, consider calling it anyway with the raw data or a sentinel, or at minimum logging at ERROR level (not WARNING) since a registered handler was unable to be notified.

---

Tests

Test coverage is good — the 7 new regression tests cover the important paths. Two nits:

1. import inside test methods (import logging and from kalshi.ws.sequence import SequenceTracker) should be at module level.

2. FakeSubManager collapses client_id == sid == server_sid. This simplification is understandable for unit tests, but it means none of the new tests exercise the real-world case where client_id != server_sid (which is the normal case in production). The _handle_server_unsubscribe logic does _sid_to_client.pop(sid) → client_id → _subscriptions.pop(client_id), so the two-step lookup is tested but never with different keys. Consider adding one test where client_id != server_sid to cover the actual mapping path.

---

Summary

Severity	Item
Medium	if "error" in data: → should be if data.get("error") is not None:
Medium	Direct access to _sid_to_client / _subscriptions — track as tech debt for integration cleanup
Low	_surface_channel_error synthesis failure swallows error silently
Low	FakeSubManager doesn't test client_id != server_sid
Nit	import inside test methods → move to module level

The fan-out and server-unsub fixes are ready as-is. The "error" in data issue is the one I'd want addressed before merge since it introduces a potential regression under unusual but valid server payloads.

[TexasCoding]

Bot feedback addressed:

Fixed

• 🟡 "error" in data → data.get("error") is not None — prevents a legitimate "error": null field on a typed envelope from being misrouted. Regression test (test_null_error_field_not_misrouted) pins it.
• 🟦 _surface_channel_error synthesis failure now still fires on_error — falls back to ErrorMessage.model_construct(...) with the raw payload wrapped in ErrorPayload(code="unknown", msg=str(data)) so a registered handler always sees a signal. Log level escalates WARNING → ERROR since the handler almost wasn't called. New test: test_channel_level_error_validation_failure_still_fires_handler.
• 🟦 FakeSubManager now supports client_id != server_sid — new test test_server_unsubscribe_with_distinct_client_id exercises the real production mapping path that the default sid==client_id collapse was hiding.
• 🟦 Inline import logging / import SequenceTracker — moved to module level.
• 🟦 CHANGELOG — [Unreleased] → Changed entry calling out the WS callback + iterator on same channel: iterator silently never receives #80 fan-out behavioral change so upgrading users see it.

Deferred (tracked)

• 🟡 _sid_to_client / _subscriptions layer violation — SubscriptionManager.remove_by_sid() is the right abstraction. channels.py is recv-loop-branch territory (fix(ws): recv-loop overhaul — reconnect races, narrowed exceptions, log scrub, ConnectionManager API #138). Tracked in the post-Wave-3 integration cleanup task — will land alongside the seq_tracker wiring.

Verify: tests/ws/test_dispatch.py 30 passed. ruff + mypy --strict clean.

[claude]

Code Review — Wave 3 Dispatcher Correctness (#80, #81, #82)

Overall this is a solid, well-structured PR. The three bugs are real, the fixes are targeted, and the test coverage is thorough (7 new tests). A few items worth addressing before merge.

---

Issues

1. Encapsulation breach in _handle_server_unsubscribe (medium risk)

dispatch.py directly mutates private attributes of SubscriptionManager:

client_id = self._sub_mgr._sid_to_client.pop(sid, None)
sub = self._sub_mgr._subscriptions.pop(client_id, None)

SubscriptionManager already owns the lifecycle of these dicts — it is the authoritative place to pop from them. If the recv-loop overhaul (#77/#83) touches either dict's structure, this dispatcher code breaks silently. Recommend adding a SubscriptionManager.reap(sid: int) -> Subscription | None method that encapsulates the pop-both-dicts-plus-sentinel pattern, and calling it from the dispatcher. The dispatcher's _handle_server_unsubscribe should only need the returned Subscription to do its logging.

The test FakeSubManager exposes the same coupling — it must now mirror the private dict names to make the dispatcher's cleanup paths work.

2. code="unknown" is a type violation in the model_construct fallback

In _surface_channel_error:

msg=ErrorPayload.model_construct(code="unknown", msg=str(data)),

ErrorPayload.code is typed int. model_construct bypasses Pydantic validation so this doesn't raise at runtime, but it's semantically wrong and will confuse any on_error handler that inspects err.msg.code. Use code=0 (or a sentinel like -1) instead.

3. Fan-out warning has a silent ordering gap

The warning in register_callback fires only when a subscription already exists at registration time:

if any(sub.channel == channel for sub in self._sub_mgr.active_subscriptions.values()):
    logger.warning(...)

The reverse case — callback registered first, then subscribe() called — produces no warning, yet the fan-out behavior is identical. Users who set up their callback before subscribing get silent queue accumulation. Consider also warning (or documenting) this in SubscriptionManager.subscribe(), or at dispatch time on first fan-out delivery.

4. OverflowStrategy.ERROR users can crash the recv loop (acknowledged but under-documented)

await sub.queue.put(parsed) is now unconditional. MessageQueue.put() raises KalshiBackpressureError when ERROR strategy hits maxsize. Since dispatch() doesn't catch this, the exception propagates to the recv loop and can crash it. This was impossible before (callback suppressed the put). The PR description mentions it, but the warning at register_callback time should explicitly name this risk so users know they need to either switch to DROP_OLDEST or consume the queue.

5. data.get("error") called twice in synthesis expression

Minor style issue in _surface_channel_error:

"msg": data.get("error")
    if isinstance(data.get("error"), dict)
    else data.get("msg"),

Assign err_val = data.get("error") once and use it:

err_val = data.get("error")
synthesized = {
    "id": data.get("id", 0),
    "type": "error",
    "msg": err_val if isinstance(err_val, dict) else data.get("msg"),
}

---

What's Good

• Fan-out fix (WS callback + iterator on same channel: iterator silently never receives #80): The change from if/else to unconditional await sub.queue.put(parsed) is exactly right. The warning at register_callback time is a good ergonomic touch.
• data.get("error") is not None (not "error" in data): correctly handles "error": null without misrouting. The test_null_error_field_not_misrouted test makes the intent explicit.
• _handle_server_unsubscribe sid-lookup flexibility: checking both top-level sid and msg.sid is robust. Both shapes are covered by tests.
• seq_tracker.reset(sid) ordering: reset happens after the _sid_to_client.pop regardless of whether client_id was found — correct, since reset is idempotent on unknown sids (_last_seq.pop(sid, None)).
• put_sentinel() after server unsubscribe: waking held iterators on server-side teardown is the right move; the StopAsyncIteration assertions in the tests confirm it.
• test_server_unsubscribe_with_distinct_client_id: the comment "collapses sid == client_id lets the test pass by accident even if the lookup is wrong" is exactly the kind of reasoning that makes a test trustworthy.
• Test coverage: 7 tests across 3 bugs, including edge cases (null error field, validation failure fallback, unknown sid no-crash). Well done.

---

Summary

The core fixes are correct. Items 1 and 2 are the most important to address before merge — item 1 because it creates a maintenance landmine during the recv-loop integration, and item 2 because it puts a runtime type violation into a code path that's already handling a degraded state. Items 3–5 are polish.

[claude]

Code Review — PR #137: Dispatcher correctness (fan-out, server-unsub, error surface)

Solid, targeted work. All three bugs are real, the fixes are minimal, the test suite is thorough (especially the distinct-client_id/sid test for #81). A few things to address.

---

Correction to previous review comment

The earlier comment stated that _surface_channel_error silently drops when validation fails — this is incorrect in the current code. The await self._on_error(error) call sits outside the try/except block, so it always fires. When model_validate raises, the except branch falls back to model_construct and the handler is still called. The test_channel_level_error_validation_failure_still_fires_handler test explicitly asserts this. The current implementation is correct.

---

#80 — Fan-out

The fix is correct and minimal. One follow-up: the WARNING fires at register_callback time but not the inverse — a user who subscribes after registering a callback gets no warning. The current warning only catches register_callback(after subscribe), not subscribe(after register_callback). Low risk since the CHANGELOG entry covers the behavioral contract, but worth a comment in register_callback's docstring that the reverse ordering is also fan-out.

---

#81 — Server-initiated unsubscribe cleanup

The private-attribute access is the main structural concern:

# dispatch.py
client_id = self._sub_mgr._sid_to_client.pop(sid, None)
sub = self._sub_mgr._subscriptions.pop(client_id, None)

SubscriptionManager already has unsubscribe(client_id) which does this cleanup (channels.py:164–165). The dispatcher could call it instead of touching privates directly, e.g.:

client_id = self._sub_mgr._sid_to_client.get(sid)
if client_id is not None:
    self._sub_mgr.unsubscribe(client_id)  # handles both dicts + sentinel

This removes the structural coupling and means a rename of _sid_to_client/_subscriptions in channels.py won't silently break the dispatcher. The PR description acknowledges Wave 3 boundaries — tracking this as a follow-up issue before the Wave 4 integration would be the right call.

---

#82 — Channel-level error surface

Asymmetry with top-level "error" type handling. When msg_type == "error" and on_error is None, the error is silently dropped (no log). When a channel-level error has no handler, a WARNING is logged. The asymmetry is pre-existing, but #82 makes it more visible. Consider adding a warning for the top-level drop path as well:

if msg_type == "error":
    if self._on_error is not None:
        error = ErrorMessage.model_validate(data)
        await self._on_error(error)
    else:
        logger.warning("Received error frame with no on_error handler: %s", data)
    return

Also minor: data.get("error") is evaluated twice in _surface_channel_error:

"msg": data.get("error")
if isinstance(data.get("error"), dict)   # second call
else data.get("msg"),

Cache it: err_val = data.get("error").

---

CHANGELOG

The CHANGELOG.md diff only documents #80. The #81 and #82 fixes are user-visible: #81 prevents reconnect collisions after server-initiated unsubscribes, and #82 surfaces errors that were previously silently swallowed. Both belong in the ### Fixed section.

---

Tests

Well-designed. Highlights:

• test_server_unsubscribe_with_distinct_client_id is the most important one — it exercises the production path where server_sid != client_id and would catch an accidental sid == client_id assumption.
• test_null_error_field_not_misrouted correctly documents the data.get("error") is not None contract.
• FakeSubManager update properly mirrors the production structure.

One gap: no test for registering a callback before subscribing, then subscribing — the subscribe-after-callback fan-out path. Given #80's warning only fires for the opposite ordering, this path gets no coverage and no warning.

---

Summary

Severity	Issue
Should fix	CHANGELOG missing #81 and #82 entries
Should fix	Layer violation on _sub_mgr._sid_to_client/_subscriptions (consider unsubscribe(client_id) or track as follow-up issue)
Nit	data.get("error") called twice in _surface_channel_error
Nit	Asymmetric logging for top-level "error" type with no handler
Nit	No warning / test for subscribe-after-register-callback fan-out path

None of these are blockers — the bugs are fixed and the behavior is correct. The layer violation is the one worth resolving before Wave 4 integrates client.py.