ibapi 5.4.0

use super::*;
use crate::client::blocking::Client;
use crate::client::ids::RequestId;
use crate::common::test_utils::helpers::{
    assert_decimal_parse_error, assert_proto_msg_id, assert_request, assert_request_msg_id, assert_tws_error_message, count_proto_msgs,
    create_blocking_test_client_with_ordered_proto_responses, proto_error_response, proto_response, request_message_count, test_notice,
    TEST_REQ_ID_FIRST,
};
use crate::contracts::Contract;
use crate::market_data::historical::BarTimestamp;
use crate::market_data::historical::{TickBidAsk, TickLast, TickMidpoint, ToDuration};
use crate::market_data::{IgnoreSize, TradingHours};
use crate::messages::{IncomingMessages, OutgoingMessages};
use crate::protocol::{Features, ProtocolFeature};
use crate::server_versions;
use crate::stubs::MessageBusStub;
use crate::testdata::builders::market_data::{
    head_timestamp_request, head_timestamp_response, histogram_data_request, histogram_data_response, histogram_entry, historical_data_bar,
    historical_data_daily_bar, historical_data_end_response, historical_data_request, historical_data_response, historical_data_update_response,
    historical_schedule_response, historical_tick_bid_ask, historical_tick_last, historical_tick_mid, historical_ticks_bid_ask_response,
    historical_ticks_last_response, historical_ticks_request, historical_ticks_response,
};
use crate::testdata::builders::ResponseProtoEncoder;
use crate::transport::common::Lease;
use crate::transport::{Signal, SubscriptionBuilder};
use crossbeam::channel;
use std::sync::Arc;
use time::macros::{date, datetime};
use time::OffsetDateTime;
use time_tz::{self, PrimitiveDateTimeExt, Tz};

// Build a `TickSubscription<T>` fed by `items`, with the inbound channel closed
// after the last item. Returns the signal-channel receiver too; keep it bound
// for the test's lifetime to silence the drop-time signal warning.
fn tick_sub_from_routed<T: TickDecoder<T>>(items: Vec<RoutedItem>) -> (TickSubscription<T>, channel::Receiver<Signal>) {
    let (sender, receiver) = channel::unbounded();
    let (signaler, signal_rx) = channel::unbounded();
    for item in items {
        sender.send(item).unwrap();
    }
    drop(sender);
    let internal = SubscriptionBuilder::new()
        .receiver(receiver)
        .signaler(signaler)
        .lease(Lease::new())
        .request_id(RequestId::nth(300))
        .build();
    let subscription = TickSubscription::new(internal, RequestId::nth(300).raw(), Arc::new(MessageBusStub::default()));
    (subscription, signal_rx)
}

// Pins server_version one below `feature.min_version` and asserts the call fails
// with the feature name in the error. Custom helper (vs. `.expect_err`) because
// subscription return types don't implement Debug.
fn assert_version_check_fails<F, T>(feature: ProtocolFeature, call: F)
where
    F: FnOnce(Client) -> Result<T, Error>,
{
    let message_bus = Arc::new(MessageBusStub::default());
    let client = Client::stubbed(message_bus, feature.min_version - 1);
    let Err(err) = call(client) else {
        panic!("expected version-check failure ({})", feature.name);
    };
    assert!(err.to_string().contains(feature.name), "expected '{}', got: {err}", feature.name);
}

// Feeds a single frame of a message type the request doesn't expect: it is
// skipped, the stream closes before the answer, and the request is cancelled
// at TWS with `cancel`.
fn assert_skips_undeclared<F, T>(server_version: i32, response: crate::messages::ResponseMessage, cancel: OutgoingMessages, call: F)
where
    F: FnOnce(Client) -> Result<T, Error>,
{
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![response]));
    let client = Client::stubbed(message_bus.clone(), server_version);
    let Err(err) = call(client) else {
        panic!("expected UnexpectedEndOfStream");
    };
    assert!(
        matches!(err, Error::UnexpectedEndOfStream),
        "expected UnexpectedEndOfStream, got: {err:?}"
    );
    let messages = message_bus.request_messages.read().unwrap();
    assert_eq!(count_proto_msgs(&messages, cancel), 1, "no answer, so the request is cancelled");
}

#[test]
fn head_timestamp() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HeadTimestamp,
        head_timestamp_response().unix_timestamp(1678323335).encode_proto(),
    )]));

    let client = Client::stubbed(message_bus.clone(), server_versions::PROTOBUF_REST_MESSAGES_3);

    let contract = Contract::stock("MSFT").build();
    let what_to_show = WhatToShow::Trades;
    let trading_hours = TradingHours::Regular;

    let head_timestamp = client
        .head_timestamp(&contract, what_to_show, trading_hours)
        .expect("head timestamp request failed");

    assert_eq!(head_timestamp, OffsetDateTime::from_unix_timestamp(1678323335).unwrap(), "bar.date");

    assert_eq!(request_message_count(&message_bus), 1);
    assert_request(
        &message_bus,
        0,
        &head_timestamp_request()
            .request_id(TEST_REQ_ID_FIRST)
            .contract(&contract)
            .what_to_show(what_to_show)
            .use_rth(trading_hours.use_rth()),
    );
}

#[test]
fn histogram_data_malformed_size_fails_the_request() {
    // End-to-end through the public API: a malformed size must fail the request
    // rather than silently decoding as 0 (issue #716).
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistogramData,
        histogram_data_response()
            .entry(histogram_entry(125.50, 0.0).size_wire(Some("abc")))
            .encode_proto(),
    )]));

    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);
    let contract = Contract::stock("MSFT").build();

    assert_decimal_parse_error(client.histogram_data(&contract, TradingHours::Regular, BarSize::Day), "abc");
}

#[test]
fn histogram_data() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistogramData,
        histogram_data_response()
            .entry(histogram_entry(125.50, 1000.0))
            .entry(histogram_entry(126.00, 2000.0))
            .entry(histogram_entry(126.50, 3000.0))
            .encode_proto(),
    )]));

    let client = Client::stubbed(message_bus.clone(), server_versions::PROTOBUF_REST_MESSAGES_3);

    let contract = Contract::stock("MSFT").build();
    let trading_hours = TradingHours::Regular;
    let period = BarSize::Day;

    let histogram_data = client
        .histogram_data(&contract, trading_hours, period)
        .expect("histogram data request failed");

    // Assert Response
    assert_eq!(histogram_data.len(), 3, "histogram_data.len()");

    assert_eq!(histogram_data[0].price, 125.50, "histogram_data[0].price");
    assert_eq!(histogram_data[0].size, Some(1000.0), "histogram_data[0].size");

    assert_eq!(histogram_data[1].price, 126.00, "histogram_data[1].price");
    assert_eq!(histogram_data[1].size, Some(2000.0), "histogram_data[1].size");

    assert_eq!(histogram_data[2].price, 126.50, "histogram_data[2].price");
    assert_eq!(histogram_data[2].size, Some(3000.0), "histogram_data[2].size");

    // Assert Request
    assert_eq!(request_message_count(&message_bus), 1);
    assert_request(
        &message_bus,
        0,
        &histogram_data_request()
            .request_id(TEST_REQ_ID_FIRST)
            .contract(&contract)
            .use_rth(trading_hours.use_rth())
            .period(period),
    );
}

#[test]
fn historical_data() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![
        proto_response(
            IncomingMessages::HistoricalData,
            historical_data_response()
                .bar(
                    historical_data_bar(1_681_344_000) // 2023-04-13 UTC
                        .ohlc(182.94, 186.50, 180.94, 185.90)
                        .volume(948_837.22)
                        .wap(184.869)
                        .count(324_891),
                )
                .bar(
                    historical_data_daily_bar("20230414")
                        .ohlc(183.88, 186.28, 182.01, 185.00)
                        .volume(810_998.27)
                        .wap(183.9865)
                        .count(277_547),
                )
                .encode_proto(),
        ),
        proto_response(
            IncomingMessages::HistoricalDataEnd,
            historical_data_end_response()
                .start_date_str("20230413 16:31:22 UTC")
                .end_date_str("20230415 16:31:22 UTC")
                .encode_proto(),
        ),
    ]));

    let client = Client::stubbed(message_bus.clone(), server_versions::PROTOBUF_REST_MESSAGES_3);

    let contract = Contract::stock("MSFT").build();
    let end_date = datetime!(2023-04-15 16:31:22 UTC);
    let duration = 2.days();
    let bar_size = BarSize::Hour;
    let what_to_show = WhatToShow::Trades;
    let trading_hours = TradingHours::Regular;

    let historical_data = client
        .historical_data(&contract, bar_size)
        .what_to_show(what_to_show)
        .duration(duration)
        .ending(end_date)
        .fetch()
        .expect("historical data request failed");

    // Assert Response

    assert_eq!(historical_data.start, datetime!(2023-04-13 16:31:22 UTC), "historical_data.start");
    assert_eq!(historical_data.end, datetime!(2023-04-15 16:31:22 UTC), "historical_data.end");
    assert_eq!(historical_data.bars.len(), 2, "historical_data.bars.len()");

    assert_eq!(
        historical_data.bars[0].date,
        BarTimestamp::DateTime(datetime!(2023-04-13 00:00:00 UTC)),
        "bar.date"
    );
    assert_eq!(historical_data.bars[0].open, 182.94, "bar.open");
    assert_eq!(historical_data.bars[0].high, 186.50, "bar.high");
    assert_eq!(historical_data.bars[0].low, 180.94, "bar.low");
    assert_eq!(historical_data.bars[0].close, 185.90, "bar.close");
    assert_eq!(historical_data.bars[0].volume, 948837.22, "bar.volume");
    assert_eq!(historical_data.bars[0].wap, 184.869, "bar.wap");
    assert_eq!(historical_data.bars[0].count, 324891, "bar.count");

    assert_eq!(historical_data.bars[1].date, BarTimestamp::Date(date!(2023 - 04 - 14)), "daily bar.date");
    assert_eq!(historical_data.bars[1].open, 183.88, "bar[1].open");

    // Assert Request
    assert_eq!(request_message_count(&message_bus), 1);
    assert_request(
        &message_bus,
        0,
        &historical_data_request()
            .request_id(TEST_REQ_ID_FIRST)
            .contract(&contract)
            .end_date(Some(end_date))
            .duration(duration)
            .bar_size(bar_size)
            .what_to_show(what_to_show)
            .use_rth(trading_hours.use_rth()),
    );
}

#[test]
fn historical_schedules() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistoricalSchedule,
        historical_schedule_response().encode_proto(),
    )]));

    let client = Client::stubbed(message_bus.clone(), server_versions::PROTOBUF_REST_MESSAGES_3);

    let contract = Contract::stock("MSFT").build();
    let end_date = datetime!(2023-04-15 16:31:22 UTC);
    let duration = 7.days();

    let schedule = client
        .historical_schedules(&contract, duration)
        .ending(end_date)
        .fetch()
        .expect("historical schedule request failed");

    // Assert Response
    assert_eq!(schedule.time_zone, "US/Eastern", "schedule.time_zone");

    let time_zone: &Tz = time_tz::timezones::db::america::NEW_YORK;
    assert_eq!(
        schedule.start,
        datetime!(2023-04-14 09:30:00).assume_timezone(time_zone).unwrap(),
        "schedule.start"
    );
    assert_eq!(
        schedule.end,
        datetime!(2023-04-14 16:00:00).assume_timezone(time_zone).unwrap(),
        "schedule.end"
    );

    assert_eq!(schedule.sessions.len(), 1, "schedule.sessions.len()");
    assert_eq!(schedule.sessions[0].reference, date!(2023 - 04 - 14), "schedule.sessions[0].reference");
    assert_eq!(
        schedule.sessions[0].start,
        datetime!(2023-04-14 09:30:00).assume_timezone(time_zone).unwrap(),
        "schedule.sessions[0].start"
    );
    assert_eq!(
        schedule.sessions[0].end,
        datetime!(2023-04-14 16:00:00).assume_timezone(time_zone).unwrap(),
        "schedule.sessions[0].end"
    );

    // Assert Request
    assert_eq!(request_message_count(&message_bus), 1);
    assert_request(
        &message_bus,
        0,
        &historical_data_request()
            .request_id(TEST_REQ_ID_FIRST)
            .contract(&contract)
            .end_date(Some(end_date))
            .duration(duration)
            .bar_size(BarSize::Day)
            .what_to_show(WhatToShow::Schedule)
            .use_rth(true),
    );
}

#[test]
fn historical_ticks_bid_ask() {
    let message_bus = Arc::new(MessageBusStub::default());

    let client = Client::stubbed(message_bus.clone(), server_versions::HISTORICAL_TICKS);

    let contract = Contract::stock("MSFT").build();
    let start = datetime!(2023-04-01 09:30:00 UTC);
    let end = datetime!(2023-04-01 16:00:00 UTC);
    let number_of_ticks = 10;
    let trading_hours = TradingHours::Regular;

    let _tick_subscription = client
        .historical_ticks(&contract, number_of_ticks)
        .starting(start)
        .ending(end)
        .trading_hours(trading_hours)
        .bid_ask(IgnoreSize::Yes)
        .expect("historical ticks bid ask request failed");

    assert_eq!(request_message_count(&message_bus), 1);
    assert_request(
        &message_bus,
        0,
        &historical_ticks_request()
            .request_id(TEST_REQ_ID_FIRST)
            .contract(&contract)
            .start(Some(start))
            .end(Some(end))
            .number_of_ticks(number_of_ticks)
            .what_to_show(WhatToShow::BidAsk)
            .use_rth(trading_hours.use_rth())
            .ignore_size(true),
    );
}

#[test]
fn historical_ticks_mid_point() {
    let message_bus = Arc::new(MessageBusStub::default());

    let client = Client::stubbed(message_bus.clone(), server_versions::HISTORICAL_TICKS);

    let contract = Contract::stock("MSFT").build();
    let start = datetime!(2023-04-01 09:30:00 UTC);
    let end = datetime!(2023-04-01 16:00:00 UTC);
    let number_of_ticks = 10;
    let trading_hours = TradingHours::Regular;

    let _tick_subscription = client
        .historical_ticks(&contract, number_of_ticks)
        .starting(start)
        .ending(end)
        .trading_hours(trading_hours)
        .mid_point()
        .expect("historical ticks mid point request failed");

    assert_eq!(request_message_count(&message_bus), 1);
    assert_request(
        &message_bus,
        0,
        &historical_ticks_request()
            .request_id(TEST_REQ_ID_FIRST)
            .contract(&contract)
            .start(Some(start))
            .end(Some(end))
            .number_of_ticks(number_of_ticks)
            .what_to_show(WhatToShow::MidPoint)
            .use_rth(trading_hours.use_rth()),
    );
}

#[test]
fn historical_ticks_trade() {
    let message_bus = Arc::new(MessageBusStub::default());

    let client = Client::stubbed(message_bus.clone(), server_versions::HISTORICAL_TICKS);

    let contract = Contract::stock("MSFT").build();
    let start = datetime!(2023-04-01 09:30:00 UTC);
    let end = datetime!(2023-04-01 16:00:00 UTC);
    let number_of_ticks = 10;
    let trading_hours = TradingHours::Regular;

    let tick_subscription = client
        .historical_ticks(&contract, number_of_ticks)
        .starting(start)
        .ending(end)
        .trading_hours(trading_hours)
        .trade()
        .expect("historical ticks trade request failed");

    assert_eq!(tick_subscription.request_id(), TEST_REQ_ID_FIRST, "request_id() is the id on the wire");
    assert_eq!(request_message_count(&message_bus), 1);
    assert_request(
        &message_bus,
        0,
        &historical_ticks_request()
            .request_id(TEST_REQ_ID_FIRST)
            .contract(&contract)
            .start(Some(start))
            .end(Some(end))
            .number_of_ticks(number_of_ticks)
            .what_to_show(WhatToShow::Trades)
            .use_rth(trading_hours.use_rth()),
    );
}

#[test]
fn historical_data_adjusted_last_validation() {
    let message_bus = Arc::new(MessageBusStub::default());

    let client = Client::stubbed(message_bus, server_versions::SIZE_RULES);
    let contract = Contract::stock("MSFT").build();
    let end_date = datetime!(2023-04-15 16:31:22 UTC);
    let duration = 2.days();
    let bar_size = BarSize::Hour;
    let what_to_show = WhatToShow::AdjustedLast;

    // This should return an error because AdjustedLast can't be used with end_date
    let result = client
        .historical_data(&contract, bar_size)
        .what_to_show(what_to_show)
        .duration(duration)
        .ending(end_date)
        .fetch();
    assert!(result.is_err(), "Expected error due to AdjustedLast with end_date");

    match result {
        Err(Error::InvalidArgument(_)) => {
            // This is the expected error type
        }
        _ => panic!("Expected InvalidArgument error"),
    }
}

#[test]
fn historical_data_error_response() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_error_response(
        9000,
        200,
        "No security definition has been found for the request",
    )]));

    let client = Client::stubbed(message_bus, server_versions::SIZE_RULES);
    let contract = Contract::stock("MSFT").build();
    let duration = 2.days();
    let bar_size = BarSize::Hour;
    let what_to_show = WhatToShow::Trades;
    let trading_hours = TradingHours::Regular;

    // This should return an error because the server sent an error response
    let result = client
        .historical_data(&contract, bar_size)
        .what_to_show(what_to_show)
        .trading_hours(trading_hours)
        .duration(duration)
        .fetch();
    assert!(result.is_err(), "Expected error due to error response from server");
}

/// TWS sends 2188 on the error callback and then delivers the bars: the account
/// may not stream the up-to-the-second tail, but the historical data itself is
/// served. Classified as an error it would end the request before
/// `HistoricalData` arrived, and the caller would see no bars at all.
#[test]
fn historical_data_up_to_the_second_advisory_is_skipped() {
    let (client, _bus) = create_blocking_test_client_with_ordered_proto_responses(vec![
        proto_error_response(
            9000,
            2188,
            "Up-to-the-second historical data requires additional subscription for the API.",
        ),
        proto_response(
            IncomingMessages::HistoricalData,
            historical_data_response()
                .bar(
                    historical_data_daily_bar("20230315")
                        .ohlc(185.75, 186.25, 185.50, 186.00)
                        .volume(1500.0)
                        .wap(185.85)
                        .count(150),
                )
                .encode_proto(),
        ),
        proto_response(
            IncomingMessages::HistoricalDataEnd,
            historical_data_end_response()
                .start_date_str("20230315 09:30:00 UTC")
                .end_date_str("20230315 16:00:00 UTC")
                .encode_proto(),
        ),
    ]);

    let result = client
        .historical_data(&Contract::stock("SPY").build(), BarSize::Day)
        .duration(Duration::days(1))
        .ending(datetime!(2023-03-15 16:00:00 UTC))
        .fetch();

    let data = result.expect("advisory 2188 must not abort the request");
    assert_eq!(data.bars.len(), 1, "bars sent after the advisory should be returned");
}

/// The window frame ahead of the bars is out of order.
#[test]
fn historical_data_unexpected_response() {
    let (result, bus) = fetch_daily(vec![proto_response(
        IncomingMessages::HistoricalDataEnd,
        historical_data_end_response().encode_proto(),
    )]);
    assert!(matches!(result, Err(Error::UnexpectedResponse(_))), "{result:?}");
    // The window frame ended the request at TWS: nothing to cancel.
    assert_eq!(historical_requests_and_cancels(&bus), (1, 0));
}

/// A message type the request doesn't expect is skipped, not taken for a frame.
#[test]
fn historical_data_skips_undeclared_message() {
    let (result, bus) = fetch_daily(vec![proto_response(IncomingMessages::MarketDataType, Vec::new())]);
    assert!(matches!(result, Err(Error::UnexpectedEndOfStream)), "{result:?}");
    assert_eq!(historical_requests_and_cancels(&bus), (1, 1));
}

#[test]
fn tick_subscription_methods() {
    // For now, we'll use a minimal test to ensure the methods exist and are called correctly
    // Testing the subscription iterators fully would require more complex setup with mocked messages

    let message_bus = Arc::new(MessageBusStub::default());

    let client = Client::stubbed(message_bus, server_versions::HISTORICAL_TICKS);

    let contract = Contract::stock("MSFT").build();
    let number_of_ticks = 10;
    let trading_hours = TradingHours::Regular;

    let tick_subscription = client
        .historical_ticks(&contract, number_of_ticks)
        .trading_hours(trading_hours)
        .trade()
        .expect("historical ticks trade request failed");

    // Just test that these methods can be called without panicking
    let _iter = tick_subscription.iter();
    let _try_iter = tick_subscription.try_iter();
    let _timeout_iter = tick_subscription.timeout_iter(std::time::Duration::from_millis(100));

    // Test IntoIterator trait exists
    let _iter_ref: TickSubscriptionIter<TickLast> = (&tick_subscription).into_iter();
}

#[test]
fn tick_subscription_buffer_and_iteration() {
    // First batch: 3 ticks, done = false. Second batch: 2 ticks, done = true.
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![
        proto_response(
            IncomingMessages::HistoricalTickLast,
            historical_ticks_last_response()
                .tick(historical_tick_last(1_681_133_400, 11.63, 24_547.0, "ISLAND").special_conditions(" O X"))
                .tick(historical_tick_last(1_681_133_401, 11.64, 179.0, "FINRA"))
                .tick(historical_tick_last(1_681_133_402, 11.65, 200.0, "NYSE"))
                .done(false)
                .encode_proto(),
        ),
        proto_response(
            IncomingMessages::HistoricalTickLast,
            historical_ticks_last_response()
                .tick(historical_tick_last(1_681_133_403, 11.66, 100.0, "ARCA"))
                .tick(historical_tick_last(1_681_133_404, 11.67, 300.0, "BATS"))
                .done(true)
                .encode_proto(),
        ),
    ]));

    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);

    let contract = Contract::stock("MSFT").build();
    let number_of_ticks = 10;
    let trading_hours = TradingHours::Regular;

    let tick_subscription = client
        .historical_ticks(&contract, number_of_ticks)
        .trading_hours(trading_hours)
        .trade()
        .expect("historical ticks trade request failed");

    // Test standard data iterator (notices filtered, errors surface via Result)
    let mut ticks = Vec::new();
    for tick in tick_subscription.iter_data() {
        ticks.push(tick.expect("decode error"));
    }

    // Should have received all 5 ticks from both messages
    assert_eq!(ticks.len(), 5, "Expected 5 ticks in total");

    // Check specific values from first and last ticks
    assert_eq!(ticks[0].price, 11.63, "First tick price");
    assert_eq!(ticks[0].exchange, "ISLAND", "First tick exchange");

    assert_eq!(ticks[4].price, 11.67, "Last tick price");
    assert_eq!(ticks[4].exchange, "BATS", "Last tick exchange");
}

#[test]
fn tick_subscription_owned_iterator() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistoricalTickLast,
        historical_ticks_last_response()
            .tick(historical_tick_last(1_681_133_400, 11.70, 24_547.0, "ISLAND").special_conditions(" O X"))
            .tick(historical_tick_last(1_681_133_401, 11.71, 179.0, "FINRA"))
            .done(true)
            .encode_proto(),
    )]));

    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);

    let contract = Contract::stock("MSFT").build();
    let tick_subscription = client
        .historical_ticks(&contract, 10)
        .trade()
        .expect("historical ticks trade request failed");

    // Convert to owned iterator — yields the SubscriptionItem envelope.
    let ticks: Vec<TickLast> = tick_subscription
        .into_iter()
        .map(|item| item.expect("decode error").into_data().expect("unexpected notice"))
        .collect();

    assert_eq!(ticks.len(), 2, "Expected 2 ticks from owned iterator");
    assert_eq!(ticks[0].price, 11.70, "First tick price");
    assert_eq!(ticks[1].price, 11.71, "Second tick price");
}

#[test]
fn tick_subscription_bid_ask() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistoricalTickBidAsk,
        historical_ticks_bid_ask_response()
            .tick(historical_tick_bid_ask(1_681_133_399, 11.63, 11.83, 2_800.0, 100.0))
            .tick(historical_tick_bid_ask(1_681_133_400, 11.64, 11.84, 2_900.0, 200.0))
            .tick(historical_tick_bid_ask(1_681_133_401, 11.65, 11.85, 3_000.0, 300.0))
            .done(true)
            .encode_proto(),
    )]));

    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);

    let contract = Contract::stock("MSFT").build();
    let tick_subscription = client
        .historical_ticks(&contract, 10)
        .bid_ask(IgnoreSize::No)
        .expect("historical ticks bid_ask request failed");

    // Collect ticks
    let ticks: Vec<TickBidAsk> = tick_subscription.iter_data().map(|t| t.expect("decode error")).collect();

    assert_eq!(ticks.len(), 3, "Expected 3 bid/ask ticks");

    // Check first tick
    assert_eq!(ticks[0].price_bid, 11.63, "First tick bid price");
    assert_eq!(ticks[0].price_ask, 11.83, "First tick ask price");
    assert_eq!(ticks[0].size_bid, Some(2800.0), "First tick bid size");
    assert_eq!(ticks[0].size_ask, Some(100.0), "First tick ask size");

    // Check last tick
    assert_eq!(ticks[2].price_bid, 11.65, "Last tick bid price");
    assert_eq!(ticks[2].price_ask, 11.85, "Last tick ask price");
}

#[test]
fn tick_subscription_midpoint() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistoricalTick,
        historical_ticks_response()
            .tick(historical_tick_mid(1_681_133_398, 91.36, 0.0))
            .tick(historical_tick_mid(1_681_133_399, 91.37, 0.0))
            .tick(historical_tick_mid(1_681_133_400, 91.38, 0.0))
            .done(true)
            .encode_proto(),
    )]));

    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);

    let contract = Contract::stock("MSFT").build();
    let tick_subscription = client
        .historical_ticks(&contract, 10)
        .mid_point()
        .expect("historical ticks mid_point request failed");

    // Collect ticks
    let ticks: Vec<TickMidpoint> = tick_subscription.iter_data().map(|t| t.expect("decode error")).collect();

    assert_eq!(ticks.len(), 3, "Expected 3 midpoint ticks");

    // Check specific tick values
    assert_eq!(ticks[0].price, 91.36, "First tick price");
    assert_eq!(ticks[1].price, 91.37, "Second tick price");
    assert_eq!(ticks[2].price, 91.38, "Third tick price");
}

#[test]
fn historical_data_time_zone_handling() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![
        proto_response(
            IncomingMessages::HistoricalData,
            historical_data_response()
                .bar(
                    historical_data_bar(1_681_344_000)
                        .ohlc(182.94, 186.50, 180.94, 185.90)
                        .volume(948_837.22)
                        .wap(184.869)
                        .count(324_891),
                )
                .bar(
                    historical_data_bar(1_681_430_400)
                        .ohlc(183.88, 186.28, 182.01, 185.00)
                        .volume(810_998.27)
                        .wap(183.9865)
                        .count(277_547),
                )
                .encode_proto(),
        ),
        proto_response(
            IncomingMessages::HistoricalDataEnd,
            historical_data_end_response()
                .start_date_str("20230413 09:30:00 US/Eastern")
                .end_date_str("20230415 16:00:00 US/Eastern")
                .encode_proto(),
        ),
    ]));

    let client = Client::stubbed(message_bus.clone(), server_versions::PROTOBUF_REST_MESSAGES_3);

    let contract = Contract::stock("MSFT").build();
    let end_date = datetime!(2023-04-15 16:00:00 UTC);
    let duration = 2.days();
    let bar_size = BarSize::Day;
    let what_to_show = WhatToShow::Trades;

    let historical_data = client
        .historical_data(&contract, bar_size)
        .what_to_show(what_to_show)
        .duration(duration)
        .ending(end_date)
        .fetch()
        .expect("historical data request failed");

    let ny_zone = time_tz::timezones::db::america::NEW_YORK;

    assert_eq!(
        historical_data.start,
        datetime!(2023-04-13 09:30:00).assume_timezone(ny_zone).unwrap(),
        "start carried in HistoricalDataEnd's embedded TZ"
    );

    assert_eq!(
        historical_data.end,
        datetime!(2023-04-15 16:00:00).assume_timezone(ny_zone).unwrap(),
        "end carried in HistoricalDataEnd's embedded TZ"
    );
}

#[test]
fn historical_data_streaming_with_updates() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![
        proto_response(
            IncomingMessages::HistoricalData,
            historical_data_response()
                .bar(
                    historical_data_bar(1_678_886_400)
                        .ohlc(185.50, 186.00, 185.25, 185.75)
                        .volume(1000.0)
                        .wap(185.70)
                        .count(100),
                )
                .encode_proto(),
        ),
        proto_response(
            IncomingMessages::HistoricalDataUpdate,
            historical_data_update_response()
                .bar(
                    historical_data_bar(1_678_890_000)
                        .ohlc(185.80, 186.10, 185.60, 185.90)
                        .volume(500.0)
                        .wap(185.85)
                        .count(50),
                )
                .encode_proto(),
        ),
    ]));

    let mut client = Client::stubbed(message_bus.clone(), server_versions::PROTOBUF_REST_MESSAGES_3);
    client.time_zone = Some(time_tz::timezones::db::UTC);

    let contract = Contract::stock("SPY").build();

    let subscription = client
        .historical_data(&contract, BarSize::Hour)
        .duration(Duration::days(1))
        .stream()
        .expect("streaming request should succeed");

    // First: receive initial historical data
    let update1 = subscription.next_data();
    assert!(update1.is_some(), "Should receive initial historical data");
    match update1.unwrap().expect("expected Ok") {
        HistoricalBarUpdate::Historical(data) => {
            assert_eq!(data.bars.len(), 1, "Should have 1 initial bar");
            assert_eq!(data.bars[0].open, 185.50, "Wrong open price");
        }
        _ => panic!("Expected Historical variant"),
    }

    // Second: receive streaming update
    let update2 = subscription.next_data();
    assert!(update2.is_some(), "Should receive streaming update");
    match update2.unwrap().expect("expected Ok") {
        HistoricalBarUpdate::Update(bar) => {
            assert_eq!(bar.open, 185.80, "Wrong open price in update");
            assert_eq!(bar.high, 186.10, "Wrong high price in update");
            assert_eq!(bar.close, 185.90, "Wrong close price in update");
        }
        _ => panic!("Expected Update variant"),
    }

    // Verify request message was sent
    assert_eq!(request_message_count(&message_bus), 1);
    assert_request(
        &message_bus,
        0,
        &historical_data_request()
            .request_id(TEST_REQ_ID_FIRST)
            .contract(&contract)
            .duration(Duration::days(1))
            .bar_size(BarSize::Hour)
            .what_to_show(WhatToShow::Trades)
            .use_rth(true)
            .keep_up_to_date(true),
    );
}

// The dispatcher classifies the error frame; this asserts that *this API* hands
// it to the caller rather than swallowing it. The transport tests prove
// delivery, not consumption — see blocking `matching_symbols` (#735).
#[test]
fn historical_data_streaming_error_response() {
    let (client, _bus) = create_blocking_test_client_with_ordered_proto_responses(vec![proto_error_response(
        9000,
        162,
        "Historical Market Data Service error message:No market data permissions.",
    )]);
    let contract = Contract::stock("SPY").build();

    let subscription = client
        .historical_data(&contract, BarSize::Hour)
        .duration(Duration::days(1))
        .stream()
        .expect("streaming request should succeed");

    let Some(Err(err)) = subscription.next_data() else {
        panic!("error should arrive as Some(Err(_))");
    };
    assert_tws_error_message(err, 162, "No market data permissions");
}

#[test]
fn tick_subscription_sends_cancel_on_drop() {
    let message_bus = Arc::new(MessageBusStub::default());

    let internal = message_bus.send_request(RequestId::nth(100), &[]).unwrap();

    {
        let _subscription: TickSubscription<TickLast> = TickSubscription::new(internal, RequestId::nth(100).raw(), message_bus.clone());
        // subscription dropped here, !done so cancel should fire
    }

    let messages = message_bus.request_messages.read().unwrap();
    let cancel_msg = messages.last().expect("should have cancel message");
    assert_proto_msg_id(cancel_msg, OutgoingMessages::CancelHistoricalTicks);
}

#[test]
fn tick_subscription_explicit_cancel_prevents_duplicate_on_drop() {
    let message_bus = Arc::new(MessageBusStub::default());

    let internal = message_bus.send_request(RequestId::nth(101), &[]).unwrap();

    {
        let subscription: TickSubscription<TickLast> = TickSubscription::new(internal, RequestId::nth(101).raw(), message_bus.clone());
        subscription.cancel();
    }

    let messages = message_bus.request_messages.read().unwrap();
    assert_eq!(
        count_proto_msgs(&messages, OutgoingMessages::CancelHistoricalTicks),
        1,
        "should send cancel only once"
    );
}

#[test]
fn tick_subscription_drop_after_done_does_not_cancel() {
    let message_bus = Arc::new(MessageBusStub::default());

    let internal = message_bus.send_request(RequestId::nth(102), &[]).unwrap();

    {
        let subscription: TickSubscription<TickLast> = TickSubscription::new(internal, RequestId::nth(102).raw(), message_bus.clone());
        subscription.done.store(true, std::sync::atomic::Ordering::Relaxed);
        // drop with done=true → no cancel
    }

    let messages = message_bus.request_messages.read().unwrap();
    assert_eq!(
        count_proto_msgs(&messages, OutgoingMessages::CancelHistoricalTicks),
        0,
        "completed subscription should not send cancel on drop"
    );
}

#[test]
fn tick_subscription_cancel_after_done_writes_nothing() {
    let message_bus = Arc::new(MessageBusStub::default());

    let internal = message_bus.send_request(RequestId::nth(103), &[]).unwrap();
    let subscription: TickSubscription<TickLast> = TickSubscription::new(internal, RequestId::nth(103).raw(), message_bus.clone());
    subscription.done.store(true, std::sync::atomic::Ordering::Relaxed);

    subscription.cancel();

    let messages = message_bus.request_messages.read().unwrap();
    assert_eq!(
        count_proto_msgs(&messages, OutgoingMessages::CancelHistoricalTicks),
        0,
        "completed subscription should not send cancel"
    );
}

#[test]
fn streaming_subscription_sends_cancel_on_drop() {
    let message_bus = Arc::new(MessageBusStub::default());

    let mut client = Client::stubbed(message_bus.clone(), server_versions::SIZE_RULES);
    client.time_zone = Some(time_tz::timezones::db::UTC);
    let contract = Contract::stock("SPY").build();

    {
        let _subscription = client
            .historical_data(&contract, BarSize::Hour)
            .duration(Duration::days(1))
            .stream()
            .expect("streaming request should succeed");
    }

    let messages = message_bus.request_messages.read().unwrap();
    assert_eq!(
        count_proto_msgs(&messages, OutgoingMessages::CancelHistoricalData),
        1,
        "should send exactly one cancel message on drop"
    );
}

#[test]
fn streaming_subscription_cancel_prevents_duplicate_on_drop() {
    let message_bus = Arc::new(MessageBusStub::default());

    let mut client = Client::stubbed(message_bus.clone(), server_versions::SIZE_RULES);
    client.time_zone = Some(time_tz::timezones::db::UTC);
    let contract = Contract::stock("SPY").build();

    {
        let subscription = client
            .historical_data(&contract, BarSize::Hour)
            .duration(Duration::days(1))
            .stream()
            .expect("streaming request should succeed");

        subscription.cancel();
    }

    let messages = message_bus.request_messages.read().unwrap();
    assert_eq!(
        count_proto_msgs(&messages, OutgoingMessages::CancelHistoricalData),
        1,
        "should send cancel only once"
    );
}

#[test]
fn head_timestamp_skips_undeclared_message() {
    // 17 = HistoricalData — wrong type for head_timestamp (expects 88).
    assert_skips_undeclared(
        server_versions::SIZE_RULES,
        proto_response(IncomingMessages::HistoricalData, Vec::new()),
        OutgoingMessages::CancelHeadTimestamp,
        |c| c.head_timestamp(&Contract::stock("MSFT").build(), WhatToShow::Trades, TradingHours::Regular),
    );
}

#[test]
fn head_timestamp_end_of_stream() {
    // Empty response set → channel closes → subscription.next() yields None.
    let message_bus = Arc::new(MessageBusStub::default());
    let client = Client::stubbed(message_bus, server_versions::SIZE_RULES);
    let result = client.head_timestamp(&Contract::stock("MSFT").build(), WhatToShow::Trades, TradingHours::Regular);
    assert!(
        matches!(result, Err(Error::UnexpectedEndOfStream)),
        "expected UnexpectedEndOfStream, got {result:?}"
    );
}

#[test]
fn historical_data_with_end_message() {
    // start/end always come on the follow-on HistoricalDataEnd (108) frame at floor 210.
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![
        proto_response(
            IncomingMessages::HistoricalData,
            historical_data_response()
                .bar(
                    historical_data_bar(1_678_886_400)
                        .ohlc(185.50, 186.00, 185.25, 185.75)
                        .volume(1000.0)
                        .wap(185.70)
                        .count(100),
                )
                .encode_proto(),
        ),
        proto_response(
            IncomingMessages::HistoricalDataEnd,
            historical_data_end_response()
                .start_date_str("20230315 09:30:00 UTC")
                .end_date_str("20230315 10:30:00 UTC")
                .encode_proto(),
        ),
    ]));

    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);

    let data = client
        .historical_data(&Contract::stock("MSFT").build(), BarSize::Hour)
        .duration(Duration::days(1))
        .fetch()
        .expect("historical_data should succeed");

    assert_eq!(data.bars.len(), 1, "should have one bar");
    assert_eq!(data.start, datetime!(2023-03-15 09:30:00 UTC), "start populated from end-message");
    assert_eq!(data.end, datetime!(2023-03-15 10:30:00 UTC), "end populated from end-message");
}

#[test]
fn historical_data_end_of_stream_is_not_retried() {
    // Empty responses → channel closes immediately → subscription.next() yields
    // None. A closed stream is not a reset: it fails on the first attempt.
    let message_bus = Arc::new(MessageBusStub::default());
    let client = Client::stubbed(message_bus.clone(), server_versions::SIZE_RULES);

    let result = client
        .historical_data(&Contract::stock("MSFT").build(), BarSize::Hour)
        .duration(Duration::days(1))
        .fetch();

    assert!(
        matches!(result, Err(Error::UnexpectedEndOfStream)),
        "expected UnexpectedEndOfStream, got {result:?}"
    );
    assert_eq!(
        historical_requests_and_cancels(&message_bus),
        (1, 1),
        "a closed stream must not be retried"
    );
}

#[test]
fn historical_data_retries_a_connection_reset() {
    let message_bus = Arc::new(
        MessageBusStub::with_ordered_responses(vec![
            proto_response(
                IncomingMessages::HistoricalData,
                historical_data_response()
                    .bar(historical_data_bar(1_678_886_400).ohlc(185.50, 186.00, 185.25, 185.75))
                    .encode_proto(),
            ),
            proto_response(IncomingMessages::HistoricalDataEnd, historical_data_end_response().encode_proto()),
        ])
        .with_connection_resets(1),
    );
    let client = Client::stubbed(message_bus.clone(), server_versions::PROTOBUF_REST_MESSAGES_3);

    let data = client
        .historical_data(&Contract::stock("MSFT").build(), BarSize::Hour)
        .duration(Duration::days(1))
        .fetch()
        .expect("the reset must be retried, not surfaced");

    assert_eq!(data.bars.len(), 1);
    let (requests, _) = historical_requests_and_cancels(&message_bus);
    assert_eq!(requests, 2, "the retry must re-send the request");
}

#[test]
fn historical_data_error_message_response() {
    // Type 4 = IncomingMessages::Error — exercises the explicit Error arm
    // (Err::from(message)), distinct from the UnexpectedResponse arm.
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_error_response(
        9000,
        162,
        "Historical Market Data Service error message:No market data permissions.",
    )]));
    let client = Client::stubbed(message_bus.clone(), server_versions::SIZE_RULES);

    let result = client
        .historical_data(&Contract::stock("MSFT").build(), BarSize::Hour)
        .duration(Duration::days(1))
        .fetch();

    let err = result.expect_err("expected error from server");
    assert!(err.to_string().contains("No market data permissions"), "got: {err}");
    // TWS ended the query with its error; a cancel would draw 366.
    assert_eq!(historical_requests_and_cancels(&message_bus), (1, 0));
}

#[test]
fn historical_schedules_ending_now() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistoricalSchedule,
        historical_schedule_response().encode_proto(),
    )]));
    let client = Client::stubbed(message_bus.clone(), server_versions::PROTOBUF_REST_MESSAGES_3);

    let contract = Contract::stock("MSFT").build();
    let duration = 7.days();

    let schedule = client
        .historical_schedules(&contract, duration)
        .fetch()
        .expect("historical schedules ending now should succeed");

    assert_eq!(schedule.sessions.len(), 1);
    assert_eq!(request_message_count(&message_bus), 1);
    assert_request(
        &message_bus,
        0,
        &historical_data_request()
            .request_id(TEST_REQ_ID_FIRST)
            .contract(&contract)
            .duration(duration)
            .bar_size(BarSize::Day)
            .what_to_show(WhatToShow::Schedule)
            .use_rth(true),
    );
}

#[test]
fn historical_schedule_skips_undeclared_message() {
    assert_skips_undeclared(
        server_versions::HISTORICAL_SCHEDULE,
        proto_response(IncomingMessages::HistoricalData, Vec::new()),
        OutgoingMessages::CancelHistoricalData,
        |c| c.historical_schedules(&Contract::stock("MSFT").build(), Duration::days(1)).fetch(),
    );
}

#[test]
fn historical_schedule_end_of_stream() {
    let message_bus = Arc::new(MessageBusStub::default());
    let client = Client::stubbed(message_bus, server_versions::HISTORICAL_SCHEDULE);
    let result = client.historical_schedules(&Contract::stock("MSFT").build(), Duration::days(1)).fetch();
    assert!(
        matches!(result, Err(Error::UnexpectedEndOfStream)),
        "expected UnexpectedEndOfStream, got {result:?}"
    );
}

#[test]
fn cancel_historical_ticks() {
    let message_bus = Arc::new(MessageBusStub::default());
    let client = Client::stubbed(message_bus.clone(), server_versions::CANCEL_CONTRACT_DATA);

    client.cancel_historical_ticks(9000).expect("cancel should succeed");

    assert_eq!(request_message_count(&message_bus), 1);
    assert_request_msg_id(&message_bus, 0, OutgoingMessages::CancelHistoricalTicks);
}

#[test]
fn cancel_historical_ticks_version_check() {
    assert_version_check_fails(Features::CANCEL_CONTRACT_DATA, |c| c.cancel_historical_ticks(9000));
}

#[test]
fn histogram_data_empty_response_returns_empty_vec() {
    // None on the first subscription.next() → returns an empty Vec.
    let message_bus = Arc::new(MessageBusStub::default());
    let client = Client::stubbed(message_bus, server_versions::SIZE_RULES);

    let histogram = client
        .histogram_data(&Contract::stock("MSFT").build(), TradingHours::Regular, BarSize::Day)
        .expect("histogram_data should succeed with empty result");

    assert!(histogram.is_empty(), "expected empty histogram, got {histogram:?}");
}

#[test]
fn tick_subscription_try_next_drains_buffer() {
    // Pre-load a single batch with done=true so try_next() can drain without blocking.
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistoricalTickLast,
        historical_ticks_last_response()
            .tick(historical_tick_last(1_681_133_400, 12.00, 100.0, "NYSE"))
            .tick(historical_tick_last(1_681_133_401, 12.01, 200.0, "NYSE"))
            .done(true)
            .encode_proto(),
    )]));
    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);

    let subscription = client
        .historical_ticks(&Contract::stock("MSFT").build(), 10)
        .trade()
        .expect("subscription should be created");

    // Drain via try_iter_data — each .next() goes through try_next() → next_helper(try_next_routed).
    let ticks: Vec<TickLast> = subscription.try_iter_data().map(|t| t.expect("decode error")).collect();
    assert_eq!(ticks.len(), 2, "should drain both ticks via try_next");
    assert_eq!(ticks[0].price, 12.00);
    assert_eq!(ticks[1].price, 12.01);

    // After done, try_next returns None immediately.
    assert!(subscription.try_next().is_none(), "no more ticks expected after done");
}

#[test]
fn tick_subscription_next_timeout_drains_buffer() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistoricalTickLast,
        historical_ticks_last_response()
            .tick(historical_tick_last(1_681_133_400, 13.00, 100.0, "NYSE"))
            .done(true)
            .encode_proto(),
    )]));
    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);

    let subscription = client
        .historical_ticks(&Contract::stock("MSFT").build(), 10)
        .trade()
        .expect("subscription should be created");

    // Drive timeout_iter_data — each .next() goes through next_timeout() → next_helper.
    let ticks: Vec<TickLast> = subscription
        .timeout_iter_data(std::time::Duration::from_millis(50))
        .map(|t| t.expect("decode error"))
        .collect();
    assert_eq!(ticks.len(), 1, "should drain the single tick");
    assert_eq!(ticks[0].price, 13.00);

    // After done, next_timeout also returns None.
    assert!(
        subscription.next_timeout(std::time::Duration::from_millis(10)).is_none(),
        "no more ticks expected after done"
    );
}

#[test]
fn tick_subscription_error_surfaces_via_err_arm() {
    // #675 regression guard: a mid-stream RoutedItem::Error must surface through
    // the `Err` arm — NOT be swallowed as a silent `None` indistinguishable from
    // end-of-data. Subsequent calls return None (stream terminated).
    let (subscription, _signal_rx) = tick_sub_from_routed::<TickLast>(vec![RoutedItem::Error(Error::Simple("injected".into()))]);

    match subscription.next() {
        Some(Err(Error::Simple(msg))) => assert_eq!(msg, "injected", "error payload preserved"),
        other => panic!("expected Some(Err(Simple)), got {other:?}"),
    }
    assert!(subscription.next().is_none(), "stream terminates after a terminal error");
    subscription.done.store(true, Ordering::Relaxed); // prevent cancel-on-drop noise
}

#[test]
fn tick_subscription_notice_passes_through_then_data() {
    // A RoutedItem::Notice surfaces as SubscriptionItem::Notice (stream stays
    // open); a following tick batch is delivered after it.
    let (subscription, _signal_rx) = tick_sub_from_routed::<TickLast>(vec![
        RoutedItem::Notice(test_notice(2100, "API client has been unsubscribed from account data")),
        RoutedItem::Response(proto_response(
            IncomingMessages::HistoricalTickLast,
            historical_ticks_last_response()
                .tick(historical_tick_last(1_681_133_400, 15.00, 100.0, "NYSE"))
                .done(true)
                .encode_proto(),
        )),
    ]);

    match subscription.next() {
        Some(Ok(SubscriptionItem::Notice(n))) => assert_eq!(n.code, 2100),
        other => panic!("expected a Notice, got {other:?}"),
    }
    match subscription.next() {
        Some(Ok(SubscriptionItem::Data(tick))) => assert_eq!(tick.price, 15.00),
        other => panic!("expected tick data after notice, got {other:?}"),
    }

    // iter_data() filters the notice and yields only the tick.
    subscription.done.store(true, Ordering::Relaxed); // already drained; avoid cancel noise
}

#[test]
fn tick_subscription_iter_data_filters_notice_keeps_error() {
    // iter_data() drops the Notice but still surfaces the terminal Error.
    let (subscription, _signal_rx) = tick_sub_from_routed::<TickLast>(vec![
        RoutedItem::Notice(test_notice(2100, "warning")),
        RoutedItem::Error(Error::Simple("boom".into())),
    ]);

    let mut data = subscription.iter_data();
    match data.next() {
        Some(Err(Error::Simple(msg))) => assert_eq!(msg, "boom", "error survives the notice filter"),
        other => panic!("expected the error after the filtered notice, got {other:?}"),
    }
    assert!(data.next().is_none(), "stream ended after the error");
    subscription.done.store(true, Ordering::Relaxed);
}

#[test]
fn tick_subscription_none_when_channel_closes() {
    // Channel closes with no done=true → next_routed() yields None → next returns None.
    let (subscription, _signal_rx) = tick_sub_from_routed::<TickBidAsk>(vec![]); // empty + closed immediately
    assert!(subscription.next().is_none(), "closed channel → next_routed None → next returns None");
    subscription.done.store(true, Ordering::Relaxed);
}

#[test]
fn tick_subscription_midpoint_try_iter_and_timeout_iter() {
    // Exercise try_next + next_timeout on the TickMidpoint monomorphization so
    // every (TickT × iterator) combination shows up in coverage data.
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistoricalTick,
        historical_ticks_response()
            .tick(historical_tick_mid(1_681_133_400, 91.50, 0.0))
            .done(true)
            .encode_proto(),
    )]));
    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);

    let subscription = client
        .historical_ticks(&Contract::stock("MSFT").build(), 10)
        .mid_point()
        .expect("subscription should be created");

    let ticks_try: Vec<TickMidpoint> = subscription.try_iter_data().map(|t| t.expect("decode error")).collect();
    assert_eq!(ticks_try.len(), 1);

    // Timeout iterator after exhaustion — exercises next_timeout None path.
    assert!(subscription.timeout_iter(std::time::Duration::from_millis(10)).next().is_none());
}

#[test]
fn tick_subscription_bid_ask_try_iter_and_timeout_iter() {
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![proto_response(
        IncomingMessages::HistoricalTickBidAsk,
        historical_ticks_bid_ask_response()
            .tick(historical_tick_bid_ask(1_681_133_399, 11.63, 11.83, 2_800.0, 100.0))
            .done(true)
            .encode_proto(),
    )]));
    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);

    let subscription = client
        .historical_ticks(&Contract::stock("MSFT").build(), 10)
        .bid_ask(IgnoreSize::No)
        .expect("subscription should be created");

    let ticks_try: Vec<TickBidAsk> = subscription.try_iter_data().map(|t| t.expect("decode error")).collect();
    assert_eq!(ticks_try.len(), 1);
    assert!(subscription.timeout_iter(std::time::Duration::from_millis(10)).next().is_none());
}

#[test]
fn tick_subscription_skips_unexpected_message_then_yields() {
    // First response is HistoricalData — fill_buffer should silently skip it
    // (the `Some(Ok(message))` non-match arm) and loop to the next batch.
    let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![
        proto_response(IncomingMessages::HistoricalData, historical_data_response().encode_proto()),
        proto_response(
            IncomingMessages::HistoricalTickLast,
            historical_ticks_last_response()
                .tick(historical_tick_last(1_681_133_400, 14.00, 100.0, "NYSE"))
                .done(true)
                .encode_proto(),
        ),
    ]));
    let client = Client::stubbed(message_bus, server_versions::PROTOBUF_REST_MESSAGES_3);

    let subscription = client
        .historical_ticks(&Contract::stock("MSFT").build(), 1)
        .trade()
        .expect("subscription should be created");

    let tick = subscription
        .next_data()
        .expect("should receive tick after skipping unexpected")
        .expect("decode error");
    assert_eq!(tick.price, 14.00, "wrong price");
    assert!(subscription.next().is_none(), "should be done");
}

/// The historical data requests and cancels written to `bus`.
fn historical_requests_and_cancels(bus: &MessageBusStub) -> (usize, usize) {
    let messages = bus.request_messages.read().unwrap();
    (
        count_proto_msgs(&messages, OutgoingMessages::RequestHistoricalData),
        count_proto_msgs(&messages, OutgoingMessages::CancelHistoricalData),
    )
}

/// Fetch one day of daily bars against `responses`.
fn fetch_daily(responses: Vec<crate::messages::ResponseMessage>) -> (Result<HistoricalData, Error>, Arc<MessageBusStub>) {
    let (client, bus) = create_blocking_test_client_with_ordered_proto_responses(responses);
    let result = client
        .historical_data(&Contract::stock("MSFT").build(), BarSize::Day)
        .duration(Duration::seconds(86_400))
        .fetch();
    (result, bus)
}

/// A decoded batch alone cannot prove a finite request completed.
#[test]
fn historical_data_requires_end_after_a_batch() {
    for populated in [false, true] {
        let response = historical_data_response();
        let response = if populated {
            response.bar(historical_data_daily_bar("20230315").ohlc(1.0, 2.0, 0.5, 1.5))
        } else {
            response
        };
        let (result, bus) = fetch_daily(vec![proto_response(IncomingMessages::HistoricalData, response.encode_proto())]);
        assert!(matches!(result, Err(Error::UnexpectedEndOfStream)), "populated={populated}: {result:?}");
        assert_eq!(historical_requests_and_cancels(&bus), (1, 1), "bare closure must not replay the request");
    }
}

/// A second data frame is not the completion marker.
#[test]
fn historical_data_requires_end_message_type() {
    let response = proto_response(IncomingMessages::HistoricalData, historical_data_response().encode_proto());
    let (result, bus) = fetch_daily(vec![response.clone(), response]);
    assert!(matches!(result, Err(Error::UnexpectedResponse(_))), "{result:?}");
    assert_eq!(historical_requests_and_cancels(&bus), (1, 1));
}

/// A decode failure before the window frame cancels the request at TWS.
#[test]
fn historical_data_local_error_cancels() {
    let (result, bus) = fetch_daily(vec![proto_response(IncomingMessages::HistoricalData, vec![0xff])]);
    assert!(result.is_err(), "malformed bars must fail");
    assert_eq!(historical_requests_and_cancels(&bus), (1, 1));
}

/// `one_shot_timeout` bounds both reads: bars, then silence, times out
/// and cancels the request at TWS.
#[test]
fn historical_data_one_shot_timeout_cancels_between_frames() {
    let bars = proto_response(IncomingMessages::HistoricalData, historical_data_response().encode_proto());
    let bus = Arc::new(MessageBusStub::with_ordered_responses(vec![bars]).keeping_channels_open());
    let client = Client::stubbed(bus.clone(), server_versions::PROTOBUF_REST_MESSAGES_3).with_one_shot_timeout(std::time::Duration::from_millis(20));

    let result = client
        .historical_data(&Contract::stock("MSFT").build(), BarSize::Day)
        .duration(Duration::seconds(86_400))
        .fetch();

    assert!(matches!(result, Err(Error::Timeout)), "{result:?}");
    assert_eq!(historical_requests_and_cancels(&bus), (1, 1));
}

/// A reconnect that outlasts `one_shot_timeout` ends the fetch without a resend.
#[test]
fn historical_data_one_shot_timeout_skips_resend_past_deadline() {
    let bars = proto_response(IncomingMessages::HistoricalData, historical_data_response().encode_proto());
    let end = proto_response(IncomingMessages::HistoricalDataEnd, historical_data_end_response().encode_proto());
    let bus = Arc::new(
        MessageBusStub::with_ordered_responses(vec![bars, end])
            .with_connection_resets(1)
            .with_reconnect_delay(std::time::Duration::from_millis(40)),
    );
    let client = Client::stubbed(bus.clone(), server_versions::PROTOBUF_REST_MESSAGES_3).with_one_shot_timeout(std::time::Duration::from_millis(20));

    let result = client
        .historical_data(&Contract::stock("MSFT").build(), BarSize::Day)
        .duration(Duration::seconds(86_400))
        .fetch();

    assert!(matches!(result, Err(Error::Timeout)), "{result:?}");
    assert_eq!(historical_requests_and_cancels(&bus).0, 1, "no resend past the deadline");
}

/// An empty result is valid when the broker explicitly completes the request.
#[test]
fn historical_data_requires_end_accepts_completed_empty_batch() {
    let (result, bus) = fetch_daily(vec![
        proto_response(IncomingMessages::HistoricalData, historical_data_response().encode_proto()),
        proto_response(
            IncomingMessages::HistoricalDataEnd,
            historical_data_end_response()
                .start_date_str("20230315 09:30:00 UTC")
                .end_date_str("20230315 10:30:00 UTC")
                .encode_proto(),
        ),
    ]);
    let result = result.unwrap();
    assert!(result.bars.is_empty());
    assert_eq!(result.start, datetime!(2023-03-15 09:30:00 UTC));
    assert_eq!(result.end, datetime!(2023-03-15 10:30:00 UTC));
    assert_eq!(request_message_count(&bus), 1);
}

// ---- buffer_limit ----------------------------------------------------------

#[test]
fn historical_data_stream_buffer_limit_reaches_the_bus() {
    let (client, message_bus) = crate::common::test_utils::helpers::create_blocking_test_client();
    let contract = Contract::stock("SPY").build();

    let _bounded = client
        .historical_data(&contract, BarSize::Hour)
        .duration(Duration::days(1))
        .buffer_limit(16)
        .stream()
        .expect("bounded stream failed");
    let _unbounded = client
        .historical_data(&contract, BarSize::Hour)
        .duration(Duration::days(1))
        .stream()
        .expect("unbounded stream failed");

    assert_eq!(
        *message_bus.buffer_limits.read().unwrap(),
        vec![16],
        "only the bounded request carries a limit"
    );
}

#[test]
fn historical_data_fetch_rejects_buffer_limit() {
    let (client, message_bus) = crate::common::test_utils::helpers::create_blocking_test_client();
    let contract = Contract::stock("SPY").build();

    let result = client
        .historical_data(&contract, BarSize::Hour)
        .duration(Duration::days(1))
        .buffer_limit(16)
        .fetch();

    assert!(matches!(result, Err(Error::InvalidArgument(_))), "got {result:?}");
    assert!(message_bus.request_messages.read().unwrap().is_empty(), "nothing is sent");
}