mxaccessgw/clients/rust/tests/client_behavior.rs

use std::pin::Pin;
use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc,
};
use std::task::{Context, Poll};
use std::time::Duration;

use futures_core::Stream;
use futures_util::StreamExt;
use serde_json::Value;
use tokio::net::TcpListener;
use tokio::sync::{mpsc, Mutex};
use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream};
use tonic::transport::Server;
use tonic::{Request, Response, Status};
use zb_mom_ww_mxgateway_client::error::ensure_mxaccess_success;
use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::mx_access_gateway_server::{
    MxAccessGateway, MxAccessGatewayServer,
};
use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::mx_command;
use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::mx_command_reply;
use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::mx_value::Kind;
use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::{
    alarm_feed_message, AcknowledgeAlarmReply, AcknowledgeAlarmRequest, ActiveAlarmSnapshot,
    AddItem2Reply, AddItemReply, AlarmConditionState, AlarmFeedMessage, AlarmTransitionKind,
    AuthenticateUserCommand, AuthenticateUserReply, BulkReadReply, BulkReadResult,
    BulkSubscribeReply, BulkWriteReply, BulkWriteResult, CloseSessionReply, CloseSessionRequest,
    MxCommandKind, MxCommandReply, MxDataType, MxEvent, MxEventFamily, MxSparseArray,
    MxSparseElement, MxStatusCategory, MxStatusProxy, MxStatusSource, MxValue,
    OnAlarmTransitionEvent, OpenSessionReply, OpenSessionRequest, ProtocolStatus,
    ProtocolStatusCode, QueryActiveAlarmsRequest, RegisterReply, ReplayGap, SessionState,
    StreamAlarmsRequest, StreamEventsRequest, SubscribeResult, Write2BulkEntry, WriteBulkEntry,
    WriteCommand, WriteSecured2BulkEntry, WriteSecuredBulkEntry,
};
use zb_mom_ww_mxgateway_client::{
    next_correlation_id, ApiKey, ClientOptions, CommandError, Error, EventItem, GatewayClient,
    MxStatus, MxValue as ClientMxValue, MxValueProjection,
};

#[tokio::test]
async fn fake_server_receives_bearer_metadata_and_raw_client_is_reachable() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let mut client = GatewayClient::connect(
        ClientOptions::new(endpoint).with_api_key(ApiKey::new("mxgw_fixture_secret")),
    )
    .await
    .unwrap();

    let _raw = client.raw_client();
    let session = client
        .open_session(OpenSessionRequest {
            client_session_name: "rust-test".to_owned(),
            ..OpenSessionRequest::default()
        })
        .await
        .unwrap();

    assert_eq!(session.id(), "session-fixture");
    assert_eq!(
        state.authorization.lock().await.as_deref(),
        Some("Bearer mxgw_fixture_secret")
    );
}

#[tokio::test]
async fn session_helpers_build_commands_and_preserve_command_errors() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let item_handle = session.add_item(12, "Plant.Area.Tag").await.unwrap();

    assert_eq!(item_handle, 34);
    let last_command = state.last_command_kind.lock().await;
    assert_eq!(*last_command, Some(MxCommandKind::AddItem as i32));
    drop(last_command);

    let error = session
        .write(12, 34, ClientMxValue::int32(123), 0)
        .await
        .unwrap_err();
    // A MXACCESS_FAILURE-coded reply is an MXAccess-level failure, routed to
    // Error::MxAccess (matching .NET/Java/Go/Python) rather than Error::Command.
    let Error::MxAccess(error) = error else {
        panic!("MXACCESS_FAILURE reply should route to Error::MxAccess: {error:?}");
    };
    assert_eq!(
        error.reply().protocol_status.as_ref().unwrap().code,
        ProtocolStatusCode::MxaccessFailure as i32
    );
    assert_eq!(error.reply().hresult, Some(-2147220992));
    assert_eq!(error.reply().statuses.len(), 2);
}

#[tokio::test]
async fn subscribe_bulk_builds_one_bulk_command_and_returns_results() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let results = session
        .subscribe_bulk(12, vec!["Area001.Pump001.Speed".to_owned()])
        .await
        .unwrap();

    assert_eq!(results[0].item_handle, 34);
    let last_command = state.last_command_kind.lock().await;
    assert_eq!(*last_command, Some(MxCommandKind::SubscribeBulk as i32));
}

#[tokio::test]
async fn event_stream_preserves_order_and_drop_cancels_server_stream() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();

    let mut stream = client
        .stream_events(StreamEventsRequest {
            session_id: "session-fixture".to_owned(),
            after_worker_sequence: 0,
        })
        .await
        .unwrap();

    assert_eq!(
        stream
            .next()
            .await
            .unwrap()
            .unwrap()
            .as_event()
            .unwrap()
            .worker_sequence,
        1
    );
    assert_eq!(
        stream
            .next()
            .await
            .unwrap()
            .unwrap()
            .as_event()
            .unwrap()
            .worker_sequence,
        2
    );

    drop(stream);
    for _ in 0..20 {
        if state.stream_dropped.load(Ordering::SeqCst) {
            return;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }

    assert!(state.stream_dropped.load(Ordering::SeqCst));
}

#[tokio::test]
async fn replay_gap_sentinel_surfaces_as_typed_event_item() {
    let state = Arc::new(FakeState::default());
    // Script a resumed stream: the reconnect-replay gap sentinel at the head
    // (family UNSPECIFIED, no body, `replay_gap` set) followed by a normal
    // event. The client must promote the sentinel to `EventItem::ReplayGap`
    // and leave the following event as a normal `EventItem::Event`.
    *state.stream_events_script.lock().await = Some(vec![
        MxEvent {
            replay_gap: Some(ReplayGap {
                requested_after_sequence: 5,
                oldest_available_sequence: 42,
            }),
            ..MxEvent::default()
        },
        event(42),
    ]);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();

    let mut stream = client
        .stream_events(StreamEventsRequest {
            session_id: "session-fixture".to_owned(),
            after_worker_sequence: 5,
        })
        .await
        .unwrap();

    // First item is the typed gap sentinel, not a normal event.
    let first = stream.next().await.unwrap().unwrap();
    match &first {
        EventItem::ReplayGap(gap) => {
            assert_eq!(gap.requested_after_sequence, 5);
            assert_eq!(gap.oldest_available_sequence, 42);
        }
        EventItem::Event(_) => panic!("expected a ReplayGap sentinel, got a normal event"),
    }
    // Accessor helpers reflect the variant.
    assert!(first.as_event().is_none());
    assert_eq!(first.replay_gap().unwrap().oldest_available_sequence, 42);

    // The normal event that follows is unaffected.
    let second = stream.next().await.unwrap().unwrap();
    assert_eq!(second.as_event().unwrap().worker_sequence, 42);
    assert!(second.replay_gap().is_none());
}

#[tokio::test]
async fn acknowledge_alarm_returns_reply_with_native_status() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();

    let reply = client
        .acknowledge_alarm(AcknowledgeAlarmRequest {
            client_correlation_id: "corr-1".to_owned(),
            alarm_full_reference: "Tank01.Level.HiHi".to_owned(),
            comment: "investigating".to_owned(),
            operator_user: "alice".to_owned(),
        })
        .await
        .unwrap();

    assert_eq!(
        reply.protocol_status.as_ref().unwrap().code,
        ProtocolStatusCode::Ok as i32
    );
    assert_eq!(reply.status.as_ref().unwrap().success, 1);
}

#[tokio::test]
async fn query_active_alarms_streams_snapshot_rows() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();

    let mut stream = client
        .query_active_alarms(QueryActiveAlarmsRequest {
            session_id: "session-fixture".to_owned(),
            ..QueryActiveAlarmsRequest::default()
        })
        .await
        .unwrap();

    let first = stream.next().await.unwrap().unwrap();
    assert_eq!(first.alarm_full_reference, "Tank01.Level.HiHi");
}

#[test]
fn value_conversion_fixtures_keep_typed_projection_and_raw_metadata() {
    let fixture = behavior_fixture("values/value-conversion-cases.json");
    let cases = fixture["cases"].as_array().unwrap();

    let int64_case = case_by_id(cases, "int64.large");
    let int64_value = ClientMxValue::from_proto(MxValue {
        data_type: MxDataType::Integer as i32,
        variant_type: "VT_I8".to_owned(),
        kind: Some(Kind::Int64Value(
            int64_case["value"]["int64Value"]
                .as_str()
                .unwrap()
                .parse()
                .unwrap(),
        )),
        ..MxValue::default()
    });
    assert_eq!(
        int64_value.projection(),
        &MxValueProjection::Int64(9_223_372_036_854_770_000)
    );

    let raw_case = case_by_id(cases, "raw-fallback.variant");
    let raw_value = ClientMxValue::from_proto(MxValue {
        data_type: MxDataType::Unknown as i32,
        variant_type: "VT_RECORD".to_owned(),
        raw_diagnostic: raw_case["value"]["rawDiagnostic"]
            .as_str()
            .unwrap()
            .to_owned(),
        raw_data_type: raw_case["value"]["rawDataType"].as_i64().unwrap() as i32,
        kind: Some(Kind::RawValue(vec![1, 2, 3, 4, 5])),
        ..MxValue::default()
    });
    assert_eq!(
        raw_value.projection(),
        &MxValueProjection::Raw(vec![1, 2, 3, 4, 5])
    );
    assert_eq!(raw_value.raw().raw_data_type, 32767);
    assert!(raw_value.raw().raw_diagnostic.contains("No lossless"));
}

#[test]
fn status_conversion_fixtures_preserve_raw_fields() {
    let fixture = behavior_fixture("statuses/status-conversion-cases.json");
    let cases = fixture["cases"].as_array().unwrap();
    let raw_case = case_by_id(cases, "raw-unknown-category");
    let status = MxStatus::from_proto(MxStatusProxy {
        success: raw_case["status"]["success"].as_i64().unwrap() as i32,
        category: MxStatusCategory::Unknown as i32,
        detected_by: MxStatusSource::Unknown as i32,
        detail: raw_case["status"]["detail"].as_i64().unwrap() as i32,
        raw_category: raw_case["status"]["rawCategory"].as_i64().unwrap() as i32,
        raw_detected_by: raw_case["status"]["rawDetectedBy"].as_i64().unwrap() as i32,
        diagnostic_text: raw_case["status"]["diagnosticText"]
            .as_str()
            .unwrap()
            .to_owned(),
    });

    assert_eq!(status.success(), 0);
    assert_eq!(status.category(), Some(MxStatusCategory::Unknown));
    assert_eq!(status.raw_category(), 99);
    assert_eq!(status.raw_detected_by(), 77);
    assert!(status.diagnostic_text().contains("preserved"));
}

#[test]
fn authentication_and_authorization_statuses_are_distinct_and_redacted() {
    let auth = Error::from(Status::unauthenticated(
        "invalid API key mxgw_visible_secret",
    ));
    let denied = Error::from(Status::permission_denied("missing scope mxaccess.write"));

    assert!(matches!(auth, Error::Authentication { .. }));
    assert!(matches!(denied, Error::Authorization { .. }));
    assert!(!auth.to_string().contains("visible_secret"));
}

#[test]
fn command_reply_validation_fixtures_branch_on_category_and_negative_hresult() {
    // The shared behavior fixtures pin both reply-validation rules: a status
    // entry fails iff its category is not OK (the raw `success` member is
    // diagnostics only) and an HRESULT fails iff it is present and negative.
    for (fixture, expect_failure) in [
        ("register.ok.reply.json", false),
        ("write.status-category-error-success-set.reply.json", true),
        ("write.status-category-ok-success-zero.reply.json", false),
        ("write.hresult-s-false.reply.json", false),
        ("write.hresult-e-fail.reply.json", true),
    ] {
        let reply = command_reply_fixture(fixture);
        let result = ensure_mxaccess_success(reply);

        assert_eq!(
            result.is_err(),
            expect_failure,
            "fixture {fixture} expected failure = {expect_failure}, got {result:?}"
        );
    }
}

#[test]
fn status_entry_verdict_ignores_the_raw_success_member() {
    // Edges the fixtures cannot express: an OK category always passes and an
    // unspecified category always fails, whatever `success` carries.
    for (category, success, expect_failure) in [
        (MxStatusCategory::Ok, 0, false),
        (MxStatusCategory::Ok, 1, false),
        (MxStatusCategory::CommunicationError, 1, true),
        (MxStatusCategory::Unspecified, 1, true),
    ] {
        let reply = MxCommandReply {
            protocol_status: Some(ok_status("command ok")),
            statuses: vec![MxStatusProxy {
                success,
                category: category as i32,
                ..MxStatusProxy::default()
            }],
            ..MxCommandReply::default()
        };

        assert_eq!(
            ensure_mxaccess_success(reply).is_err(),
            expect_failure,
            "category {category:?} with success {success} expected failure = {expect_failure}"
        );
    }
}

#[test]
fn command_error_display_keeps_raw_reply_accessible() {
    let reply = mxaccess_failure_reply();
    let error = CommandError::new(reply.clone());

    assert_eq!(error.reply().hresult, Some(-2147220992));
    assert!(error.to_string().contains("MxaccessFailure"));
}

// ---- Client.Rust-022 / 024 regression coverage ---------------------------

#[tokio::test]
async fn register_returns_malformed_reply_when_ok_reply_has_no_payload() {
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::OkWithoutPayload);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session.register("client").await.unwrap_err();

    assert!(
        matches!(error, Error::MalformedReply { .. }),
        "expected MalformedReply, got {error:?}"
    );
}

#[tokio::test]
async fn add_item_returns_malformed_reply_when_ok_reply_has_no_payload() {
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::OkWithoutPayload);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session.add_item(12, "Plant.Area.Tag").await.unwrap_err();

    assert!(
        matches!(error, Error::MalformedReply { .. }),
        "expected MalformedReply, got {error:?}"
    );
}

#[tokio::test]
async fn add_item2_returns_malformed_reply_when_ok_reply_has_no_payload() {
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::OkWithoutPayload);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session
        .add_item2(12, "Plant.Area.Tag", "ctx")
        .await
        .unwrap_err();

    assert!(
        matches!(error, Error::MalformedReply { .. }),
        "expected MalformedReply, got {error:?}"
    );
}

#[tokio::test]
async fn subscribe_bulk_returns_malformed_reply_on_mismatched_payload_arm() {
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::OkWithMismatchedPayload);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session
        .subscribe_bulk(12, vec!["Area001.Pump001.Speed".to_owned()])
        .await
        .unwrap_err();

    assert!(
        matches!(error, Error::MalformedReply { .. }),
        "expected MalformedReply, got {error:?}"
    );
}

#[tokio::test]
async fn read_bulk_returns_malformed_reply_on_mismatched_payload_arm() {
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::OkWithMismatchedPayload);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session
        .read_bulk(12, &["Area001.Pump001.Speed".to_owned()], 1000)
        .await
        .unwrap_err();

    assert!(
        matches!(error, Error::MalformedReply { .. }),
        "expected MalformedReply, got {error:?}"
    );
}

#[tokio::test]
async fn write_bulk_returns_malformed_reply_on_mismatched_payload_arm() {
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::OkWithMismatchedPayload);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session
        .write_bulk(
            12,
            vec![WriteBulkEntry {
                item_handle: 34,
                value: Some(ClientMxValue::int32(1).into_proto()),
                user_id: 0,
            }],
        )
        .await
        .unwrap_err();

    assert!(
        matches!(error, Error::MalformedReply { .. }),
        "expected MalformedReply, got {error:?}"
    );
}

#[tokio::test]
async fn unary_invoke_maps_status_unavailable_to_error_unavailable() {
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await =
        Some(InvokeOverride::Unavailable("gateway restarting".to_owned()));
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session.add_item(12, "Plant.Area.Tag").await.unwrap_err();

    assert!(
        matches!(error, Error::Unavailable { .. }),
        "expected Unavailable, got {error:?}"
    );
}

#[tokio::test]
async fn read_bulk_round_trips_through_the_fake_gateway() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let results = session
        .read_bulk(12, &["Area001.Pump001.Speed".to_owned()], 1000)
        .await
        .unwrap();

    assert_eq!(results.len(), 1);
    assert!(results[0].was_successful);
    assert!(results[0].was_cached);
}

#[tokio::test]
async fn write_bulk_round_trips_through_the_fake_gateway() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let results = session
        .write_bulk(
            12,
            vec![WriteBulkEntry {
                item_handle: 34,
                value: Some(ClientMxValue::int32(1).into_proto()),
                user_id: 0,
            }],
        )
        .await
        .unwrap();

    assert_eq!(results.len(), 1);
    assert!(results[0].was_successful);
    let last_command = state.last_command_kind.lock().await;
    assert_eq!(*last_command, Some(MxCommandKind::WriteBulk as i32));
}

#[tokio::test]
async fn write2_bulk_round_trips_through_the_fake_gateway() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let results = session
        .write2_bulk(
            12,
            vec![Write2BulkEntry {
                item_handle: 34,
                value: Some(ClientMxValue::int32(1).into_proto()),
                timestamp_value: Some(ClientMxValue::string("2026-05-24T00:00:00Z").into_proto()),
                user_id: 0,
            }],
        )
        .await
        .unwrap();

    assert_eq!(results.len(), 1);
    assert!(results[0].was_successful);
    let last_command = state.last_command_kind.lock().await;
    assert_eq!(*last_command, Some(MxCommandKind::Write2Bulk as i32));
}

#[tokio::test]
async fn write_secured_bulk_round_trips_through_the_fake_gateway() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let results = session
        .write_secured_bulk(
            12,
            vec![WriteSecuredBulkEntry {
                item_handle: 34,
                value: Some(ClientMxValue::int32(1).into_proto()),
                current_user_id: 0,
                verifier_user_id: 0,
            }],
        )
        .await
        .unwrap();

    assert_eq!(results.len(), 1);
    assert!(results[0].was_successful);
    let last_command = state.last_command_kind.lock().await;
    assert_eq!(*last_command, Some(MxCommandKind::WriteSecuredBulk as i32));
}

#[tokio::test]
async fn write_secured2_bulk_round_trips_through_the_fake_gateway() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let results = session
        .write_secured2_bulk(
            12,
            vec![WriteSecured2BulkEntry {
                item_handle: 34,
                value: Some(ClientMxValue::int32(1).into_proto()),
                timestamp_value: Some(ClientMxValue::string("2026-05-24T00:00:00Z").into_proto()),
                current_user_id: 0,
                verifier_user_id: 0,
            }],
        )
        .await
        .unwrap();

    assert_eq!(results.len(), 1);
    assert!(results[0].was_successful);
    let last_command = state.last_command_kind.lock().await;
    assert_eq!(*last_command, Some(MxCommandKind::WriteSecured2Bulk as i32));
}

#[tokio::test]
async fn advise_supervisory_round_trips_and_sends_advise_supervisory_kind() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    session.advise_supervisory(12, 34).await.unwrap();

    let last_command = state.last_command_kind.lock().await;
    assert_eq!(*last_command, Some(MxCommandKind::AdviseSupervisory as i32));
}

#[tokio::test]
async fn unregister_round_trips_and_sends_unregister_kind() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    session.unregister(12).await.unwrap();

    let last_command = state.last_command_kind.lock().await;
    assert_eq!(*last_command, Some(MxCommandKind::Unregister as i32));
}

#[tokio::test]
async fn write_secured_surfaces_native_mxaccess_failure_and_redacts_diagnostic() {
    // MXAccess parity: WriteSecured failing (e.g. before authenticate +
    // advise-supervisory) is a native outcome, surfaced — not smoothed over.
    // The scripted reply carries an Ok protocol envelope but a negative
    // hresult, so this also proves the typed helper runs ensure_mxaccess_success.
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::MxAccessFailure);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session
        .write_secured(12, 34, 0, 0, ClientMxValue::int32(1))
        .await
        .unwrap_err();

    let Error::MxAccess(mx_access) = &error else {
        panic!("write_secured must surface the native failure as Error::MxAccess: {error:?}");
    };
    assert_eq!(mx_access.reply().hresult, Some(-2_147_217_900));
    let rendered = error.to_string();
    assert!(rendered.contains("<redacted>"), "diagnostic: {rendered}");
    assert!(
        !rendered.contains("leaked_secret"),
        "credential-shaped diagnostic must be scrubbed: {rendered}"
    );

    let last_command = state.last_command_kind.lock().await;
    assert_eq!(*last_command, Some(MxCommandKind::WriteSecured as i32));
}

#[tokio::test]
async fn authenticate_user_returns_user_id_and_transmits_credential_on_wire() {
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let user_id = session
        .authenticate_user(7, "verifier", "sup3r-s3cret-pw")
        .await
        .unwrap();

    assert_eq!(user_id, 4242);
    let captured = state
        .last_authenticate_user
        .lock()
        .await
        .take()
        .expect("fake should have captured an AuthenticateUserCommand");
    assert_eq!(captured.server_handle, 7);
    assert_eq!(captured.verify_user, "verifier");
    // The credential must reach the wire so authentication can succeed...
    assert_eq!(captured.verify_user_password, "sup3r-s3cret-pw");
    let last_command = state.last_command_kind.lock().await;
    assert_eq!(*last_command, Some(MxCommandKind::AuthenticateUser as i32));
}

#[tokio::test]
async fn authenticate_user_keeps_credentials_out_of_surfaced_errors() {
    // ...but a native authentication failure must never leak the credential
    // into the error's Display or Debug rendering.
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::MxAccessFailure);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let password = "unique-credential-9f3b2";
    let error = session
        .authenticate_user(7, "verifier", password)
        .await
        .unwrap_err();

    assert!(
        matches!(error, Error::MxAccess(_)),
        "native auth failure should surface as Error::MxAccess: {error:?}"
    );
    let display = error.to_string();
    let debug = format!("{error:?}");
    assert!(
        !display.contains(password),
        "credential leaked into Display: {display}"
    );
    assert!(
        !debug.contains(password),
        "credential leaked into Debug: {debug}"
    );
}

#[tokio::test]
async fn authenticate_user_scrubs_exact_caller_credential_echoed_in_diagnostic() {
    // CLI-40: MXAccess can echo the supplied credential back inside its failure
    // diagnostic (here in statuses[0].diagnostic_text). The token has no
    // mxgw_/bearer shape, so the pattern scrub alone cannot catch it — the
    // exact-secret scrub must replace the caller's password with <redacted>.
    let credential = "sup3rSecretVerify9f3a2b";
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::CannedReply(Box::new(
        command_reply_fixture("authenticate-user.echoed-credential.reply.json"),
    )));
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session
        .authenticate_user(7, "verifier", credential)
        .await
        .unwrap_err();

    assert!(
        matches!(error, Error::MxAccess(_)),
        "OK protocol + negative hresult must route to Error::MxAccess: {error:?}"
    );
    let rendered = error.to_string();
    assert!(
        !rendered.contains(credential),
        "exact caller credential leaked into the surfaced error: {rendered}"
    );
    assert!(
        rendered.contains("<redacted>"),
        "credential occurrence must be replaced with <redacted>: {rendered}"
    );
}

/// Drive `authenticate_user` against a canned reply that echoes the caller's
/// credential in every string field, then assert the surfaced
/// [`Error::MxAccess`] leaks it nowhere — neither through the structured reply a
/// caller can read back (`reply().protocol_status.message`,
/// `reply().diagnostic_message`, `reply().statuses[i].diagnostic_text`) nor
/// through `Display`/`Debug`.
async fn assert_authenticate_user_scrubs_structured_reply(fixture: &str) {
    let credential = "sup3rSecretVerify9f3a2b";
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::CannedReply(Box::new(
        command_reply_fixture(fixture),
    )));
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session
        .authenticate_user(7, "verifier", credential)
        .await
        .unwrap_err();

    let Error::MxAccess(mx_access) = &error else {
        panic!("{fixture}: credential-echoed reply must route to Error::MxAccess, got {error:?}");
    };

    // The structured reply a caller can read back must be scrubbed too — the raw
    // MxCommandReply otherwise reintroduces the leak Display/Debug already close.
    let reply = mx_access.reply();
    if let Some(status) = reply.protocol_status.as_ref() {
        assert!(
            !status.message.contains(credential),
            "{fixture}: credential leaked via reply().protocol_status.message: {}",
            status.message
        );
    }
    assert!(
        !reply.diagnostic_message.contains(credential),
        "{fixture}: credential leaked via reply().diagnostic_message: {}",
        reply.diagnostic_message
    );
    for (index, status) in reply.statuses.iter().enumerate() {
        assert!(
            !status.diagnostic_text.contains(credential),
            "{fixture}: credential leaked via reply().statuses[{index}].diagnostic_text: {}",
            status.diagnostic_text
        );
    }

    let display = error.to_string();
    let debug = format!("{error:?}");
    assert!(
        !display.contains(credential),
        "{fixture}: credential leaked into Display: {display}"
    );
    assert!(
        !debug.contains(credential),
        "{fixture}: credential leaked into Debug: {debug}"
    );
    assert!(
        display.contains("<redacted>"),
        "{fixture}: Display must mark the redaction: {display}"
    );
}

#[tokio::test]
async fn authenticate_user_scrubs_credential_from_structured_reply_ok_protocol_variant() {
    // OK protocol envelope + negative hresult: already Error::MxAccess before
    // ISSUE 2, but the stored reply's string fields still leaked the credential.
    assert_authenticate_user_scrubs_structured_reply(
        "authenticate-user.echoed-credential.reply.json",
    )
    .await;
}

#[tokio::test]
async fn authenticate_user_scrubs_credential_from_structured_reply_mxaccess_failure_variant() {
    // PROTOCOL_STATUS_CODE_MXACCESS_FAILURE: before ISSUE 2 this landed in
    // Error::Command (unscrubbed, raw Display/Debug) — the red-first case.
    assert_authenticate_user_scrubs_structured_reply(
        "authenticate-user.echoed-credential-mxaccess-failure.reply.json",
    )
    .await;
}

#[tokio::test]
async fn authenticate_user_maps_missing_payload_reply_to_malformed_reply() {
    // CLI-41: an OK reply with neither a typed AuthenticateUser payload nor a
    // return_value is malformed.
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::CannedReply(Box::new(
        command_reply_fixture("authenticate-user.missing-payload.reply.json"),
    )));
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let error = session
        .authenticate_user(7, "verifier", "pw")
        .await
        .unwrap_err();

    assert!(
        matches!(error, Error::MalformedReply { .. }),
        "missing payload + missing return_value must be MalformedReply, got {error:?}"
    );
}

#[tokio::test]
async fn authenticate_user_falls_back_to_return_value_when_typed_payload_absent() {
    // CLI-41: an OK reply that carries only a return_value (legacy worker) must
    // resolve the user id from it, mirroring add_buffered_item's fallback.
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::CannedReply(Box::new(
        command_reply_fixture("authenticate-user.return-value-only.reply.json"),
    )));
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    let user_id = session
        .authenticate_user(7, "verifier", "pw")
        .await
        .unwrap();

    assert_eq!(
        user_id, 7,
        "user id must resolve from the int32 return_value"
    );
}

#[tokio::test]
async fn stream_alarms_emits_snapshot_then_complete_then_transition_in_order() {
    let state = Arc::new(FakeState::default());
    *state.stream_alarms_script.lock().await = Some(vec![
        AlarmFeedMessage {
            payload: Some(alarm_feed_message::Payload::ActiveAlarm(
                ActiveAlarmSnapshot {
                    alarm_full_reference: "Tank01.Level.HiHi".to_owned(),
                    current_state: AlarmConditionState::Active as i32,
                    ..ActiveAlarmSnapshot::default()
                },
            )),
        },
        AlarmFeedMessage {
            payload: Some(alarm_feed_message::Payload::SnapshotComplete(true)),
        },
        AlarmFeedMessage {
            payload: Some(alarm_feed_message::Payload::Transition(
                OnAlarmTransitionEvent {
                    alarm_full_reference: "Tank01.Level.HiHi".to_owned(),
                    transition_kind: AlarmTransitionKind::Raise as i32,
                    ..OnAlarmTransitionEvent::default()
                },
            )),
        },
    ]);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();

    let mut stream = client
        .stream_alarms(StreamAlarmsRequest {
            client_correlation_id: next_correlation_id("test-stream-alarms"),
            alarm_filter_prefix: String::new(),
        })
        .await
        .unwrap();

    let first = stream.next().await.unwrap().unwrap();
    let second = stream.next().await.unwrap().unwrap();
    let third = stream.next().await.unwrap().unwrap();

    assert!(matches!(
        first.payload,
        Some(alarm_feed_message::Payload::ActiveAlarm(_))
    ));
    assert!(matches!(
        second.payload,
        Some(alarm_feed_message::Payload::SnapshotComplete(true))
    ));
    assert!(matches!(
        third.payload,
        Some(alarm_feed_message::Payload::Transition(_))
    ));
}

#[tokio::test]
async fn cli_subcommands_propagate_unique_correlation_ids_from_next_correlation_id() {
    // The CLI's `stream-alarms` and `acknowledge-alarm` paths used to
    // hard-code their correlation ids (Client.Rust-023). Verify the
    // resolution end-to-end through `next_correlation_id`: every call
    // observed at the fake gateway has a unique id that embeds the
    // `cli-...` label, so concurrent CLI smokes can tell collisions apart.
    let state = Arc::new(FakeState::default());
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();

    let first_corr = next_correlation_id("cli-stream-alarms");
    let _ = client
        .stream_alarms(StreamAlarmsRequest {
            client_correlation_id: first_corr.clone(),
            alarm_filter_prefix: String::new(),
        })
        .await
        .unwrap();
    assert_eq!(
        *state.last_correlation_id.lock().await,
        Some(first_corr.clone())
    );

    let second_corr = next_correlation_id("cli-stream-alarms");
    assert_ne!(first_corr, second_corr);
    assert!(second_corr.contains("cli-stream-alarms"));

    let third_corr = next_correlation_id("cli-acknowledge-alarm");
    let _ = client
        .acknowledge_alarm(AcknowledgeAlarmRequest {
            client_correlation_id: third_corr.clone(),
            alarm_full_reference: "Tank01.Level.HiHi".to_owned(),
            comment: String::new(),
            operator_user: String::new(),
        })
        .await
        .unwrap();
    assert_eq!(*state.last_correlation_id.lock().await, Some(third_corr));
}

#[derive(Default)]
struct FakeState {
    authorization: Mutex<Option<String>>,
    last_command_kind: Mutex<Option<i32>>,
    last_correlation_id: Mutex<Option<String>>,
    /// Captures the last `WriteCommand` payload received, populated when the
    /// `WriteOk` override is active. Used by `write_array_elements` e2e test.
    last_write_command: Mutex<Option<WriteCommand>>,
    /// Captures the last `AuthenticateUserCommand` payload received, populated
    /// by the `AuthenticateUser` happy-path handler so a test can confirm the
    /// credential reaches the wire (but never a surfaced error).
    last_authenticate_user: Mutex<Option<AuthenticateUserCommand>>,
    stream_dropped: Arc<AtomicBool>,
    /// Optional per-test override that pins the fake's `Invoke` handler to
    /// a specific reply shape (or `Err(Status)`). The default of `None`
    /// keeps the existing happy-path dispatcher.
    invoke_override: Mutex<Option<InvokeOverride>>,
    /// Optional per-test override that pins the fake's `StreamAlarms`
    /// handler to emit a synthetic ConditionRefresh -> snapshot_complete
    /// -> transition sequence.
    stream_alarms_script: Mutex<Option<Vec<AlarmFeedMessage>>>,
    /// Optional per-test override that pins the fake's `StreamEvents`
    /// handler to emit a scripted `MxEvent` sequence (e.g. a `replay_gap`
    /// sentinel followed by a normal event). When `None`, the handler falls
    /// back to the default `event(1)` / `event(2)` pair.
    stream_events_script: Mutex<Option<Vec<MxEvent>>>,
}

/// Per-test override for the fake's `Invoke` handler.
#[allow(dead_code)]
enum InvokeOverride {
    /// Reply with `protocol_status = Ok` and no `payload` set.
    OkWithoutPayload,
    /// Reply with `protocol_status = Ok` and a deliberately wrong payload
    /// arm — e.g. an `AddItemReply` body when the caller invoked a bulk
    /// command. The variant carries the kind to recognise in tests but the
    /// reply itself is the mismatched-payload shape.
    OkWithMismatchedPayload,
    /// Fail the unary call with `Status::unavailable(...)` so the client's
    /// `Code::Unavailable` -> `Error::Unavailable` mapping is exercised.
    Unavailable(String),
    /// Accept a `Write` command (return `protocol_status = Ok`, no payload)
    /// and capture the decoded `WriteCommand` in
    /// `FakeState::last_write_command` for inspection.
    WriteOk,
    /// Reply with an `Ok` protocol envelope but a negative `hresult` and a
    /// non-success status entry carrying a credential-shaped diagnostic. This
    /// mimics the worker's COMException path (e.g. `WriteSecured` /
    /// `AuthenticateUser` rejected by MXAccess) so the client's
    /// `ensure_mxaccess_success` check is exercised on the typed helper path.
    MxAccessFailure,
    /// Reply with a caller-supplied canned [`MxCommandReply`]. Lets a test
    /// drive a helper with a shared behavior fixture (e.g. the
    /// echoed-credential / missing-payload / return-value-only
    /// authenticate-user replies). Boxed to keep the enum small.
    CannedReply(Box<MxCommandReply>),
}

#[derive(Clone)]
struct FakeGateway {
    state: Arc<FakeState>,
}

#[tonic::async_trait]
impl MxAccessGateway for FakeGateway {
    async fn open_session(
        &self,
        request: Request<OpenSessionRequest>,
    ) -> Result<Response<OpenSessionReply>, Status> {
        *self.state.authorization.lock().await = request
            .metadata()
            .get("authorization")
            .and_then(|value| value.to_str().ok())
            .map(str::to_owned);

        Ok(Response::new(OpenSessionReply {
            session_id: "session-fixture".to_owned(),
            backend_name: "fake".to_owned(),
            worker_process_id: 1234,
            worker_protocol_version: 1,
            gateway_protocol_version: 1,
            protocol_status: Some(ok_status("opened")),
            ..OpenSessionReply::default()
        }))
    }

    async fn close_session(
        &self,
        request: Request<CloseSessionRequest>,
    ) -> Result<Response<CloseSessionReply>, Status> {
        Ok(Response::new(CloseSessionReply {
            session_id: request.into_inner().session_id,
            final_state: SessionState::Closed as i32,
            protocol_status: Some(ok_status("closed")),
        }))
    }

    async fn invoke(
        &self,
        request: Request<
            zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::MxCommandRequest,
        >,
    ) -> Result<Response<MxCommandReply>, Status> {
        let request = request.into_inner();
        let kind = request
            .command
            .as_ref()
            .map(|command| command.kind)
            .unwrap_or_default();
        *self.state.last_command_kind.lock().await = Some(kind);
        *self.state.last_correlation_id.lock().await = Some(request.client_correlation_id.clone());

        // Honour any per-test override before falling through to the
        // happy-path dispatcher.
        if let Some(override_) = self.state.invoke_override.lock().await.take() {
            return match override_ {
                InvokeOverride::OkWithoutPayload => Ok(Response::new(MxCommandReply {
                    session_id: request.session_id,
                    correlation_id: "fake-correlation".to_owned(),
                    kind,
                    protocol_status: Some(ok_status("command ok")),
                    payload: None,
                    ..MxCommandReply::default()
                })),
                InvokeOverride::OkWithMismatchedPayload => Ok(Response::new(MxCommandReply {
                    session_id: request.session_id,
                    correlation_id: "fake-correlation".to_owned(),
                    kind,
                    protocol_status: Some(ok_status("command ok")),
                    // Deliberately the wrong payload arm — `AddItemReply`
                    // for whatever command was actually invoked.
                    payload: Some(mx_command_reply::Payload::AddItem(AddItemReply {
                        item_handle: 99,
                    })),
                    ..MxCommandReply::default()
                })),
                InvokeOverride::Unavailable(message) => Err(Status::unavailable(message)),
                InvokeOverride::MxAccessFailure => Ok(Response::new(MxCommandReply {
                    session_id: request.session_id,
                    correlation_id: "fake-correlation".to_owned(),
                    kind,
                    // Protocol envelope succeeds; MXAccess itself failed.
                    protocol_status: Some(ok_status("command ok")),
                    // 0x80040E14 (a COM failure) as a signed 32-bit value.
                    hresult: Some(-2_147_217_900),
                    statuses: vec![MxStatusProxy {
                        success: 0,
                        category: MxStatusCategory::SecurityError as i32,
                        detected_by: MxStatusSource::RespondingLmx as i32,
                        detail: 123,
                        // A credential-shaped token that must be scrubbed from
                        // any surfaced diagnostic text.
                        diagnostic_text: "denied for mxgw_leaked_secret".to_owned(),
                        ..MxStatusProxy::default()
                    }],
                    payload: None,
                    ..MxCommandReply::default()
                })),
                InvokeOverride::CannedReply(reply) => Ok(Response::new(*reply)),
                InvokeOverride::WriteOk => {
                    // Extract and capture the WriteCommand payload so the test
                    // can assert on server_handle, item_handle, user_id, and value.
                    if let Some(mx_command::Payload::Write(write_cmd)) =
                        request.command.and_then(|c| c.payload)
                    {
                        *self.state.last_write_command.lock().await = Some(write_cmd);
                    }
                    Ok(Response::new(MxCommandReply {
                        session_id: request.session_id,
                        correlation_id: "fake-correlation".to_owned(),
                        kind,
                        protocol_status: Some(ok_status("write ok")),
                        payload: None,
                        ..MxCommandReply::default()
                    }))
                }
            };
        }

        if kind == MxCommandKind::Write as i32 {
            return Ok(Response::new(mxaccess_failure_reply()));
        }

        if kind == MxCommandKind::SubscribeBulk as i32 {
            return Ok(Response::new(MxCommandReply {
                session_id: request.session_id,
                correlation_id: "fake-correlation".to_owned(),
                kind,
                protocol_status: Some(ok_status("command ok")),
                payload: Some(mx_command_reply::Payload::SubscribeBulk(
                    BulkSubscribeReply {
                        results: vec![SubscribeResult {
                            server_handle: 12,
                            tag_address: "Area001.Pump001.Speed".to_owned(),
                            item_handle: 34,
                            was_successful: true,
                            error_message: String::new(),
                        }],
                    },
                )),
                ..MxCommandReply::default()
            }));
        }

        if kind == MxCommandKind::Register as i32 {
            return Ok(Response::new(MxCommandReply {
                session_id: request.session_id,
                correlation_id: "fake-correlation".to_owned(),
                kind,
                protocol_status: Some(ok_status("command ok")),
                payload: Some(mx_command_reply::Payload::Register(RegisterReply {
                    server_handle: 12,
                })),
                ..MxCommandReply::default()
            }));
        }

        if kind == MxCommandKind::AddItem2 as i32 {
            return Ok(Response::new(MxCommandReply {
                session_id: request.session_id,
                correlation_id: "fake-correlation".to_owned(),
                kind,
                protocol_status: Some(ok_status("command ok")),
                payload: Some(mx_command_reply::Payload::AddItem2(AddItem2Reply {
                    item_handle: 56,
                })),
                ..MxCommandReply::default()
            }));
        }

        if kind == MxCommandKind::ReadBulk as i32 {
            return Ok(Response::new(MxCommandReply {
                session_id: request.session_id,
                correlation_id: "fake-correlation".to_owned(),
                kind,
                protocol_status: Some(ok_status("command ok")),
                payload: Some(mx_command_reply::Payload::ReadBulk(BulkReadReply {
                    results: vec![BulkReadResult {
                        server_handle: 12,
                        tag_address: "Area001.Pump001.Speed".to_owned(),
                        item_handle: 34,
                        was_successful: true,
                        was_cached: true,
                        ..BulkReadResult::default()
                    }],
                })),
                ..MxCommandReply::default()
            }));
        }

        if kind == MxCommandKind::WriteBulk as i32 {
            return Ok(Response::new(write_bulk_reply_for(
                request.session_id,
                kind,
                mx_command_reply::Payload::WriteBulk(BulkWriteReply {
                    results: vec![bulk_write_result_ok(12, 34)],
                }),
            )));
        }

        if kind == MxCommandKind::Write2Bulk as i32 {
            return Ok(Response::new(write_bulk_reply_for(
                request.session_id,
                kind,
                mx_command_reply::Payload::Write2Bulk(BulkWriteReply {
                    results: vec![bulk_write_result_ok(12, 34)],
                }),
            )));
        }

        if kind == MxCommandKind::WriteSecuredBulk as i32 {
            return Ok(Response::new(write_bulk_reply_for(
                request.session_id,
                kind,
                mx_command_reply::Payload::WriteSecuredBulk(BulkWriteReply {
                    results: vec![bulk_write_result_ok(12, 34)],
                }),
            )));
        }

        if kind == MxCommandKind::WriteSecured2Bulk as i32 {
            return Ok(Response::new(write_bulk_reply_for(
                request.session_id,
                kind,
                mx_command_reply::Payload::WriteSecured2Bulk(BulkWriteReply {
                    results: vec![bulk_write_result_ok(12, 34)],
                }),
            )));
        }

        if kind == MxCommandKind::AuthenticateUser as i32 {
            // Capture the transmitted command so a test can confirm the
            // credential reaches the wire but never an error message.
            if let Some(mx_command::Payload::AuthenticateUser(auth)) =
                request.command.and_then(|command| command.payload)
            {
                *self.state.last_authenticate_user.lock().await = Some(auth);
            }
            return Ok(Response::new(MxCommandReply {
                session_id: request.session_id,
                correlation_id: "fake-correlation".to_owned(),
                kind,
                protocol_status: Some(ok_status("command ok")),
                payload: Some(mx_command_reply::Payload::AuthenticateUser(
                    AuthenticateUserReply { user_id: 4242 },
                )),
                ..MxCommandReply::default()
            }));
        }

        Ok(Response::new(MxCommandReply {
            session_id: request.session_id,
            correlation_id: "fake-correlation".to_owned(),
            kind,
            protocol_status: Some(ok_status("command ok")),
            payload: Some(mx_command_reply::Payload::AddItem(AddItemReply {
                item_handle: 34,
            })),
            ..MxCommandReply::default()
        }))
    }

    type StreamEventsStream = DropAwareStream;

    async fn stream_events(
        &self,
        _request: Request<StreamEventsRequest>,
    ) -> Result<Response<Self::StreamEventsStream>, Status> {
        let script = self.state.stream_events_script.lock().await.take();
        let events = script.unwrap_or_else(|| vec![event(1), event(2)]);
        let (sender, receiver) = mpsc::channel(events.len().max(1));
        for event in events {
            sender.send(Ok(event)).await.unwrap();
        }

        Ok(Response::new(DropAwareStream {
            inner: ReceiverStream::new(receiver),
            dropped: self.state.stream_dropped.clone(),
        }))
    }

    async fn acknowledge_alarm(
        &self,
        request: Request<AcknowledgeAlarmRequest>,
    ) -> Result<Response<AcknowledgeAlarmReply>, Status> {
        *self.state.last_correlation_id.lock().await =
            Some(request.into_inner().client_correlation_id);
        Ok(Response::new(AcknowledgeAlarmReply {
            correlation_id: "corr-1".to_owned(),
            protocol_status: Some(ok_status("ack ok")),
            status: Some(MxStatusProxy {
                success: 1,
                category: MxStatusCategory::Ok as i32,
                detected_by: MxStatusSource::RespondingLmx as i32,
                ..MxStatusProxy::default()
            }),
            ..AcknowledgeAlarmReply::default()
        }))
    }

    type StreamAlarmsStream =
        Pin<Box<dyn Stream<Item = Result<AlarmFeedMessage, Status>> + Send + 'static>>;

    async fn stream_alarms(
        &self,
        request: Request<StreamAlarmsRequest>,
    ) -> Result<Response<Self::StreamAlarmsStream>, Status> {
        *self.state.last_correlation_id.lock().await =
            Some(request.into_inner().client_correlation_id);
        let script = self.state.stream_alarms_script.lock().await.take();
        let (sender, receiver) =
            mpsc::channel::<Result<AlarmFeedMessage, Status>>(script.as_ref().map_or(1, Vec::len));
        if let Some(messages) = script {
            for message in messages {
                sender.send(Ok(message)).await.unwrap();
            }
        }
        let stream = ReceiverStream::new(receiver);
        Ok(Response::new(Box::pin(stream)))
    }

    type QueryActiveAlarmsStream =
        Pin<Box<dyn Stream<Item = Result<ActiveAlarmSnapshot, Status>> + Send + 'static>>;

    async fn query_active_alarms(
        &self,
        _request: Request<QueryActiveAlarmsRequest>,
    ) -> Result<Response<Self::QueryActiveAlarmsStream>, Status> {
        let (sender, receiver) = mpsc::channel(4);
        sender
            .send(Ok(ActiveAlarmSnapshot {
                alarm_full_reference: "Tank01.Level.HiHi".to_owned(),
                ..ActiveAlarmSnapshot::default()
            }))
            .await
            .unwrap();
        let stream = ReceiverStream::new(receiver);
        Ok(Response::new(Box::pin(stream)))
    }
}

struct DropAwareStream {
    inner: ReceiverStream<Result<MxEvent, Status>>,
    dropped: Arc<AtomicBool>,
}

impl Stream for DropAwareStream {
    type Item = Result<MxEvent, Status>;

    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        Pin::new(&mut self.inner).poll_next(context)
    }
}

impl Drop for DropAwareStream {
    fn drop(&mut self) {
        self.dropped.store(true, Ordering::SeqCst);
    }
}

async fn spawn_fake_gateway(state: Arc<FakeState>) -> String {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap();
    let incoming = TcpListenerStream::new(listener);
    let service = MxAccessGatewayServer::new(FakeGateway { state });
    tokio::spawn(async move {
        Server::builder()
            .add_service(service)
            .serve_with_incoming(incoming)
            .await
            .unwrap();
    });

    format!("http://{address}")
}

fn write_bulk_reply_for(
    session_id: String,
    kind: i32,
    payload: mx_command_reply::Payload,
) -> MxCommandReply {
    MxCommandReply {
        session_id,
        correlation_id: "fake-correlation".to_owned(),
        kind,
        protocol_status: Some(ok_status("command ok")),
        payload: Some(payload),
        ..MxCommandReply::default()
    }
}

fn bulk_write_result_ok(server_handle: i32, item_handle: i32) -> BulkWriteResult {
    BulkWriteResult {
        server_handle,
        item_handle,
        was_successful: true,
        hresult: Some(0),
        statuses: Vec::new(),
        error_message: String::new(),
    }
}

fn ok_status(message: &str) -> ProtocolStatus {
    ProtocolStatus {
        code: ProtocolStatusCode::Ok as i32,
        message: message.to_owned(),
    }
}

fn mxaccess_failure_reply() -> MxCommandReply {
    MxCommandReply {
        session_id: "session-fixture".to_owned(),
        correlation_id: "gateway-correlation-write-1".to_owned(),
        kind: MxCommandKind::Write as i32,
        protocol_status: Some(ProtocolStatus {
            code: ProtocolStatusCode::MxaccessFailure as i32,
            message: "MXAccess rejected the write.".to_owned(),
        }),
        hresult: Some(-2147220992),
        statuses: vec![
            MxStatusProxy {
                success: 0,
                category: MxStatusCategory::SecurityError as i32,
                detected_by: MxStatusSource::RespondingLmx as i32,
                detail: 321,
                raw_category: 8,
                raw_detected_by: 3,
                diagnostic_text: "Write denied by provider security.".to_owned(),
            },
            MxStatusProxy {
                success: 0,
                category: MxStatusCategory::OperationalError as i32,
                detected_by: MxStatusSource::RespondingNmx as i32,
                detail: 902,
                raw_category: 7,
                raw_detected_by: 5,
                diagnostic_text: "Provider rejected the item state.".to_owned(),
            },
        ],
        ..MxCommandReply::default()
    }
}

fn event(sequence: u64) -> MxEvent {
    MxEvent {
        family: MxEventFamily::OnDataChange as i32,
        session_id: "session-fixture".to_owned(),
        worker_sequence: sequence,
        ..MxEvent::default()
    }
}

/// Load a shared command-reply fixture into an [`MxCommandReply`].
///
/// The fixtures are protobuf JSON, which prost cannot parse directly, so this
/// reads the fields the reply-validation rules actually consume (`hresult` and
/// the status `success`/`category` pair) and rebuilds the message. Enum names
/// resolve through the generated `from_str_name`, so a fixture naming a
/// category the contract does not define fails the test rather than silently
/// degrading to `Unspecified`.
fn command_reply_fixture(file_name: &str) -> MxCommandReply {
    let fixture = behavior_fixture(&format!("command-replies/{file_name}"));

    let statuses = fixture["statuses"]
        .as_array()
        .map(Vec::as_slice)
        .unwrap_or_default()
        .iter()
        .map(|status| {
            let category_name = status["category"].as_str().unwrap();
            MxStatusProxy {
                success: status["success"].as_i64().unwrap() as i32,
                category: MxStatusCategory::from_str_name(category_name)
                    .unwrap_or_else(|| panic!("unknown status category {category_name}"))
                    as i32,
                detail: status["detail"].as_i64().unwrap_or_default() as i32,
                diagnostic_text: status["diagnosticText"]
                    .as_str()
                    .unwrap_or_default()
                    .to_owned(),
                ..MxStatusProxy::default()
            }
        })
        .collect();

    // The fixtures that exercise the return_value fallback path carry a typed
    // `returnValue` (VT_I4). Project it so a canned reply can drive the
    // helper's payload -> return_value -> MalformedReply precedence.
    let return_value = fixture.get("returnValue").and_then(|value| {
        value["int32Value"].as_i64().map(|int32| MxValue {
            data_type: MxDataType::Integer as i32,
            variant_type: value["variantType"].as_str().unwrap_or("VT_I4").to_owned(),
            kind: Some(Kind::Int32Value(int32 as i32)),
            ..MxValue::default()
        })
    });

    // Honor the fixture's real protocol status (code + message) so a canned
    // reply can drive the MXACCESS_FAILURE routing path, not just an OK
    // envelope. Falls back to an OK envelope when the fixture omits it.
    let protocol_status = fixture.get("protocolStatus").map_or_else(
        || ok_status("command ok"),
        |status| {
            let code_name = status["code"].as_str().unwrap_or("PROTOCOL_STATUS_CODE_OK");
            ProtocolStatus {
                code: ProtocolStatusCode::from_str_name(code_name)
                    .unwrap_or_else(|| panic!("unknown protocol status code {code_name}"))
                    as i32,
                message: status["message"].as_str().unwrap_or_default().to_owned(),
            }
        },
    );

    MxCommandReply {
        session_id: fixture["sessionId"].as_str().unwrap_or_default().to_owned(),
        correlation_id: fixture["correlationId"]
            .as_str()
            .unwrap_or_default()
            .to_owned(),
        protocol_status: Some(protocol_status),
        hresult: fixture["hresult"].as_i64().map(|hresult| hresult as i32),
        statuses,
        diagnostic_message: fixture["diagnosticMessage"]
            .as_str()
            .unwrap_or_default()
            .to_owned(),
        return_value,
        ..MxCommandReply::default()
    }
}

fn behavior_fixture(path: &str) -> Value {
    let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("../proto/fixtures/behavior")
        .join(path);
    let data = std::fs::read_to_string(&path).unwrap();
    serde_json::from_str(&data).unwrap()
}

fn case_by_id<'a>(cases: &'a [Value], id: &str) -> &'a Value {
    cases
        .iter()
        .find(|case| case["id"].as_str() == Some(id))
        .unwrap_or_else(|| panic!("missing fixture case {id}"))
}

// ---------------------------------------------------------------------------
// write_array_elements — end-to-end fake-server test
// ---------------------------------------------------------------------------

#[tokio::test]
async fn write_array_elements_routes_sparse_array_write_through_fake_gateway() {
    // Arrange: stand up the fake gateway with WriteOk so the Write command
    // succeeds and the sent WriteCommand is captured for inspection.
    let state = Arc::new(FakeState::default());
    *state.invoke_override.lock().await = Some(InvokeOverride::WriteOk);
    let endpoint = spawn_fake_gateway(state.clone()).await;
    let client = GatewayClient::connect(ClientOptions::new(endpoint))
        .await
        .unwrap();
    let session = client.session("session-fixture");

    // Act: call the public write_array_elements helper.
    session
        .write_array_elements(
            12,
            34,
            MxDataType::Integer,
            10,
            [(2u32, ClientMxValue::int32(42))],
            7,
        )
        .await
        .unwrap();

    // Assert: the fake captured a Write command with the expected handles and
    // a SparseArrayValue whose total_length and element index/value are correct.
    let captured = state
        .last_write_command
        .lock()
        .await
        .take()
        .expect("fake should have captured a WriteCommand");

    assert_eq!(captured.server_handle, 12, "server_handle must round-trip");
    assert_eq!(captured.item_handle, 34, "item_handle must round-trip");
    assert_eq!(captured.user_id, 7, "user_id must round-trip");

    let value = captured.value.expect("WriteCommand must carry a value");
    assert_eq!(
        value.data_type, 0,
        "outer MxValue.data_type must be Unspecified (0), not the element type"
    );

    let Kind::SparseArrayValue(ref sparse) = value.kind.as_ref().unwrap() else {
        panic!(
            "expected SparseArrayValue kind on the outer MxValue, got {:?}",
            value.kind
        );
    };
    assert_eq!(
        sparse.element_data_type,
        MxDataType::Integer as i32,
        "element_data_type must carry the element type"
    );
    assert_eq!(sparse.total_length, 10, "total_length must round-trip");
    assert_eq!(sparse.elements.len(), 1, "one element supplied");

    let elem = &sparse.elements[0];
    assert_eq!(elem.index, 2, "element index must round-trip");
    assert_eq!(
        elem.value.as_ref().unwrap().kind,
        Some(Kind::Int32Value(42)),
        "element value must round-trip"
    );
}

// ---------------------------------------------------------------------------
// write_array_elements — proto shape unit tests
// ---------------------------------------------------------------------------

/// Build the proto `MxValue` that `write_array_elements` would send and assert
/// the sparse oneof variant has the correct `total_length` and elements.
fn sparse_int32_value(
    total_length: u32,
    elements: impl IntoIterator<Item = (u32, i32)>,
) -> MxValue {
    let sparse_elements: Vec<MxSparseElement> = elements
        .into_iter()
        .map(|(index, v)| MxSparseElement {
            index,
            value: Some(MxValue {
                data_type: MxDataType::Integer as i32,
                variant_type: "VT_I4".to_owned(),
                kind: Some(Kind::Int32Value(v)),
                ..MxValue::default()
            }),
        })
        .collect();

    MxValue {
        // outer data_type must be 0 (Unspecified); the element type lives only
        // inside MxSparseArray.element_data_type, matching the
        // `..ProtoMxValue::default()` used in Session::write_array_elements.
        variant_type: String::new(),
        kind: Some(Kind::SparseArrayValue(MxSparseArray {
            element_data_type: MxDataType::Integer as i32,
            total_length,
            elements: sparse_elements,
        })),
        ..MxValue::default()
    }
}

#[test]
fn write_array_elements_proto_shape_has_sparse_oneof_kind() {
    let proto = sparse_int32_value(5, [(0, 10), (3, 30)]);

    assert_eq!(
        proto.data_type, 0,
        "outer MxValue.data_type must be 0 (Unspecified); element type lives in element_data_type"
    );

    let Kind::SparseArrayValue(ref sparse) = proto.kind.as_ref().unwrap() else {
        panic!("expected SparseArrayValue kind, got {:?}", proto.kind);
    };

    assert_eq!(sparse.total_length, 5, "total_length must round-trip");
    assert_eq!(sparse.elements.len(), 2, "two elements supplied");
    assert_eq!(sparse.element_data_type, MxDataType::Integer as i32);

    let elem0 = &sparse.elements[0];
    assert_eq!(elem0.index, 0);
    assert_eq!(
        elem0.value.as_ref().unwrap().kind,
        Some(Kind::Int32Value(10))
    );

    let elem3 = &sparse.elements[1];
    assert_eq!(elem3.index, 3);
    assert_eq!(
        elem3.value.as_ref().unwrap().kind,
        Some(Kind::Int32Value(30))
    );
}

#[test]
fn write_array_elements_empty_elements_is_valid_all_defaults() {
    let proto = sparse_int32_value(8, []);
    assert_eq!(
        proto.data_type, 0,
        "outer MxValue.data_type must be 0 (Unspecified) even with no elements"
    );
    let Kind::SparseArrayValue(ref sparse) = proto.kind.as_ref().unwrap() else {
        panic!("expected SparseArrayValue kind");
    };
    assert_eq!(sparse.total_length, 8);
    assert!(
        sparse.elements.is_empty(),
        "no elements means every index defaults"
    );
}

#[test]
fn sparse_array_value_round_trips_through_client_mx_value_projection_as_unset() {
    // SparseArrayValue is write-only. If it ever arrives on the read path
    // (e.g. a future version bug), the projection should degrade to Unset
    // rather than panic, because the enum variant is not readable.
    let proto = sparse_int32_value(4, [(1, 99)]);
    assert_eq!(
        proto.data_type, 0,
        "outer MxValue.data_type must be 0 (Unspecified)"
    );
    let client_value = ClientMxValue::from_proto(proto);
    assert_eq!(
        client_value.projection(),
        &MxValueProjection::Unset,
        "write-only SparseArrayValue must project to Unset, not panic"
    );
}