refactor: unify the three skip filters and make one-shot narrowing structural (#738)

* refactor: unify the three skip filters and make one-shot narrowing structural

Closes the remaining one-shot/decoder follow-ups in
plans/claude-md-knowledge-graph.md.

TickDecoder::MESSAGE_TYPE becomes RESPONSE_MESSAGE_IDS and the tick
driver filters with the same is_undeclared helper as the two
subscription drivers — one skip decision, not three. A tick type
declares one entry; the arity was the only difference.

New request_helpers::expect_proto(expected, decode_proto) pairs the
message type a one-shot asked for with its payload decoder. Narrowing
was opt-in: each domain hand-wrote a decode_*_message wrapper for it and
24 sites had none, decoding whatever arrived on the shared channel. 44
call sites now narrow by construction; 30 decoder functions deleted (8
dispatchers, 22 require_proto wrappers). The two wsh wrappers stay —
dual-use with a StreamDecoder.

Converts the last seven hand-rolled folds: head_timestamp,
histogram_data, market_depth_exchanges, historical_schedule (all
sync+async) plus the four option-computation sites via a new
fold_one_shot_mut that fold_one_shot delegates to. Only
historical_data's fetch stays hand-rolled — it reads two frames.

Behaviour: those four APIs retry a connection reset 3 times instead of
unboundedly (async head_timestamp recursed, the others looped), and
sync/async now agree on what a closed stream means.

Two test fixtures were carrying wrong message-type prefixes (MarketRule
87 vs 93, news providers a non-numeric literal) — invisible until
narrowing read the field no assertion depended on.

* refactor: gate the one-shot pairing, sweep the sites the first pass missed

/simplify findings.

The sweep had missed wsh (4 sites) and next_valid_order_id (2), so the
"every one-shot narrows" claim was false while they existed.
next_valid_order_id is the instructive one: it reads the shared
RequestIds channel through fold_one_shot with a bare decoder — adopting
the fold helper reads as adopting the convention, and isn't.

Nothing gated the (IncomingMessages, decoder) pair. A review mutated
three sites to mismatched pairs and only incidental round-trip tests
failed. Adds test_expect_proto_sites_match_the_roster: every site
scraped from src/, checked against a PAIRS roster both directions, each
pair required exactly twice so sync and async cannot drift. The same
mutation now fails by name.

Deletes 22 tests that were retargeted onto expect_proto in the previous
commit. In each the decoder argument is never invoked — expect_type or
require_proto fails first — so they asserted only the combinator's own
behaviour and would pass with |_| Ok(()). Three canonical tests plus the
roster replace them.

Reverts one thing from the previous commit: removing expect_type from
the two single-type StreamDecoder::decode impls. A decoder with no match
has no `_ =>` arm, so expect_type IS its backstop —
test_response_message_ids_match_decode_arms went red immediately. The
misleading doc comment that invited the removal was the actual defect.

Adds docs/rules/wire/one-shot-narrowing.md — the convention had shipped
as a doc comment on a pub(crate) fn, which rules-check cannot see.
Records on proto-only-decoding that TickDecoder got the const's name but
not its gate, with the follow-up to close it.

expect_type's error now names the expected type; it was the only runtime
signal for a mispairing and withheld the identifying fact.

Author: wboayue

CHANGELOG.md

@@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - A text-framed message reaching a proto-only decoder now fails the subscription instead of being skipped. At `server_versions::PROTOBUF_REST_MESSAGES_3` every message with a proto decoder arrives proto-framed, so reaching this means the gateway broke protocol — previously the message was dropped and the subscription silently yielded nothing. Wrong-message-type frames are still skipped, which is what shared channels need (#731).
 - Historical tick and histogram sizes are now `Option<f64>` instead of `i32`: `TickMidpoint.size`, `TickLast.size`, `TickBidAsk.size_bid`/`size_ask`, and `HistogramEntry.size`. IBKR models these as decimals on the wire, and the old `i32` parse silently truncated fractional sizes — a crypto tick of `0.5` decoded as `0`. `None` means TWS sent no value (field absent, empty, or an "unset" sentinel); `Some(0.0)` is a real zero. This also changes the serialized shape — a size is now `100.0` rather than `100`, absent is `null` rather than `0`, and the `utoipa` schema becomes a nullable `number` (#716).
 - `ContractDetails.min_size`, `size_increment`, and `suggested_size_increment` are now `Option<f64>` instead of `f64`. Contracts without size rules omit these on the wire, where the old `0.0` was indistinguishable from a real value and a `size_increment` of `0.0` is nonsense (#716).
+- Every one-shot request now narrows the inbound frame to the message type it asked for before decoding it. Narrowing used to be opt-in — each domain hand-wrote a `decode_*_message` wrapper for it, and 26 of the 50 sites had no wrapper, decoding whatever arrived on the shared channel. A foreign frame now surfaces as `Error::UnexpectedResponse` naming both the expected and received type, rather than being fed to the wrong payload decoder — where overlapping protobuf field numbers usually produce a plausible struct full of wrong values instead of an error. Affects `next_valid_order_id` in particular, which reads the shared `RequestIds` channel (#738).
+- `head_timestamp`, `histogram_data`, `market_depth_exchanges`, and `historical_schedules(..).fetch()` retry a connection reset at most three times instead of unboundedly. The async `head_timestamp` recursed on a closed stream and the async `histogram_data`, `market_depth_exchanges`, and schedule fetch looped on one; a gateway that keeps resetting would hang the call rather than return. They also now agree with their blocking twins on what a closed stream means: `Error::UnexpectedEndOfStream` for `head_timestamp` and the schedule fetch, an empty list for `histogram_data` and `market_depth_exchanges` (#738).
 
 ### Removed
 

CLAUDE.md

@@ -54,6 +54,9 @@ precedents. Every rule now lives in a node; nothing is inline.
 - **Adding a `ResponseMessage` accessor, or a public API on a proto inbound message type** →
   [proto-aware accessors](docs/rules/wire/proto-aware-accessors.md)
 - **Typing a `String` field as an enum** → [wire enum typing](docs/rules/wire/enum-typing.md)
+- **Adding a one-shot client method, or passing a processor to a `one_shot_*` helper** →
+  [one-shot narrowing](docs/rules/wire/one-shot-narrowing.md) — the `(message type, decoder)`
+  pair is gated by a hand-listed roster, not by the type system
 
 ### Code structure and style
 

docs/rules/wire/one-shot-narrowing.md

@@ -0,0 +1,78 @@
+---
+id: one-shot-narrowing
+title: One-shot requests narrow through expect_proto
+cluster: wire
+status: active
+triggers:
+  - adding a one-shot client method
+  - passing a processor to one_shot_request or one_shot_with_retry
+  - adding a decode_*_proto sibling for a one-shot response
+symbols: [expect_proto, one_shot_request, one_shot_with_retry, one_shot_request_with_retry, fold_one_shot, expect_type, PAIRS]
+related: [proto-only-decoding, proto-aware-accessors, fixture-builders]
+precedents: ["#736", "#738"]
+memory: [project_protobuf_only, feedback_request_id_index_registration]
+---
+
+A one-shot request reads one frame off a **shared** channel, so it must check that the frame is
+the one it asked for. Do that with `request_helpers::expect_proto`, never with a bare decoder:
+
+```rust
+request_helpers::blocking::one_shot_request_with_retry(
+    self,
+    encoders::encode_request_user_info,
+    expect_proto(IncomingMessages::UserInfo, decoders::decode_user_info_proto),
+    || Err(Error::UnexpectedEndOfStream),
+)
+```
+
+The processor argument is the only place the expected type appears, so a new one-shot API cannot
+narrow "later" — there is no wrapper to add it to. Give the decoder a `decode_*_proto(&[u8])`
+sibling if it lacks one; the message-level `decode_x(&ResponseMessage)` form exists only for
+`StreamDecoder` impls now.
+
+**Nothing in the type system enforces this**, and treating the signature as the guarantee is the
+mistake this node exists to prevent. `expected` and `decode` are unrelated — `R` is inferred from
+the decoder — so `expect_proto(IncomingMessages::UserInfo, decode_family_codes_proto)` compiles
+and feeds one message's bytes to another's prost type. The helpers also still accept a bare
+`impl Fn(&ResponseMessage)`, so skipping the narrow compiles too.
+
+The gate is `test_expect_proto_sites_match_the_roster` (`src/common/one_shot_pairing_tests.rs`).
+It scrapes every `expect_proto` site out of `src/` and checks each pair against a declared
+`PAIRS` roster, both directions, plus `SITES_PER_PAIR == 2` — sync and async each spell the pair
+once, and a count of 1 means they have drifted. **Adding a one-shot API means adding a `PAIRS`
+line.** That is the standing cost of keeping the pairing outside the type system.
+
+Do not reach for `expect_proto` inside `impl StreamDecoder::decode`; see
+[proto-only decoding](proto-only-decoding.md) for what that surface owes instead.
+
+## Why
+
+Narrowing used to be opt-in. Each domain hand-wrote a `decode_*_message` wrapper doing
+`expect_type(..)?` before its real decoder, and 24 of the 44 one-shot sites had no wrapper at
+all — they decoded whatever arrived on the channel. The follow-up that scoped #738 put that
+number at four; the real one was six times larger, which is the usual shape of a
+completeness claim nobody ran a command for.
+
+The consequence is quiet. A foreign proto handed to the wrong `prost` type usually decodes to
+*something* — field numbers overlap across messages — so the caller gets a plausible struct full
+of wrong values rather than an error.
+
+`expect_proto` is a combinator returning `impl Fn(&ResponseMessage)` rather than a fourth
+parameter on the three one-shot helpers, because those helpers already carry six arguments
+against a budget of three (see [param budget](../style/param-budget.md)) and none of them has a
+builder in front of it.
+
+The honest end state is a `trait ProtoPayload { const MESSAGE_ID; fn decode(&[u8]) }` implemented
+once per payload, which makes `expect_proto::<T>()` take no literals and retires both the roster
+and this node's standing cost. Tracked in
+[plans/claude-md-knowledge-graph.md](../../../plans/claude-md-knowledge-graph.md).
+
+## Precedents
+
+- #736 — collapsed seven `decode_*_message` dispatchers onto `ResponseMessage::expect_type`.
+  Right direction, but it left narrowing as a per-domain wrapper, so the sites without one
+  stayed unnarrowed and invisible.
+- #738 — moved the pair to the call site, deleted 30 wrapper functions, and swept the last
+  unnarrowed site (`next_valid_order_id`, on the shared `RequestIds` channel). The gate came
+  second, after a review mutated three sites and found only incidental round-trip tests failed;
+  the same mutation now fails the roster by name.

docs/rules/wire/proto-only-decoding.md

@@ -9,8 +9,8 @@ triggers:
   - a subscription terminates on an unexpected message
   - adding a decode arm for a new message type
 symbols: [require_proto, process_decode_result, StreamDecoder, RESPONSE_MESSAGE_IDS, Error::UnexpectedResponse, Error::UnexpectedWireFormat]
-related: [proto-aware-accessors, enum-typing, fixture-builders]
-precedents: ["#508", "#731", "#732", "#733", "#734", "#735"]
+related: [proto-aware-accessors, enum-typing, fixture-builders, one-shot-narrowing]
+precedents: ["#508", "#731", "#732", "#733", "#734", "#735", "#738"]
 memory: [project_protobuf_only, feedback_unreachable_regression_guards]
 ---
 
@@ -19,17 +19,25 @@ Every domain decoder reads its payload with `message.require_proto()` and feeds
 was retired with the floor ratchet.
 
 **`RESPONSE_MESSAGE_IDS` must list every type the `decode` match handles.** It is the skip
-filter: the sync and async subscription drivers drop anything not listed there *before* calling
-`decode`, because shared channels carry several types. A `decode` arm for an unlisted type is
-dead code. (The historical-tick driver in `market_data/historical/common/tick.rs` has its own
-single-valued `TickDecoder::MESSAGE_TYPE` doing the same job — a third mechanism, not yet
-unified.)
+filter: all three drivers drop anything not listed there *before* calling `decode`, because
+shared channels carry several types. A `decode` arm for an unlisted type is dead code. The third
+driver is `market_data/historical/common/tick.rs::classify`, over `TickDecoder`, which declares
+the same `RESPONSE_MESSAGE_IDS` const and filters with the same `is_undeclared` helper — a tick
+type simply declares one entry (#738). **The filter is shared; the gate below is not.**
+`collect_stream_decoder_impls` matches `impl StreamDecoder<` only, so the three `TickDecoder`
+consts are unchecked in both directions — over-declaring one routes a foreign frame's bytes into
+`proto::HistoricalTicks*`, and no test fails. Closing that is a follow-up in
+[plans/claude-md-knowledge-graph.md](../../../plans/claude-md-knowledge-graph.md).
 
 End every `impl StreamDecoder<T>::decode` match with `_ => Err(Error::unexpected_response(message))`
-anyway. It is now a backstop for the two lists disagreeing, not a control-flow signal — it
-terminates the subscription, loudly. Never `Error::NotImplemented` or `Error::Simple(...)`.
-
-**Both drift directions are gated**, by `test_response_message_ids_match_decode_arms`
+anyway. A decoder that consumes exactly one type has no match to hang that on, so
+`message.expect_type(IncomingMessages::X)?` is its backstop and is *not* redundant with the
+const — #738 removed two of them as duplication and the gate immediately read those decoders as
+claiming an arm for every message type. Either form is a backstop for the two lists disagreeing,
+not a control-flow signal — it terminates the subscription, loudly. Never
+`Error::NotImplemented` or `Error::Simple(...)`.
+
+**Both drift directions are gated for `StreamDecoder`**, by `test_response_message_ids_match_decode_arms`
 (`src/subscriptions/response_message_ids_tests.rs`). It probes every decoder with a minimal
 text-framed message of every `IncomingMessages` discriminant and requires the two lists to
 agree exactly: `UnexpectedResponse` means "no arm" (nothing else reaches the backstop),
@@ -109,3 +117,10 @@ other two:
 - #735 — same removal on the one-shot side, and `MessageBusStub` now classifies error frames
   so the input those arms handled can no longer be constructed. Surfaced a real bug behind one
   of them: blocking `matching_symbols` discarded routed errors and returned `Ok(vec![])`.
+- #738 — extended the const to `TickDecoder` so all three drivers share one skip filter, and
+  moved one-shot narrowing to `request_helpers::expect_proto` at the call site (gated separately
+  by `test_expect_proto_sites_match_the_roster`). **A counter-example too:** the same PR first
+  deleted the `expect_type` from the two single-type `StreamDecoder::decode` impls as a "third
+  copy of the same fact", and `test_response_message_ids_match_decode_arms` caught it within the
+  hour. The narrow is the single-arm form of the backstop, not a duplicate of the const. It also
+  showed the const's rename to `TickDecoder` bought the name without the gate.