fix(contracts): propagate routed errors from blocking matching_symbols (#735)

* fix(contracts): propagate routed errors from blocking matching_symbols

Retires the one-shot IncomingMessages::Error arms, the other half of the
class #734 proved dead. fold_one_shot runs its processor only on Some(Ok(_)),
and RoutedItem::into_legacy yields that only for RoutedItem::Response, so an
error frame reaches no decoder. Thirteen sites go: seven decode_*_message
dispatchers (accounts x3, contracts, scanner, config x2) and six inline
message_type() == Error guards in contracts::{sync,async} and
market_data::historical::{sync,async}. connection/common.rs keeps its arm —
it reads frames during the handshake, before a dispatcher exists.

One dead arm was hiding a live bug. Blocking matching_symbols read its
response with `if let Some(Ok(mut message))`, so a routed error fell through
to Ok(Vec::new()) — a rejected pattern was indistinguishable from "no symbols
matched". The async twin already returned Err(e). Both now match.

MessageBusStub classifies error frames like the dispatcher. Removing the arms
was about to cost a third batch of tests deleted for the same reason (#734
dropped 17), so the fixture was fixed instead of the tests: routed_items()
runs each fixture through determine_routing/classify_error, so an error frame
arrives as RoutedItem::Error/Notice as it would on the wire. Both tests that
would have been deleted then passed unmodified, and the sync matching_symbols
test became a real regression guard — reverting the fix reproduces Ok(vec![]).

Renames two config tests that claimed the removed behaviour in their names
and asserted only is_err(), which would have passed through the change
silently; they now pin UnexpectedResponse.

* fix(orders): surface rejected what-if orders from OrderBuilder::analyze

Restores the two historical streaming-error tests deleted in #734. They pass
unmodified against the classifying stub, which shows the #734 reasoning was
wrong in principle: "covered at the transport layer" conflates delivery with
consumption. The transport tests prove the dispatcher produces
RoutedItem::Error; they say nothing about whether a given public API hands it
to the caller.

Sweeping every Some(Ok(..)) consumption site for that shape found one more
API that did not. OrderBuilder::analyze dropped routed errors on both sides —
`if let Ok(..)` inside the loop on sync, `while let Some(Ok(..))` on async —
returning UnexpectedEndOfStream instead of the rejection, on the one API
where rejection is a routine outcome (code 201). Every other site handles
Some(Err(e)).

The four existing analyze tests missed it because they exercise a
re-implementation: the builder test modules define their own analyze on
OrderBuilder<'a, MockOrderClient> returning Vec<PlaceOrder>, so the
production methods on Client had no coverage at all. Added two tests at the
real seam; reverting either fix reproduces UnexpectedEndOfStream. The rest of
that mock surface is recorded as a follow-up.

* refactor: /simplify pass on the routed-error changes

Stub classification now covers the intercepted *set*, not just Error:
`_ => Response(message)` reproduced `RoutingDecision::Error` by name, so
Shutdown still reached decoders as a Response. Both types the dispatcher
intercepts are handled, matching DISPATCHER_INTERCEPTED, and the fixture
conversion goes through the existing From<ResponseMessage> impl.

Dead code the arm removals left behind: From<ResponseMessage> for Error and
From<&ResponseMessage> for Error had zero callers once the decoders stopped
using them; deleted with the two tests that only exercised the impls
themselves. Config decoder doc still described its deleted Error arm.

Both analyze loops collapse to `?` — the three-arm match with `Ok(_) => {}`
and `Err(e) => return Err(e)` is what `?` does, and the sync version was a
145-column line. The async mock shadow carried the same discard this PR
fixed, so it was corrected in place.

Tests now use existing helpers instead of hand-rolling: the two orders and
two historical tests build their client via
create_{,blocking_}test_client_with_ordered_proto_responses, and the
historical pair asserts with assert_tws_error_message rather than
to_string().contains(), which pins the error code too. Dropped their dead
time_zone setup (no bars decoded) and unified the sync/async shapes.

Three identical async stub bodies extracted to seeded_subscription, and
routed_items gained a direct test — the Shutdown arm had none.

Deferred as restructuring: fold_one_shot adoption at ~14 hand-rolled sites,
an expect_type helper for the seven identical dispatchers, and deleting the
order-builder mock shadows. All recorded in the plan.

Author: wboayue

CHANGELOG.md

@@ -23,6 +23,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- `OrderBuilder::analyze()` (what-if orders, blocking and async) now returns the TWS rejection instead of `Error::UnexpectedEndOfStream`. A rejected what-if order arrives as a routed error, which the response read discarded — the blocking path via `if let Ok(..)` inside the loop, the async path by ending its `while let Some(Ok(..))` loop — so the caller lost the reason (e.g. code 201, `Order rejected - reason:...`) and got a generic end-of-stream error. Rejection is a routine outcome for a what-if order, so this was the likeliest path to hit it (#735).
+
+- `matching_symbols()` on the blocking client now returns the TWS error instead of an empty list. A routed error arrives as `Some(Err(_))`, which the `if let Some(Ok(_))` read discarded, so a rejected pattern silently returned `Ok(vec![])` — indistinguishable from "no symbols matched". The async client already propagated it (#735).
+
 - `TickTypes::MarketDataType` now reaches `Client::market_data` subscriptions. The message type was missing from the request-id routing allow-list, so TWS's market-data-type notifications (real-time / frozen / delayed / delayed-frozen, sent on subscribe and whenever the feed switches) were routed to a shared channel nobody subscribes to and dropped. The decoder has produced the variant since #516; nothing could ever yield it (#730).
 
 - Decimal-typed wire fields no longer fall back to `0` when the value fails to parse; a malformed value now surfaces as `Error::Parse` and fails the request or subscription instead of being silently swallowed. Covers order quantities, execution shares, positions, contract-detail sizes, bar volume/WAP, tick and market-depth sizes (#716).

docs/rules/testing/fixture-builders.md

@@ -51,6 +51,23 @@ and refactored them into named builders in `testdata/builders/market_data.rs`.
 A builder needs a **current test consumer**. Matching the encoder file one-to-one "for
 completeness" is not a consumer.
 
+## What the stub does and does not simulate
+
+`MessageBusStub` sits below the dispatcher, so a fixture reaches the subscription without
+being routed. Two consequences pull in opposite directions:
+
+- **It does classify `Error` frames** (since #735). `routed_items()` runs each fixture through
+  `determine_routing`/`classify_error`, so an error frame arrives as
+  `RoutedItem::Error`/`Notice` exactly as it would on the wire. Before that it arrived as a
+  `RoutedItem::Response`, which no real transport produces — that gap is what let decoders grow
+  unreachable `IncomingMessages::Error` arms with passing tests to match. A warning or
+  data-advisory code now becomes a `Notice` and is filtered by `iter_data()`/`filter_data`;
+  assert on `SubscriptionItem::Notice` if that is the point of the test.
+- **It does not route by channel.** The stub has one channel per request, so a fixture reaches
+  the subscription regardless of whether `determine_routing` could have addressed it there.
+  `debug_assert_request_id_routable` covers that half; see
+  [proto-aware accessors](../wire/proto-aware-accessors.md).
+
 ## Note on `server_version`
 
 The version passed to `Client::stubbed` gates *outbound encoder* feature checks, not the
@@ -63,6 +80,7 @@ constant like `SIZE_RULES` in a stub test is correct, not a leftover.
 - #534 — field-minimal builders for deeply-nested protos.
 - #543 — /simplify caught fixture helpers misplaced under `<domain>/common/`.
 - #731 — made a mis-framed fixture fail its test instead of silently skipping.
+- #735 — taught the stub to classify `Error` frames like the dispatcher.
 
 See also [docs/testing-patterns.md](../../testing-patterns.md) for choosing between
 `MessageBusStub`, `MemoryStream`, and `spawn_handshake_listener`.

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", "#733", "#734"]
+precedents: ["#508", "#731", "#732", "#733", "#734", "#735"]
 memory: [project_protobuf_only, feedback_unreachable_regression_guards]
 ---
 
@@ -106,3 +106,6 @@ other two:
 - #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.
+- #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![])`.

plans/claude-md-knowledge-graph.md

@@ -468,26 +468,74 @@ life.
   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`.
+- ~~**Retire the remaining one-shot `IncomingMessages::Error` arms.**~~ ~~**Make
+  `MessageBusStub` classify like the dispatcher.**~~ **Both shipped in #735, and they turned
+  out to be one job.** Thirteen dead sites went: seven `decode_*_message` dispatchers
+  (`accounts` ×3, `contracts`, `scanner`, `config` ×2) and six inline
+  `message.message_type() == IncomingMessages::Error` guards in `contracts::{sync,async}` and
+  `market_data::historical::{sync,async}`. Kept: `connection/common.rs`, which reads frames
+  during the handshake before a dispatcher exists, and `messages/parser_registry.rs`, which is
+  a trace-parser table.
+
+  **Deleting the arms was going to cost a third batch of tests, and that was the signal to stop
+  deleting.** #734 dropped 17 tests because only `MessageBusStub` could produce their input;
+  two more were about to go the same way. The rule of three tripped, so the stub was fixed
+  instead: `routed_items()` runs every fixture through `determine_routing`/`classify_error`, so
+  an error frame arrives as `RoutedItem::Error`/`Notice` exactly as on the wire. Both tests then
+  passed unmodified. **Fixing the fixture kept the coverage that deleting the arms would have
+  destroyed** — and the two tests #734 deleted on the same grounds would have survived it too.
+
+  The feared tail — warning and data-advisory codes becoming `Notice` and being filtered —
+  never materialised: one test needed touching across both legs, and only because its assertion
+  encoded the old wart.
+
+  **One dead arm was hiding a live bug.** Blocking `matching_symbols` read its response with
+  `if let Some(Ok(mut message))`, so a routed error fell through to `Ok(Vec::new())` — a
+  rejected pattern was indistinguishable from "no symbols matched". Its async twin already had
+  `Some(Err(e)) => return Err(e)`. The dead `IncomingMessages::Error` arm sitting next to the
+  hole is what makes it legible: someone meant to handle errors and wired it one layer too low,
+  and the arm's unreachability is exactly why the omission below it stayed invisible.
+  **A dead arm is worth reading before deleting — it marks where someone expected a case to
+  arrive, which is where to check whether the real case is handled at all.**
+
+  **The #734 deletions were wrong in principle, not just in outcome.** The justification was
+  "the mechanism is covered at the transport layer", which conflates *delivery* with
+  *consumption*. The transport tests prove the dispatcher produces `RoutedItem::Error`; they say
+  nothing about whether a given public API hands it to the caller — and two APIs did not. Both
+  deleted tests were restored and passed unmodified against the classifying stub.
+
+  A sweep of every `Some(Ok(..))` consumption site for the same shape found one more:
+  `OrderBuilder::analyze()` dropped routed errors on both sides (`if let Ok(..)` inside the
+  loop on sync, `while let Some(Ok(..))` on async), returning `UnexpectedEndOfStream` instead of
+  the rejection — on the one API where rejection is a routine outcome. Every other site handles
+  `Some(Err(e))`. **The per-API question "does this consume `Err`?" needs asking once per
+  public entry point; no shared layer answers it.**
+
+- **Adopt `fold_one_shot` at the ~14 sites that hand-roll it.** `Some(Ok(m)) => decode`,
+  `Some(Err(e)) => Err(e)`, `None => default` is spelled out at four sites each in
+  `contracts::{sync,async}`, plus `news::{sync,async}` ×2 each and `scanner::{sync,async}` —
+  while `fold_one_shot` itself has no callers outside `request_helpers.rs`. Two blockers, both
+  vestigial: `decode_contract_descriptions` and `decode_market_rule` take `&mut ResponseMessage`
+  and an unused `server_version`, though their bodies are only `require_proto()?` (a `&self`
+  method). The option-computation sites are genuinely blocked — they need `&mut` plus a
+  `DecoderContext`. `request_helpers` also lacks a request-id/no-retry variant, which is exactly
+  the slot `matching_symbols` falls into. Deferred from #735's `/simplify` as restructuring.
+
+- **Collapse the seven identical `decode_*_message` dispatchers.** After #734/#735 they are all
+  `match message.message_type() { X => decode_x(m), _ => Err(unexpected_response(m)) }` across
+  `accounts` ×3, `contracts`, `config` ×2, `scanner`. An `expect_type(message, expected)` helper
+  makes each a one-liner. Rule of three is well past, but it touches seven modules.
+
+- **The order-builder tests re-implement the code they test.** `src/orders/builder/{sync,async}_impl/tests.rs`
+  define their own `analyze` / `submit` on `OrderBuilder<'a, MockOrderClient>` returning
+  `Vec<PlaceOrder>` rather than a `Subscription`, so the production methods on `Client` had zero
+  coverage — which is why the `analyze` bug above survived four tests named for it. #735 added
+  two tests at the real seam for `analyze` only; `submit`, `build_order`, and the bracket-order
+  builders still have none. The async shadow carried the very discard the PR fixed
+  (`while let Some(Ok(..))`) and was corrected in place, which is the argument for deleting the
+  shadows rather than maintaining two copies. A textbook
+  [exercise production code](../docs/rules/testing/exercise-production-code.md) violation, and
+  the largest one left in the tree.
 
 - **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/decoders/mod.rs

@@ -209,7 +209,6 @@ pub(crate) fn decode_soft_dollar_tiers_proto(bytes: &[u8]) -> Result<Vec<SoftDol
 pub(in crate::accounts) fn decode_soft_dollar_tiers_message(message: &ResponseMessage) -> Result<Vec<SoftDollarTier>, Error> {
     match message.message_type() {
         IncomingMessages::SoftDollarTier => decode_soft_dollar_tiers(message),
-        IncomingMessages::Error => Err(Error::from(message)),
         _ => Err(Error::unexpected_response(message)),
     }
 }
@@ -228,7 +227,6 @@ pub(crate) fn decode_user_info_proto(bytes: &[u8]) -> Result<UserInfo, Error> {
 pub(in crate::accounts) fn decode_user_info_message(message: &ResponseMessage) -> Result<UserInfo, Error> {
     match message.message_type() {
         IncomingMessages::UserInfo => decode_user_info(message),
-        IncomingMessages::Error => Err(Error::from(message)),
         _ => Err(Error::unexpected_response(message)),
     }
 }
@@ -263,7 +261,6 @@ pub(crate) fn decode_replace_fa_end_proto(bytes: &[u8]) -> Result<ReplaceFaResul
 pub(in crate::accounts) fn decode_replace_fa_end_message(message: &ResponseMessage) -> Result<ReplaceFaResult, Error> {
     match message.message_type() {
         IncomingMessages::ReplaceFAEnd => decode_replace_fa_end(message),
-        IncomingMessages::Error => Err(Error::from(message)),
         _ => Err(Error::unexpected_response(message)),
     }
 }

src/common/request_helpers.rs

@@ -6,6 +6,13 @@
 /// fanned out to one-shot shared channels — instead of masking it as a
 /// default value (#694). `on_none` decides what a closed stream means for
 /// the caller (a default value, or `Error::UnexpectedEndOfStream`).
+///
+/// `processor` therefore never sees an `IncomingMessages::Error` frame. The
+/// dispatcher classifies those into `RoutedItem::Error`/`Notice`, and
+/// `RoutedItem::into_legacy` turns them into the `Some(Err)` arm above — so a
+/// `decode_*_message` dispatcher that matches on `IncomingMessages::Error` is
+/// writing an arm that cannot fire. See
+/// `docs/rules/wire/proto-only-decoding.md`.
 pub(crate) fn fold_one_shot<R>(
     response: Option<Result<crate::messages::ResponseMessage, crate::Error>>,
     processor: impl FnOnce(&crate::messages::ResponseMessage) -> Result<R, crate::Error>,

src/config/common/decoders.rs

@@ -11,13 +11,13 @@ use crate::messages::{IncomingMessages, ResponseMessage};
 use crate::proto;
 use crate::Error;
 
-/// Dispatch on the incoming message type and forward to the typed decoder.
-/// Routes `Error` frames into `Error::from` and any other variant into
-/// `Error::UnexpectedResponse`.
+/// Dispatch on the incoming message type and forward to the typed decoder. Any
+/// other variant becomes `Error::UnexpectedResponse`. There is deliberately no
+/// `IncomingMessages::Error` arm — the dispatcher classifies error frames, so
+/// one never reaches a decoder; see `docs/rules/wire/proto-only-decoding.md`.
 pub(in crate::config) fn decode_config_message(message: &ResponseMessage) -> Result<Config, Error> {
     match message.message_type() {
         IncomingMessages::ConfigResponse => decode_config_proto(message.require_proto()?),
-        IncomingMessages::Error => Err(Error::from(message)),
         _ => Err(Error::unexpected_response(message)),
     }
 }
@@ -133,7 +133,6 @@ fn convert_smart_routing(p: proto::OrdersSmartRoutingConfig) -> OrdersSmartRouti
 pub(in crate::config) fn decode_update_config_message(message: &ResponseMessage) -> Result<UpdateConfigResponse, Error> {
     match message.message_type() {
         IncomingMessages::UpdateConfigResponse => decode_update_config_proto(message.require_proto()?),
-        IncomingMessages::Error => Err(Error::from(message)),
         _ => Err(Error::unexpected_response(message)),
     }
 }

src/config/common/decoders_tests.rs

@@ -60,10 +60,15 @@ fn test_decode_config_message_dispatches_config_response() {
 }
 
 #[test]
-fn test_decode_config_message_routes_error() {
-    // IncomingMessages::Error == 4; a text-framed error surfaces as an Err.
+fn test_decode_config_message_rejects_error_frames() {
+    // IncomingMessages::Error == 4. Regression guard for the arm removed in
+    // #735: the dispatcher classifies error frames, so one never reaches a
+    // decoder and the `_` backstop must treat it like any other foreign type.
     let message = ResponseMessage::from("4\09000\0322\0error text\0");
-    assert!(decode_config_message(&message).is_err());
+    match decode_config_message(&message) {
+        Err(Error::UnexpectedResponse(_)) => {}
+        other => panic!("expected UnexpectedResponse, got {other:?}"),
+    }
 }
 
 #[test]
@@ -109,9 +114,13 @@ fn test_decode_update_config_message_populated() {
 }
 
 #[test]
-fn test_decode_update_config_message_routes_error() {
+fn test_decode_update_config_message_rejects_error_frames() {
+    // Mirrors test_decode_config_message_rejects_error_frames.
     let message = ResponseMessage::from("4\09000\0322\0error text\0");
-    assert!(decode_update_config_message(&message).is_err());
+    match decode_update_config_message(&message) {
+        Err(Error::UnexpectedResponse(_)) => {}
+        other => panic!("expected UnexpectedResponse, got {other:?}"),
+    }
 }
 
 #[test]

src/contracts/async.rs

@@ -8,7 +8,7 @@ use crate::messages::{IncomingMessages, OutgoingMessages};
 use crate::protocol::{check_version, Features};
 use crate::subscriptions::{StreamDecoder, Subscription};
 use crate::{Client, Error};
-use log::{error, info};
+use log::info;
 
 impl Client {
     /// Requests contract information.
@@ -57,7 +57,6 @@ impl Client {
                             contract_details.push(decoded);
                         }
                         IncomingMessages::ContractDataEnd => return Ok(contract_details),
-                        IncomingMessages::Error => return Err(Error::from(response)),
                         _ => return Err(Error::unexpected_response(&response)),
                     }
                 }
@@ -99,23 +98,15 @@ impl Client {
 
         match subscription.next().await {
             Some(Ok(mut message)) => match message.message_type() {
-                IncomingMessages::SymbolSamples => {
-                    return decoders::decode_contract_descriptions(self.server_version(), &mut message);
-                }
-                IncomingMessages::Error => {
-                    error!("unexpected error: {message:?}");
-                    return Err(Error::unexpected_response(&message));
-                }
+                IncomingMessages::SymbolSamples => decoders::decode_contract_descriptions(self.server_version(), &mut message),
                 _ => {
                     info!("unexpected message: {message:?}");
-                    return Err(Error::unexpected_response(&message));
+                    Err(Error::unexpected_response(&message))
                 }
             },
-            Some(Err(e)) => return Err(e),
-            None => {}
+            Some(Err(e)) => Err(e),
+            None => Ok(Vec::new()),
         }
-
-        Ok(Vec::default())
     }
 
     /// Requests details about a given market rule.

src/contracts/async_tests.rs

@@ -575,13 +575,17 @@ async fn contract_details_propagates_verify_failure() {
 
 #[tokio::test]
 async fn matching_symbols_returns_server_error() {
-    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![text_response(
-        "4|2|9000|321|invalid pattern|",
+    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_error_response(
+        9000,
+        321,
+        "invalid pattern",
     )]));
     let client = Client::stubbed(message_bus, server_versions::BOND_ISSUERID);
 
+    // The TWS error itself, not a generic UnexpectedResponse: the dispatcher
+    // classifies error frames, so `matching_symbols` receives `Some(Err(_))`.
     let err = client.matching_symbols("???").await.unwrap_err();
-    assert!(matches!(err, crate::Error::UnexpectedResponse(_)), "got {err:?}");
+    assert_tws_error_message(err, 321, "invalid pattern");
 }
 
 #[tokio::test]