feat(fix): drop-copy message flow + listener sessions (closes #425)

[TexasCoding]

Summary

Drop-copy FIX flow plus listener-session support. Kalshi drop copy is a request/response history query (identical on both products), not a streaming feed — for a live stream you use a listener session on KalshiRT. Builds on the foundation (#422), groups (#423), and order entry (#424).

What's included

• messages/drop_copy.py — EventResendRequest (35=U1, outbound: BeginExecID required, EndExecID optional), EventResendComplete (35=U2) and EventResendReject (35=U3) inbound; new EventResendRejectReason enum (1 rate-limited / 2 server-error / 3 begin-too-small / 4 end-too-large).
• messages/dispatch.py — the inbound dispatch registry (APP_MESSAGE_MODELS) + decode_app_message moved here from order_entry.py and extended to the drop-copy types, so it spans message flows cleanly (re-exported unchanged from kalshi.fix and kalshi.fix.messages — no public-API change).
• Listener sessions — FixConfig.listener_session + skip_pending_exec_reports (read-only streaming execution reports on NR/RT), plumbed into the Logon. listener_session=True requires skip_pending_exec_reports=True (validated at construction).

Usage

async with client.drop_copy(on_message=handle) as session:
    await session.send(EventResendRequest(begin_exec_id="1;1", end_exec_id="9;9"))
# handle(raw): decode_app_message(raw) -> ExecutionReport | EventResendComplete | EventResendReject

Resent reports carry new FIX sequence numbers (reconcile by the compound "int;int" ExecID); 3-hour lookback; rejects/pending (-1;-1) excluded — per the Kalshi docs.

Tests (tests/fix/test_drop_copy.py, 8 new)

U1/U2/U3 round-trips (incl. begin-only request), decode_app_message dispatch for the drop-copy types, a DC query session-integration test (send EventResendRequest → receive replayed ExecutionReport + EventResendComplete), listener logon-flag emission, and the listener-requires-skip validation.

Verification

• uv run ruff check . — clean
• uv run mypy kalshi/ (strict) — clean (152 files)
• uv run pytest tests/fix/ — 131 passed

Closes #425. Part of #402.

🤖 Generated with Claude Code

[claude]

Code Review — feat(fix): drop-copy message flow + listener sessions

Overview

Clean, well-structured PR. The main architectural decision — pulling APP_MESSAGE_MODELS / decode_app_message out of order_entry.py and into a dedicated dispatch.py — is the right call as the number of flows grows. All public APIs are re-exported without change. The stack dependency (tags and MsgType enum values from #422) is handled correctly; this PR does not need to touch tags.py or enums.py for those.

---

Issues

1. Potential race condition in test_listener_logon_emits_flags (minor, reliability)

await session.start()
logon = acceptor.first("A")  # synchronous — no until()
assert logon is not None

Every other integration test in the file uses until() to wait for messages. If session.start() does not guarantee the logon has landed at the acceptor before it returns, this test will be intermittently flaky. The companion test test_drop_copy_query_returns_execution_reports correctly uses until() for the U1 check. Suggestion:

await session.start()
await until(lambda: acceptor.first("A") is not None)
logon = acceptor.first("A")

2. event_resend_reject_reason type vs. test assertion inconsistency (minor, correctness)

EventResendReject.event_resend_reject_reason is annotated int | None, but the roundtrip test asserts:

assert back.event_resend_reject_reason == EventResendRejectReason.BEGIN_EXECID_TOO_SMALL

This passes only because IntEnum supports cross-type equality. The established pattern in this codebase (exec_type, ord_status) is to keep the field as raw int and let callers compare against the enum — so the type annotation is correct. The test should match the contract:

assert back.event_resend_reject_reason == 3  # EventResendRejectReason.BEGIN_EXECID_TOO_SMALL

---

Low-priority observations

3. FixConfig.margin(listener_session=True) not covered by the validation test

test_listener_session_requires_skip_pending only tests FixConfig.prediction(...). The margin() factory feeds the same __post_init__ validator, so one extra pytest.raises block would close the gap.

4. decode_app_message docstring references GH #432

The note "See GH #432 for surfacing decode failures to consumers" is a useful breadcrumb — just make sure that issue exists and is open.

5. No validation of the "int;int" ExecID format

EventResendRequest.begin_exec_id accepts any str. If the gateway returns a clear U3 reject for bad formats, client-side validation is not needed. Worth a conscious decision either way.

---

What is done well

• Dispatch isolation: MappingProxyType protection correctly carried forward; the import chain through messages/__init__.py → dispatch.py keeps the public API stable with no changes to call sites.
• Logon flag pattern: True if self._config.listener_session else None correctly omits the tag when false rather than sending "N".
• Listener session guard: The __post_init__ validation (listener requires skip) is the right place and gives a clear error message.
• Test depth: Roundtrips for all 3 message types + dispatch + integration query flow + logon-flag emission covers the meaningful scenarios well.
• GH [FIX] Surface inbound decode failures (decode_app_message error routing) #432 docstring reference: Explicitly calling out that swallowed decode errors are a known limitation (not a design oversight) is good engineering hygiene.
• No breaking changes: Re-exports from kalshi.fix and kalshi.fix.messages are unchanged.

---

Summary

Two actionable items: fix the potential race in test_listener_logon_emits_flags (item 1) and align the test assertion style for event_resend_reject_reason (item 2). Items 3–5 are low-priority polish. The core implementation is solid.

[claude]

Code Review

Overview

PR adds the drop-copy FIX flow (EventResendRequest / EventResendComplete / EventResendReject, 35=U1/U2/U3), a new EventResendRejectReason enum, and listener-session support (listener_session / skip_pending_exec_reports on FixConfig). The cleanest structural win is extracting APP_MESSAGE_MODELS + decode_app_message from order_entry.py into a dedicated dispatch.py so the registry can span multiple message flows without coupling.

Tests cover the round-trips, dispatch, the session-integration scenario, and the config validator. ruff, mypy --strict, and 131 FIX tests all pass.

---

Findings

1. event_resend_reject_reason typed int | None instead of EventResendRejectReason | None

kalshi/fix/messages/drop_copy.py:50

event_resend_reject_reason: int | None = fixfield(
    Tag.EVENT_RESEND_REJECT_REASON, FixType.INT, default=None
)

The docstring notes that callers must compare against EventResendRejectReason themselves. Pydantic can coerce an int to an IntEnum at parse time, which would make the type self-documenting and let callers match/compare without a manual cast. exec_type/ord_status on ExecutionReport use raw chars (pre-existing pattern), but those are FIX standard chars; this is a Kalshi custom tag with a known enum — the enum type makes sense here.

The test exploits IntEnum.__eq__ int-compatibility to pass:

assert back.event_resend_reject_reason == EventResendRejectReason.BEGIN_EXECID_TOO_SMALL

This works today but would be cleaner (and the intent clearer) if the field were EventResendRejectReason | None.

2. # type: ignore[union-attr] in integration test

tests/fix/test_drop_copy.py:113

assert acceptor.first("U1").get(Tag.BEGIN_EXEC_ID) == "1;1"  # type: ignore[union-attr]

The preceding await until(...) guarantees the message exists, but the type checker can't see that. A two-liner is cleaner and consistent with test_listener_logon_emits_flags (which does the null check):

u1 = acceptor.first("U1")
assert u1 is not None
assert u1.get(Tag.BEGIN_EXEC_ID) == "1;1"

3. listener_session validation only one-way

kalshi/fix/config.py:159

The guard correctly requires skip_pending_exec_reports=True whenever listener_session=True. But skip_pending_exec_reports=True is allowed standalone (no listener_session). If the spec treats this as a pure sub-flag of listener sessions (i.e., it has no meaning without listener_session), a symmetric check would prevent silent misconfiguration. If it's independently valid, the current one-way check is correct — but a brief comment on the field or the guard would make the intent explicit for the next reader.

4. dispatch.py docstring references GH #432

kalshi/fix/messages/dispatch.py:54

See GH #432 for surfacing decode failures to consumers.

Referencing issues in code comments tends to rot (closed, superseded, renumbered). The non-obvious invariant here is: silently swallowing a ValidationError is intentional — raising into on_message would kill the session loop. That constraint is worth one inline comment; the issue number can live in the PR description or CHANGELOG.md instead.

---

What's good

• dispatch.py extraction is the right call: order_entry.py no longer owns the cross-flow registry, and the MappingProxyType guard survives the move cleanly.
• Validation in __post_init__ (not deferred) is the right place to enforce the listener_session → skip_pending constraint.
• True if x else None pattern for Logon flags is consistent with the existing codebase pattern in session.py.
• Test structure is solid: each message type gets its own round-trip test, the dispatch test covers both inbound types, the integration test exercises the full send→receive→decode path, and the validator test covers both prediction() and margin() factories.
• Re-exports from both kalshi.fix.messages and kalshi.fix are complete and symmetric.

---

Summary

No blockers. Items 1 and 2 are the most actionable: the enum type for event_resend_reject_reason is a small change that pays forward for callers, and cleaning up the type: ignore brings the test in line with the rest of the file's style. Items 3 and 4 are documentation/defensive suggestions.