refactor(subscriptions): drop dispatcher-intercepted RESPONSE_MESSAGE_IDS entries (#734)

Audit of every declared entry against the meaning #732 gave the const —
"the complete set of types that reach `decode`". Sixteen decoders declared
IncomingMessages::Error, which determine_routing classifies before the
allow-list: an error reaches a subscription as RoutedItem::Error/Notice and
never as the Response that decode consumes. The declarations and their
decode arms are removed.

routable_to_request_id_subscription loses its Error exemption, which existed
only so those declarations would not trip the routing guard — const declares,
guard exempts, circular. The circle is broken from the const side.

test_response_message_ids_match_decode_arms now diagnoses Error and Shutdown
specifically instead of reporting them as generic declared-but-unhandled.

The proof also covers the one-shot path: fold_one_shot runs its processor
only on Some(Ok(_)), and RoutedItem::into_legacy yields that only for
Response. wsh's two nested dispatchers are shared with a StreamDecoder and
lose their Error arm here; the remaining one-shot arms are a follow-up.

Drops 15 decoder-level test_decode_error_message* tests plus the two
historical streaming-error tests. All asserted a path only MessageBusStub can
produce — it delivers every fixture as RoutedItem::Response. The behaviour is
covered where it is implemented, by
test_hard_error_with_request_id_terminates_subscription and
test_subscription_hard_error_terminates_stream in both transports.

No production behaviour change: the removed arms were unreachable.

Closes the third follow-up in plans/claude-md-knowledge-graph.md.

Author: wboayue

docs/rules/wire/proto-aware-accessors.md

@@ -47,9 +47,16 @@ live-gateway smoke test surfaced the gap.
 `debug_assert_request_id_routable` (`src/subscriptions/common.rs`) runs in both subscription
 constructors and panics when a `request_id`-keyed subscription is built for a decoder whose
 `RESPONSE_MESSAGE_IDS` names a type the dispatcher cannot route to it. The classification is
-`first_unroutable_by_request_id` in `src/transport/routing.rs`, whose three accepting arms
-mirror `determine_routing`: `Error`, order-scoped types, and anything with a
-`text_request_id_field` entry.
+`first_unroutable_by_request_id` in `src/transport/routing.rs`, whose two accepting arms
+mirror `determine_routing`: order-scoped types, and anything with a `text_request_id_field`
+entry.
+
+`Error` was a third accepting arm until #734. It is not routable by `request_id` — it has no
+`text_request_id_field` entry — but sixteen decoders declared it, so the guard exempted it to
+keep them passing: const declares, guard exempts, circular. `determine_routing` classifies
+`Error` before the allow-list, so it reaches a subscription as `RoutedItem::Error`/`Notice`
+and never as a `Response` for `decode` to see. The declarations were the wrong half; they were
+removed and the exemption with them.
 
 Stub tests bypass the dispatcher but not the constructor, so this fires in exactly the tests
 that used to pass silently. It is compiled out of release builds — the invariant is over

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

@@ -10,7 +10,7 @@ triggers:
   - 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"]
+precedents: ["#508", "#731", "#732", "#733", "#734"]
 memory: [project_protobuf_only, feedback_unreachable_regression_guards]
 ---
 
@@ -41,6 +41,14 @@ Its companion `test_decoder_roster_is_complete` counts `impl StreamDecoder` bloc
 against the hand-listed roster, so adding a decoder without registering it fails rather than
 going unchecked.
 
+**Never declare `IncomingMessages::Error` or `Shutdown`, and never write a `decode` arm for
+them.** `determine_routing` classifies both before any allow-list, so an error reaches the
+subscription as `RoutedItem::Error`/`Notice` and a shutdown ends the dispatcher loop — neither
+ever arrives as the `RoutedItem::Response` that `decode` consumes. The same holds for one-shot
+requests, which read through `RoutedItem::into_legacy`: an error becomes `Some(Err(_))` and the
+processor never runs. Sixteen decoders declared `Error` anyway; the check above names the
+mistake specifically.
+
 ```rust
 pub(in crate::news) fn decode_news_bulletin(message: &ResponseMessage) -> Result<NewsBulletin, Error> {
     decode_news_bulletin_proto(message.require_proto()?)
@@ -94,3 +102,7 @@ other two:
 - #731 — split the framing failure out of the skip path so the fixture trap fails loudly.
 - #732 — retired skip-by-error-variant entirely; `RESPONSE_MESSAGE_IDS` is now the filter and
   the const lost its default so every impl must declare one.
+- #733 — gated both drift directions; the roster is counted against the tree.
+- #734 — audited all 78 declared entries under the const's new meaning. Sixteen declared
+  `IncomingMessages::Error`, which the dispatcher intercepts; those and their `decode` arms are
+  gone, along with the guard exemption that had been keeping them legal.

plans/claude-md-knowledge-graph.md

@@ -442,14 +442,52 @@ life.
   semantics has to be made twice. The rule node now says so out loud rather than claiming
   there are two drivers.
 
-- **Audit the 41 `RESPONSE_MESSAGE_IDS` entries against the const's new meaning.** Eight
-  decoders declare `IncomingMessages::Error`, which `determine_routing` classifies *before* the
-  allow-list — it arrives as `RoutedItem::Error`/`Notice` and can never reach `decode`.
-  Harmless, but #732 redefined the const as "the complete set of types that reach `decode`",
-  and under that contract declaring an unreachable type is exactly the kind of misleading
-  declaration the change set out to remove. `routable_to_request_id_subscription` already
-  special-cases `Error` to stop those declarations tripping the routing guard — const declares,
-  guard exempts, circular.
+- ~~**Audit the 41 `RESPONSE_MESSAGE_IDS` entries against the const's new meaning.**~~
+  **Shipped.** Both numbers in the original bullet were wrong, and the audit is what caught
+  them: there are **78** declared entries, not 41, and **16** decoders declared
+  `IncomingMessages::Error`, not eight. Counting them took one throwaway test over the roster
+  #733 had already built — which is the argument for building the roster, and one more instance
+  of the count class this file keeps failing.
+
+  The finding itself held. `determine_routing` classifies `Error` before the allow-list, so it
+  arrives as `RoutedItem::Error`/`Notice` and never as the `Response` that `decode` consumes.
+  All 16 declarations and their `decode` arms are gone, and `routable_to_request_id_subscription`
+  lost the `Error` exemption that had been keeping them legal — the circle is broken from the
+  const side, not the guard side. `test_response_message_ids_match_decode_arms` now names the
+  mistake specifically for `Error` and `Shutdown` rather than reporting it as a generic
+  declared-but-unhandled failure.
+
+  **The proof generalises further than the const.** The one-shot request path reads through
+  `RoutedItem::into_legacy`, which maps an error to `Some(Err(_))` and never runs the
+  processor — so every `IncomingMessages::Error` arm in a one-shot `decode_*_message` is dead
+  too. Two of those (`wsh::decode_metadata_message`, `decode_event_data_message`) are shared
+  with a `StreamDecoder` and were removed here; the rest are a separate follow-up below.
+
+  Removing the arms cost 15 decoder-level `test_decode_error_message*` tests, all asserting a
+  path only `MessageBusStub` can produce. The behaviour they claimed to cover is tested at both
+  layers that actually implement it — `test_hard_error_with_request_id_terminates_subscription`
+  and `test_subscription_hard_error_terminates_stream`, in each transport's tests.
+
+- **Retire the remaining one-shot `IncomingMessages::Error` arms.** The audit above proved the
+  whole class dead, not just the `StreamDecoder` half: `fold_one_shot` only ever calls its
+  processor on `Some(Ok(_))`, and `RoutedItem::into_legacy` yields that only for
+  `RoutedItem::Response`. So the `Error` arms in `accounts::decode_soft_dollar_tiers_message` /
+  `decode_user_info_message` / `decode_replace_fa_end_message`,
+  `contracts::decode_smart_components_message`, `scanner::decode_scanner_message`, and both
+  `config::common::decoders` dispatchers can never fire, as can the inline
+  `Ok(message) if message.message_type() == IncomingMessages::Error` guards in
+  `contracts::{sync,async}` and `market_data::historical::{sync,async}`. Mechanical, but each
+  deletion needs its test checked — the same 15-test cost pattern as #734. Left out of #734 to
+  keep that PR's boundary at the surface the follow-up named.
+
+- **Make `MessageBusStub` classify like the dispatcher.** It hands every fixture to the
+  subscription as `RoutedItem::Response`, including `Error` frames, which the real dispatcher
+  never does — that mismatch is why 15 decoder-level error tests existed and passed. This is
+  the same structural blind spot #730 recorded (stub tests inject below `determine_routing`),
+  seen from the fixture side rather than the routing side. Running fixtures through
+  `determine_routing`/`classify_error` in `mock_request` would close it; the risk is the tail of
+  stub tests using warning or data-advisory codes, which would become `RoutedItem::Notice` and
+  be skipped rather than surfacing as `Err`.
 
 - **Cache the message discriminant on `ResponseMessage`.** `message_type()` re-parses
   `fields[0]` with `i32::from_str` on every call, and it is called 4–6 times per inbound

src/accounts/common/stream_decoders/mod.rs

@@ -12,17 +12,12 @@ use super::{decoders, encoders};
 use crate::common::error_helpers;
 
 impl StreamDecoder<AccountSummaryResult> for AccountSummaryResult {
-    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[
-        IncomingMessages::AccountSummary,
-        IncomingMessages::AccountSummaryEnd,
-        IncomingMessages::Error,
-    ];
+    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::AccountSummary, IncomingMessages::AccountSummaryEnd];
 
     fn decode(_context: &DecoderContext, message: &mut ResponseMessage) -> Result<Self, Error> {
         match message.message_type() {
             IncomingMessages::AccountSummary => Ok(AccountSummaryResult::Summary(decoders::decode_account_summary(message)?)),
             IncomingMessages::AccountSummaryEnd => Ok(AccountSummaryResult::End),
-            IncomingMessages::Error => Err(Error::from(message.clone())),
             _ => Err(Error::unexpected_response(message)),
         }
     }
@@ -34,12 +29,11 @@ impl StreamDecoder<AccountSummaryResult> for AccountSummaryResult {
 }
 
 impl StreamDecoder<PnL> for PnL {
-    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::PnL, IncomingMessages::Error];
+    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::PnL];
 
     fn decode(_context: &DecoderContext, message: &mut ResponseMessage) -> Result<Self, Error> {
         match message.message_type() {
             IncomingMessages::PnL => decoders::decode_pnl(message),
-            IncomingMessages::Error => Err(Error::from(message.clone())),
             _ => Err(Error::unexpected_response(message)),
         }
     }
@@ -51,12 +45,11 @@ impl StreamDecoder<PnL> for PnL {
 }
 
 impl StreamDecoder<PnLSingle> for PnLSingle {
-    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::PnLSingle, IncomingMessages::Error];
+    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::PnLSingle];
 
     fn decode(_context: &DecoderContext, message: &mut ResponseMessage) -> Result<Self, Error> {
         match message.message_type() {
             IncomingMessages::PnLSingle => decoders::decode_pnl_single(message),
-            IncomingMessages::Error => Err(Error::from(message.clone())),
             _ => Err(Error::unexpected_response(message)),
         }
     }
@@ -68,13 +61,12 @@ impl StreamDecoder<PnLSingle> for PnLSingle {
 }
 
 impl StreamDecoder<PositionUpdate> for PositionUpdate {
-    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::Position, IncomingMessages::PositionEnd, IncomingMessages::Error];
+    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::Position, IncomingMessages::PositionEnd];
 
     fn decode(_context: &DecoderContext, message: &mut ResponseMessage) -> Result<Self, Error> {
         match message.message_type() {
             IncomingMessages::Position => Ok(PositionUpdate::Position(decoders::decode_position(message)?)),
             IncomingMessages::PositionEnd => Ok(PositionUpdate::PositionEnd),
-            IncomingMessages::Error => Err(Error::from(message.clone())),
             _ => Err(Error::unexpected_response(message)),
         }
     }
@@ -85,17 +77,12 @@ impl StreamDecoder<PositionUpdate> for PositionUpdate {
 }
 
 impl StreamDecoder<PositionUpdateMulti> for PositionUpdateMulti {
-    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[
-        IncomingMessages::PositionMulti,
-        IncomingMessages::PositionMultiEnd,
-        IncomingMessages::Error,
-    ];
+    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::PositionMulti, IncomingMessages::PositionMultiEnd];
 
     fn decode(_context: &DecoderContext, message: &mut ResponseMessage) -> Result<Self, Error> {
         match message.message_type() {
             IncomingMessages::PositionMulti => Ok(PositionUpdateMulti::Position(decoders::decode_position_multi(message)?)),
             IncomingMessages::PositionMultiEnd => Ok(PositionUpdateMulti::PositionEnd),
-            IncomingMessages::Error => Err(Error::from(message.clone())),
             _ => Err(Error::unexpected_response(message)),
         }
     }
@@ -112,7 +99,6 @@ impl StreamDecoder<AccountUpdate> for AccountUpdate {
         IncomingMessages::PortfolioValue,
         IncomingMessages::AccountUpdateTime,
         IncomingMessages::AccountDownloadEnd,
-        IncomingMessages::Error,
     ];
 
     fn decode(_context: &DecoderContext, message: &mut ResponseMessage) -> Result<Self, Error> {
@@ -121,7 +107,6 @@ impl StreamDecoder<AccountUpdate> for AccountUpdate {
             IncomingMessages::PortfolioValue => Ok(AccountUpdate::PortfolioValue(decoders::decode_account_portfolio_value(message)?)),
             IncomingMessages::AccountUpdateTime => Ok(AccountUpdate::UpdateTime(decoders::decode_account_update_time(message)?)),
             IncomingMessages::AccountDownloadEnd => Ok(AccountUpdate::End),
-            IncomingMessages::Error => Err(Error::from(message.clone())),
             _ => Err(Error::unexpected_response(message)),
         }
     }
@@ -132,17 +117,12 @@ impl StreamDecoder<AccountUpdate> for AccountUpdate {
 }
 
 impl StreamDecoder<AccountUpdateMulti> for AccountUpdateMulti {
-    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[
-        IncomingMessages::AccountUpdateMulti,
-        IncomingMessages::AccountUpdateMultiEnd,
-        IncomingMessages::Error,
-    ];
+    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::AccountUpdateMulti, IncomingMessages::AccountUpdateMultiEnd];
 
     fn decode(_context: &DecoderContext, message: &mut ResponseMessage) -> Result<Self, Error> {
         match message.message_type() {
             IncomingMessages::AccountUpdateMulti => Ok(AccountUpdateMulti::AccountMultiValue(decoders::decode_account_multi_value(message)?)),
             IncomingMessages::AccountUpdateMultiEnd => Ok(AccountUpdateMulti::End),
-            IncomingMessages::Error => Err(Error::from(message.clone())),
             _ => Err(Error::unexpected_response(message)),
         }
     }

src/accounts/common/stream_decoders/tests.rs

@@ -46,15 +46,6 @@ mod account_summary_tests {
         assert!(matches!(result, AccountSummaryResult::End));
     }
 
-    #[test]
-    fn test_decode_error_message() {
-        // Error on the same request_id channel surfaces as Error::Notice, not a
-        // parse failure or "unexpected message" error (#434).
-        let mut message = proto_error_response(123, 10089, "Requested market data is not subscribed");
-        let err = AccountSummaryResult::decode(&test_context(), &mut message).unwrap_err();
-        assert_tws_error_message(err, 10089, "not subscribed");
-    }
-
     #[test]
     fn test_cancel_message() {
         let bytes = AccountSummaryResult::cancel_message(TEST_SERVER_VERSION, Some(TEST_REQUEST_ID), None).unwrap();
@@ -73,11 +64,7 @@ mod account_summary_tests {
     fn test_response_message_ids() {
         assert_eq!(
             AccountSummaryResult::RESPONSE_MESSAGE_IDS,
-            &[
-                IncomingMessages::AccountSummary,
-                IncomingMessages::AccountSummaryEnd,
-                IncomingMessages::Error
-            ]
+            &[IncomingMessages::AccountSummary, IncomingMessages::AccountSummaryEnd]
         );
     }
 }
@@ -119,14 +106,7 @@ mod pnl_tests {
 
     #[test]
     fn test_response_message_ids() {
-        assert_eq!(PnL::RESPONSE_MESSAGE_IDS, &[IncomingMessages::PnL, IncomingMessages::Error]);
-    }
-
-    #[test]
-    fn test_decode_error_message() {
-        let mut message = proto_error_response(123, 10089, "Requested market data is not subscribed");
-        let err = PnL::decode(&test_context(), &mut message).unwrap_err();
-        assert_tws_error_message(err, 10089, "not subscribed");
+        assert_eq!(PnL::RESPONSE_MESSAGE_IDS, &[IncomingMessages::PnL]);
     }
 }
 
@@ -164,14 +144,7 @@ mod pnl_single_tests {
 
     #[test]
     fn test_response_message_ids() {
-        assert_eq!(PnLSingle::RESPONSE_MESSAGE_IDS, &[IncomingMessages::PnLSingle, IncomingMessages::Error]);
-    }
-
-    #[test]
-    fn test_decode_error_message() {
-        let mut message = proto_error_response(123, 10089, "Requested market data is not subscribed");
-        let err = PnLSingle::decode(&test_context(), &mut message).unwrap_err();
-        assert_tws_error_message(err, 10089, "not subscribed");
+        assert_eq!(PnLSingle::RESPONSE_MESSAGE_IDS, &[IncomingMessages::PnLSingle]);
     }
 }
 
@@ -226,16 +199,9 @@ mod position_update_tests {
     fn test_response_message_ids() {
         assert_eq!(
             PositionUpdate::RESPONSE_MESSAGE_IDS,
-            &[IncomingMessages::Position, IncomingMessages::PositionEnd, IncomingMessages::Error]
+            &[IncomingMessages::Position, IncomingMessages::PositionEnd]
         );
     }
-
-    #[test]
-    fn test_decode_error_message() {
-        let mut message = proto_error_response(123, 10089, "Requested market data is not subscribed");
-        let err = PositionUpdate::decode(&test_context(), &mut message).unwrap_err();
-        assert_tws_error_message(err, 10089, "not subscribed");
-    }
 }
 
 mod position_update_multi_tests {
@@ -296,20 +262,9 @@ mod position_update_multi_tests {
     fn test_response_message_ids() {
         assert_eq!(
             PositionUpdateMulti::RESPONSE_MESSAGE_IDS,
-            &[
-                IncomingMessages::PositionMulti,
-                IncomingMessages::PositionMultiEnd,
-                IncomingMessages::Error
-            ]
+            &[IncomingMessages::PositionMulti, IncomingMessages::PositionMultiEnd]
         );
     }
-
-    #[test]
-    fn test_decode_error_message() {
-        let mut message = proto_error_response(123, 10089, "Requested market data is not subscribed");
-        let err = PositionUpdateMulti::decode(&test_context(), &mut message).unwrap_err();
-        assert_tws_error_message(err, 10089, "not subscribed");
-    }
 }
 
 mod account_update_tests {
@@ -399,17 +354,9 @@ mod account_update_tests {
                 IncomingMessages::PortfolioValue,
                 IncomingMessages::AccountUpdateTime,
                 IncomingMessages::AccountDownloadEnd,
-                IncomingMessages::Error,
             ]
         );
     }
-
-    #[test]
-    fn test_decode_error_message() {
-        let mut message = proto_error_response(123, 10089, "Requested market data is not subscribed");
-        let err = AccountUpdate::decode(&test_context(), &mut message).unwrap_err();
-        assert_tws_error_message(err, 10089, "not subscribed");
-    }
 }
 
 mod account_update_multi_tests {
@@ -468,20 +415,9 @@ mod account_update_multi_tests {
     fn test_response_message_ids() {
         assert_eq!(
             AccountUpdateMulti::RESPONSE_MESSAGE_IDS,
-            &[
-                IncomingMessages::AccountUpdateMulti,
-                IncomingMessages::AccountUpdateMultiEnd,
-                IncomingMessages::Error
-            ]
+            &[IncomingMessages::AccountUpdateMulti, IncomingMessages::AccountUpdateMultiEnd]
         );
     }
-
-    #[test]
-    fn test_decode_error_message() {
-        let mut message = proto_error_response(123, 10089, "Requested market data is not subscribed");
-        let err = AccountUpdateMulti::decode(&test_context(), &mut message).unwrap_err();
-        assert_tws_error_message(err, 10089, "not subscribed");
-    }
 }
 
 // Edge case tests