use super::*;
use crate::test_support::{env::ScopedEnv, secrets::random_test_secret};
use std::error::Error as _;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use tempfile::tempdir;

fn test_integrity_value() -> String {
    random_test_secret(32)
}

#[test]
fn test_try_new_reports_activity_service_startup_error() {
    let err = WsServerState::try_new_with_activity_service_factory_for_test(
        WsServerConfig::default(),
        || {
            Err(channels::activity::ActivityStartupError::ThreadSpawn {
                source: crate::StartupThreadSpawnError::new(
                    "activity-test-thread",
                    std::io::Error::other("simulated activity thread exhaustion"),
                ),
            })
        },
    )
    .expect_err("startup should report activity service initialization failure");

    match err {
        WsConfigError::ActivityStartup(error) => {
            assert!(error
                .to_string()
                .contains("simulated activity thread exhaustion"));
        }
        other => panic!("expected typed activity startup error, got {other:?}"),
    }
}

#[test]
fn test_try_new_persistent_unloaded_reports_activity_service_startup_error() {
    let temp = tempdir().expect("tempdir");
    let err = WsServerState::try_new_persistent_unloaded_with_activity_service_factory_for_test(
        WsServerConfig::default(),
        temp.path().to_path_buf(),
        |_state_dir| {
            Err(channels::activity::ActivityStartupError::ThreadSpawn {
                source: crate::StartupThreadSpawnError::new(
                    "activity-persistent-test-thread",
                    std::io::Error::other("simulated persistent activity thread exhaustion"),
                ),
            })
        },
    )
    .expect_err("persistent startup should report activity service initialization failure");

    match err {
        WsConfigError::ActivityStartup(error) => {
            assert!(error
                .to_string()
                .contains("simulated persistent activity thread exhaustion"));
        }
        other => panic!("expected typed activity startup error, got {other:?}"),
    }
}

#[test]
fn test_try_new_preserves_activity_startup_error_chain() {
    let err = WsServerState::try_new_with_activity_service_factory_for_test(
        WsServerConfig::default(),
        || {
            Err(channels::activity::ActivityStartupError::ThreadSpawn {
                source: crate::StartupThreadSpawnError::new(
                    "activity-chain-test-thread",
                    std::io::Error::from_raw_os_error(11),
                ),
            })
        },
    )
    .expect_err("startup should preserve activity service error chain");

    let activity_source = err
        .source()
        .expect("ws startup error should expose activity startup source");
    let activity_error = activity_source
        .downcast_ref::<channels::activity::ActivityStartupError>()
        .expect("activity startup error should remain in the source chain");
    let spawn_source = activity_error
        .source()
        .expect("activity startup error should expose startup thread spawn source");
    let spawn_error = spawn_source
        .downcast_ref::<crate::StartupThreadSpawnError>()
        .expect("startup thread spawn error should remain in the source chain");
    let io_source = spawn_error
        .source()
        .expect("startup thread spawn error should preserve the original io::Error source");
    let io_error = io_source
        .downcast_ref::<std::io::Error>()
        .expect("original io::Error should remain at the end of the source chain");

    assert_eq!(spawn_error.thread_name(), "activity-chain-test-thread");
    assert_eq!(io_error.raw_os_error(), Some(11));
}

#[test]
fn test_try_new_persistent_unloaded_builds_registries_before_activity_service() {
    let temp = tempdir().expect("tempdir");
    let invalid_state_dir = temp.path().join("state");
    std::fs::create_dir_all(&invalid_state_dir).expect("create state dir");
    std::fs::write(
        invalid_state_dir.join("node-pairing.json"),
        "{not valid json",
    )
    .expect("write invalid node pairing state");
    let activity_factory_called = Arc::new(AtomicBool::new(false));
    let activity_factory_called_for_closure = Arc::clone(&activity_factory_called);

    let err = WsServerState::try_new_persistent_unloaded_with_activity_service_factory_for_test(
        WsServerConfig::default(),
        invalid_state_dir,
        move |_state_dir| {
            activity_factory_called_for_closure.store(true, Ordering::SeqCst);
            Err(channels::activity::ActivityStartupError::ThreadSpawn {
                source: crate::StartupThreadSpawnError::new(
                    "activity-ordering-test-thread",
                    std::io::Error::other(
                        "activity service factory should not run before registry setup",
                    ),
                ),
            })
        },
    )
    .expect_err("persistent startup should fail on registry initialization first");

    assert!(
        matches!(err, WsConfigError::Nodes(_)),
        "expected node registry initialization failure, got {err:?}"
    );
    assert!(
        !activity_factory_called.load(Ordering::SeqCst),
        "activity service factory should not run when persistent registry initialization fails"
    );
}

#[test]
fn test_error_shape() {
    let err = error_shape(ERROR_INVALID_REQUEST, "test error", None);
    assert_eq!(err.code, "invalid_request");
    assert_eq!(err.message, "test error");
    assert!(!err.retryable);

    let err2 = error_shape(ERROR_UNAVAILABLE, "temp error", Some(json!({"foo": "bar"})));
    assert_eq!(err2.code, "unavailable");
    assert!(err2.retryable);
    assert!(err2.details.is_some());
}

/// `rate_limited` must surface `retryable: true` on the wire.
#[test]
fn test_error_shape_rate_limited_is_retryable() {
    let err = error_shape(ERROR_RATE_LIMITED, "rate limit exceeded", None);
    assert_eq!(err.code, "rate_limited");
    assert!(
        err.retryable,
        "rate_limited must be retryable per RETRYABLE_CODES"
    );
}

/// COMPAT: rate-limit emits a CORRELATED `res`-frame on rejection,
/// not an asynchronous event. Pre-Batch-48 the rate-limit was a
/// pre-decode check that emitted `error.rateLimited` event without
/// a request id, hanging any released client awaiting a response
/// by id. Master's wire-shape (`send_response(tx, req_id, false,
/// None, Some(err))`) is restored: rate-limit fires AFTER decode,
/// `req_id` is in scope, and the awaiting promise resolves with a
/// retryable error.
#[test]
fn test_ws_rate_limit_emits_correlated_res_frame() {
    let (tx, mut rx) = mpsc::channel(8);
    let tx = ConnectionTx::Raw(tx);
    let mut limiter = crate::server::ratelimit::WsRateLimiter::new(0.0, 0.0);
    let mut warn_count = 0;

    assert!(matches!(
        check_rate_limit(&tx, "req-42", &mut limiter, &mut warn_count),
        Err(LoopSignal::Continue)
    ));
    assert_eq!(warn_count, 1);
    let frame = rx
        .try_recv()
        .expect("rate limiting should emit a correlated res-frame");
    let Message::Text(text) = frame else {
        panic!("expected text rate-limit res-frame");
    };
    let parsed: serde_json::Value = serde_json::from_str(text.as_str()).unwrap();
    assert_eq!(parsed["type"], "res");
    assert_eq!(parsed["id"], "req-42");
    assert_eq!(parsed["ok"], false);
    assert_eq!(parsed["error"]["code"], "rate_limited");
}

/// Configuration-error codes (`unknown_route`, `missing_model`) must
/// surface `retryable: false`. Pinning both rules out a future change
/// to `RETRYABLE_CODES` that mis-classifies domain codes as retryable.
#[test]
fn test_error_shape_config_errors_are_not_retryable() {
    let err = error_shape("unknown_route", "requested route is not configured", None);
    assert_eq!(err.code, "unknown_route");
    assert!(!err.retryable);

    let err2 = error_shape("missing_model", "agent model is not configured", None);
    assert_eq!(err2.code, "missing_model");
    assert!(!err2.retryable);
}

#[test]
fn test_connect_params_validate_optional_client_metadata_types() {
    let base = json!({
        "minProtocol": 3,
        "maxProtocol": 3,
        "client": {
            "id": "webchat-ui",
            "version": "test",
            "platform": "web",
            "mode": "control"
        }
    });

    let mut valid = base.clone();
    valid["locale"] = json!("en-US");
    valid["userAgent"] = json!("carapace-test");
    serde_json::from_value::<ConnectParams>(valid)
        .expect("string locale/userAgent should remain accepted");

    let mut invalid_locale = base.clone();
    invalid_locale["locale"] = json!({});
    assert!(
        serde_json::from_value::<ConnectParams>(invalid_locale).is_err(),
        "locale must remain type-validated when present"
    );

    let mut invalid_user_agent = base;
    invalid_user_agent["userAgent"] = json!({});
    assert!(
        serde_json::from_value::<ConnectParams>(invalid_user_agent).is_err(),
        "userAgent must remain type-validated when present"
    );
}

#[test]
fn test_connect_params_accept_unknown_public_handshake_fields() {
    let base = json!({
        "minProtocol": 3,
        "maxProtocol": 3,
        "client": {
            "id": "webchat-ui",
            "version": "test",
            "platform": "web",
            "mode": "control"
        },
        "device": {
            "id": "device-1",
            "publicKey": "public-key",
            "signature": "signature",
            "signedAt": 1700000000
        },
        "auth": {
            "token": "test-token"
        }
    });

    let mut top_level = base.clone();
    top_level["futureField"] = json!(true);
    assert!(
        serde_json::from_value::<ConnectParams>(top_level).is_ok(),
        "released ConnectParams must tolerate unknown top-level fields"
    );

    let mut client = base.clone();
    client["client"]["futureField"] = json!(true);
    assert!(
        serde_json::from_value::<ConnectParams>(client).is_ok(),
        "released ClientInfo must tolerate unknown fields"
    );

    let mut device = base.clone();
    device["device"]["futureField"] = json!(true);
    assert!(
        serde_json::from_value::<ConnectParams>(device).is_ok(),
        "released DeviceIdentity must tolerate unknown fields"
    );

    let mut auth = base;
    auth["auth"]["futureField"] = json!(true);
    assert!(
        serde_json::from_value::<ConnectParams>(auth).is_ok(),
        "released AuthParams must tolerate unknown fields"
    );
}

#[tokio::test(flavor = "current_thread")]
async fn test_startup_rejects_plaintext_credential_file() {
    let temp = tempdir().expect("tempdir");
    let state_dir = temp.path().join("state");
    let credentials_dir = state_dir.join("credentials");
    std::fs::create_dir_all(&credentials_dir).expect("create credentials dir");
    let plaintext_path = credentials_dir.join("oauth.json");
    std::fs::write(&plaintext_path, r#"{"access_token":"secret"}"#)
        .expect("write plaintext credential file");

    let mut env = ScopedEnv::new();
    env.set("CARAPACE_STATE_DIR", state_dir.as_os_str());

    let err = build_ws_state_owned_from_value(&json!({}))
        .await
        .expect_err("plaintext credential file should fail startup");
    assert!(matches!(
        err,
        WsConfigError::Credentials(
            crate::credentials::CredentialError::PlaintextCredentialFilesDetected(_)
        )
    ));
    assert!(err.to_string().contains("delete it and re-enroll"));
}

#[tokio::test(flavor = "current_thread")]
async fn test_startup_rejects_corrupt_task_queue_file() {
    let temp = tempdir().expect("tempdir");
    let state_dir = temp.path().join("state");
    let task_queue_path = state_dir.join("tasks").join("queue.json");
    std::fs::create_dir_all(task_queue_path.parent().expect("tasks dir"))
        .expect("create tasks dir");
    std::fs::write(&task_queue_path, "not json").expect("write corrupt task queue file");

    let mut env = ScopedEnv::new();
    env.set("CARAPACE_STATE_DIR", state_dir.as_os_str());

    let err = build_ws_state_owned_from_value(&json!({}))
        .await
        .expect_err("corrupt task queue file should fail startup");
    assert!(matches!(err, WsConfigError::Runtime(_)));
    let message = err.to_string();
    assert!(message.contains(&task_queue_path.display().to_string()));
    assert!(message.contains("A copy was written to"));
    assert!(message.contains("Remove or repair the file"));
}

#[test]
fn test_get_value_at_path() {
    let root = json!({
        "gateway": {
            "port": 8080,
            "auth": {
                "mode": "token"
            }
        }
    });

    assert_eq!(get_value_at_path(&root, "gateway.port"), Some(json!(8080)));
    assert_eq!(
        get_value_at_path(&root, "gateway.auth.mode"),
        Some(json!("token"))
    );
    assert_eq!(get_value_at_path(&root, "gateway.missing"), None);
    assert_eq!(get_value_at_path(&root, "unknown"), None);
}

#[test]
fn test_resolve_session_integrity_config_defaults_to_enabled_warn() {
    let cfg = json!({});
    let integrity = crate::sessions::resolve_session_integrity_config(&cfg)
        .expect("missing sessions.integrity must return defaults, not error");
    assert!(integrity.enabled);
    assert_eq!(
        integrity.action,
        crate::sessions::integrity::IntegrityAction::Warn
    );
}

#[test]
fn test_resolve_session_integrity_config_respects_overrides() {
    let cfg = json!({
        "sessions": {
            "integrity": {
                "enabled": false,
                "action": "reject"
            }
        }
    });
    let integrity = crate::sessions::resolve_session_integrity_config(&cfg)
        .expect("valid sessions.integrity must succeed");
    assert!(!integrity.enabled);
    assert_eq!(
        integrity.action,
        crate::sessions::integrity::IntegrityAction::Reject
    );
}

/// Batch 102: present-but-invalid sessions.integrity must fail
/// closed. The prior behavior silently fell back to default
/// (enabled=true, action=Warn) — bad because an operator typing
/// `action: "rEject"` (meaning the stricter Reject tier) would
/// instead end up on the weaker Warn tier without knowing.
#[test]
fn test_resolve_session_integrity_config_invalid_value_fails_closed() {
    let cfg = json!({
        "sessions": {
            "integrity": {
                "enabled": false,
                "action": "invalid"
            }
        }
    });
    let err = crate::sessions::resolve_session_integrity_config(&cfg)
        .expect_err("invalid sessions.integrity must refuse to fall back");
    assert!(
        err.contains("invalid sessions.integrity") && err.contains("typo"),
        "error must explain the refusal: {err}"
    );
    // Silence the now-unused assertions from the old behavior to keep
    // the test focused on the refusal contract.
    let _ = crate::sessions::integrity::IntegrityAction::Warn;
}

/// Batch 102 companion: present-but-invalid sessions.encryption
/// must fail closed. The prior behavior silently fell back to
/// default (mode=IfPassword) — an operator typing
/// `mode: "Required"` (capital R, meaning enforce encryption) would
/// instead end up on the weaker IfPassword tier, leaving session
/// bytes unencrypted when CARAPACE_CONFIG_PASSWORD is absent.
#[test]
fn test_resolve_session_encryption_config_invalid_value_fails_closed() {
    let cfg = json!({
        "sessions": {
            "encryption": {
                "mode": "Required"
            }
        }
    });
    let err = crate::sessions::resolve_session_encryption_config(&cfg)
        .expect_err("invalid sessions.encryption must refuse to fall back");
    assert!(
        err.contains("invalid sessions.encryption") && err.contains("encryption tier"),
        "error must explain the refusal: {err}"
    );
}

#[test]
fn test_resolve_session_encryption_config_missing_returns_defaults() {
    let cfg = json!({});
    let encryption = crate::sessions::resolve_session_encryption_config(&cfg)
        .expect("missing sessions.encryption must return defaults, not error");
    assert!(matches!(
        encryption.mode,
        crate::sessions::EncryptionMode::IfPassword
    ));
}

#[test]
fn test_resolve_session_integrity_secret_prefers_explicit_server_secret() {
    let mut env_guard = ScopedEnv::new();
    let server_secret = test_integrity_value();
    let cfg = json!({
        "gateway": {
            "auth": {
                "token": test_integrity_value(),
                "password": test_integrity_value()
            }
        }
    });
    env_guard.set("CARAPACE_SERVER_SECRET", &server_secret);
    let (secret, source) = crate::sessions::resolve_session_integrity_secret_from_value(&cfg, None)
        .expect("secret should resolve from explicit server secret");
    assert_eq!(secret, server_secret);
    assert_eq!(source, "CARAPACE_SERVER_SECRET");
}

#[test]
fn test_resolve_session_integrity_secret_falls_back_to_gateway_credentials() {
    let token_value = test_integrity_value();
    let from_token = json!({
        "gateway": {
            "auth": {
                "token": token_value.clone()
            }
        }
    });
    let (secret, source) =
        crate::sessions::resolve_session_integrity_secret_from_value(&from_token, None)
            .expect("secret should resolve from gateway token");
    assert_eq!(secret, token_value);
    assert_eq!(source, "gateway.auth.token");

    let password_value = test_integrity_value();
    let from_password = json!({
        "gateway": {
            "auth": {
                "password": password_value.clone()
            }
        }
    });
    let (secret, source) =
        crate::sessions::resolve_session_integrity_secret_from_value(&from_password, None)
            .expect("secret should resolve from gateway password");
    assert_eq!(secret, password_value);
    assert_eq!(source, "gateway.auth.password");
}

#[test]
fn test_resolve_session_integrity_secret_prefers_gateway_env_over_config() {
    let mut env_guard = ScopedEnv::new();
    let env_token = test_integrity_value();
    env_guard
        .set("CARAPACE_GATEWAY_TOKEN", &env_token)
        .set("CARAPACE_GATEWAY_PASSWORD", test_integrity_value());
    let cfg = json!({
        "gateway": {
            "auth": {
                "token": test_integrity_value(),
                "password": test_integrity_value()
            }
        }
    });

    let (secret, source) = crate::sessions::resolve_session_integrity_secret_from_value(&cfg, None)
        .expect("gateway env token should resolve before config");
    assert_eq!(secret, env_token);
    assert_eq!(source, "CARAPACE_GATEWAY_TOKEN");
}

#[test]
fn test_resolve_session_integrity_secret_ignores_empty_values() {
    let mut env_guard = ScopedEnv::new();
    let cfg = json!({
        "gateway": {
            "auth": {
                "token": "",
                "password": ""
            }
        }
    });
    env_guard.set("CARAPACE_SERVER_SECRET", "");
    let secret = crate::sessions::resolve_session_integrity_secret_from_value(&cfg, None);
    assert!(secret.is_none());
}

#[test]
fn test_resolve_session_integrity_secret_returns_none_when_all_missing() {
    let cfg = json!({
        "gateway": {
            "auth": {}
        }
    });
    let secret = crate::sessions::resolve_session_integrity_secret_from_value(&cfg, None);
    assert!(secret.is_none());
}

#[tokio::test]
async fn test_handle_node_invoke_enforces_allowlist() {
    let state = Arc::new(WsServerState::new(WsServerConfig::default()));
    let (tx, mut rx) = mpsc::channel(256);
    let node_conn = ConnectionContext {
        conn_id: "conn-1".to_string(),
        role: "node".to_string(),
        scopes: vec![],
        client: ClientInfo {
            id: "node-1".to_string(),
            version: "1.0".to_string(),
            platform: "test".to_string(),
            mode: "test".to_string(),
            display_name: None,
            device_family: None,
            model_identifier: None,
            instance_id: None,
        },
        device_id: Some("node-1".to_string()),
    };
    state.register_connection(&node_conn, tx, None);
    {
        let mut registry = state.node_registry.lock();
        registry.register(NodeSession {
            node_id: "node-1".to_string(),
            conn_id: "conn-1".to_string(),
            display_name: None,
            platform: Some("test".to_string()),
            version: Some("1.0".to_string()),
            device_family: None,
            model_identifier: None,
            remote_ip: None,
            caps: vec![],
            commands: HashSet::from(["system.run".to_string()]),
            permissions: None,
            path_env: None,
            connected_at_ms: now_ms(),
        });
    }

    let outcome = state
        .node_pairing
        .request_pairing_with_status(
            "node-1".to_string(),
            None,
            vec!["system.run".to_string()],
            None,
            None,
        )
        .unwrap();
    let _ = state
        .node_pairing
        .approve_request(&outcome.request.request_id)
        .unwrap();

    let node_state = Arc::clone(&state);
    let node_conn = node_conn.clone();
    let responder = tokio::spawn(async move {
        if let Some(Message::Text(text)) = rx.recv().await {
            let value: Value = serde_json::from_str(&text).unwrap();
            let invoke_id = value
                .get("payload")
                .and_then(|v| v.get("id"))
                .and_then(|v| v.as_str())
                .unwrap()
                .to_string();
            let params = json!({
                "id": invoke_id,
                "nodeId": "node-1",
                "ok": true,
                "payload": { "ok": true }
            });
            let _ = handle_node_invoke_result(Some(&params), node_state.as_ref(), &node_conn);
        }
    });

    let ok_params = json!({
        "nodeId": "node-1",
        "command": "system.run",
        "idempotencyKey": "req-1"
    });
    assert!(handle_node_invoke(Some(&ok_params), state.as_ref())
        .await
        .is_ok());
    let _ = responder.await;

    let bad_params = json!({
        "nodeId": "node-1",
        "command": "sms.send",
        "idempotencyKey": "req-2"
    });
    let err = handle_node_invoke(Some(&bad_params), state.as_ref())
        .await
        .unwrap_err();
    assert_eq!(err.code, ERROR_INVALID_REQUEST);
}

#[tokio::test]
async fn test_handle_node_invoke_allows_unmapped_commands() {
    let state = Arc::new(WsServerState::new(WsServerConfig::default()));
    let (tx, mut rx) = mpsc::channel(256);
    let node_conn = ConnectionContext {
        conn_id: "conn-2".to_string(),
        role: "node".to_string(),
        scopes: vec![],
        client: ClientInfo {
            id: "node-2".to_string(),
            version: "1.0".to_string(),
            platform: "test".to_string(),
            mode: "test".to_string(),
            display_name: None,
            device_family: None,
            model_identifier: None,
            instance_id: None,
        },
        device_id: Some("node-2".to_string()),
    };
    state.register_connection(&node_conn, tx, None);
    {
        let mut registry = state.node_registry.lock();
        registry.register(NodeSession {
            node_id: "node-2".to_string(),
            conn_id: "conn-2".to_string(),
            display_name: None,
            platform: Some("test".to_string()),
            version: Some("1.0".to_string()),
            device_family: None,
            model_identifier: None,
            remote_ip: None,
            caps: vec![],
            commands: HashSet::from(["custom.action".to_string()]),
            permissions: None,
            path_env: None,
            connected_at_ms: now_ms(),
        });
    }

    let outcome = state
        .node_pairing
        .request_pairing_with_status(
            "node-2".to_string(),
            None,
            vec!["custom.action".to_string()],
            None,
            None,
        )
        .unwrap();
    let _ = state
        .node_pairing
        .approve_request(&outcome.request.request_id)
        .unwrap();

    let node_state = Arc::clone(&state);
    let node_conn = node_conn.clone();
    let responder = tokio::spawn(async move {
        if let Some(Message::Text(text)) = rx.recv().await {
            let value: Value = serde_json::from_str(&text).unwrap();
            let invoke_id = value
                .get("payload")
                .and_then(|v| v.get("id"))
                .and_then(|v| v.as_str())
                .unwrap()
                .to_string();
            let params = json!({
                "id": invoke_id,
                "nodeId": "node-2",
                "ok": true,
                "payload": { "ok": true }
            });
            let _ = handle_node_invoke_result(Some(&params), node_state.as_ref(), &node_conn);
        }
    });

    let ok_params = json!({
        "nodeId": "node-2",
        "command": "custom.action",
        "idempotencyKey": "req-custom-1"
    });
    assert!(handle_node_invoke(Some(&ok_params), state.as_ref())
        .await
        .is_ok());
    let _ = responder.await;
}

#[tokio::test]
async fn test_handle_node_invoke_allows_missing_paired_permission_key() {
    let state = Arc::new(WsServerState::new(WsServerConfig::default()));
    let (tx, mut rx) = mpsc::channel(256);
    let node_conn = ConnectionContext {
        conn_id: "conn-3".to_string(),
        role: "node".to_string(),
        scopes: vec![],
        client: ClientInfo {
            id: "node-3".to_string(),
            version: "1.0".to_string(),
            platform: "test".to_string(),
            mode: "test".to_string(),
            display_name: None,
            device_family: None,
            model_identifier: None,
            instance_id: None,
        },
        device_id: Some("node-3".to_string()),
    };
    state.register_connection(&node_conn, tx, None);
    {
        let mut registry = state.node_registry.lock();
        registry.register(NodeSession {
            node_id: "node-3".to_string(),
            conn_id: "conn-3".to_string(),
            display_name: None,
            platform: Some("test".to_string()),
            version: Some("1.0".to_string()),
            device_family: None,
            model_identifier: None,
            remote_ip: None,
            caps: vec![],
            commands: HashSet::from(["system.run".to_string()]),
            permissions: None,
            path_env: None,
            connected_at_ms: now_ms(),
        });
    }

    let builder = crate::nodes::NodePairingRequestBuilder {
        node_id: "node-3".to_string(),
        commands: vec!["system.run".to_string()],
        permissions: Some(HashMap::from([("camera".to_string(), true)])),
        ..Default::default()
    };
    let outcome = state
        .node_pairing
        .request_pairing_with_builder(builder)
        .unwrap();
    let request_id = outcome.request.request_id.clone();
    let _ = state.node_pairing.approve_request(&request_id).unwrap();

    let node_state = Arc::clone(&state);
    let node_conn = node_conn.clone();
    let responder = tokio::spawn(async move {
        if let Some(Message::Text(text)) = rx.recv().await {
            let value: Value = serde_json::from_str(&text).unwrap();
            let invoke_id = value
                .get("payload")
                .and_then(|v| v.get("id"))
                .and_then(|v| v.as_str())
                .unwrap()
                .to_string();
            let params = json!({
                "id": invoke_id,
                "nodeId": "node-3",
                "ok": true,
                "payload": { "ok": true }
            });
            let _ = handle_node_invoke_result(Some(&params), node_state.as_ref(), &node_conn);
        }
    });

    let ok_params = json!({
        "nodeId": "node-3",
        "command": "system.run",
        "idempotencyKey": "req-missing-perm-1"
    });
    assert!(handle_node_invoke(Some(&ok_params), state.as_ref())
        .await
        .is_ok());
    let _ = responder.await;
}

#[tokio::test]
async fn test_handle_node_invoke_enforces_permissions() {
    let state = Arc::new(WsServerState::new(WsServerConfig::default()));
    {
        let mut registry = state.node_registry.lock();
        let mut permissions = HashMap::new();
        permissions.insert("camera".to_string(), false);
        permissions.insert("exec".to_string(), true);
        registry.register(NodeSession {
            node_id: "node-perms".to_string(),
            conn_id: "conn-perms".to_string(),
            display_name: None,
            platform: Some("test".to_string()),
            version: Some("1.0".to_string()),
            device_family: None,
            model_identifier: None,
            remote_ip: None,
            caps: vec!["camera".to_string(), "exec".to_string()],
            commands: HashSet::from(["camera.snap".to_string(), "system.run".to_string()]),
            permissions: Some(permissions),
            path_env: None,
            connected_at_ms: now_ms(),
        });
    }

    let params = json!({
        "nodeId": "node-perms",
        "command": "camera.snap",
        "idempotencyKey": "req-perms-1"
    });
    let err = handle_node_invoke(Some(&params), state.as_ref())
        .await
        .unwrap_err();
    assert_eq!(err.code, ERROR_INVALID_REQUEST);
    assert_eq!(err.message, "node permission denied");
}

#[tokio::test]
async fn test_handle_node_invoke_enforces_caps() {
    let state = Arc::new(WsServerState::new(WsServerConfig::default()));
    {
        let mut registry = state.node_registry.lock();
        let mut permissions = HashMap::new();
        permissions.insert("camera".to_string(), true);
        registry.register(NodeSession {
            node_id: "node-caps".to_string(),
            conn_id: "conn-caps".to_string(),
            display_name: None,
            platform: Some("test".to_string()),
            version: Some("1.0".to_string()),
            device_family: None,
            model_identifier: None,
            remote_ip: None,
            caps: vec!["location".to_string()],
            commands: HashSet::from(["camera.snap".to_string()]),
            permissions: Some(permissions),
            path_env: None,
            connected_at_ms: now_ms(),
        });
    }

    let params = json!({
        "nodeId": "node-caps",
        "command": "camera.snap",
        "idempotencyKey": "req-caps-1"
    });
    let err = handle_node_invoke(Some(&params), state.as_ref())
        .await
        .unwrap_err();
    assert_eq!(err.code, ERROR_INVALID_REQUEST);
    assert_eq!(err.message, "node capability missing");
}

#[test]
fn test_normalize_platform_id() {
    assert_eq!(normalize_platform_id(Some("Darwin"), None), "macos");
    assert_eq!(normalize_platform_id(None, Some("iPhone13,3")), "ios");
    assert_eq!(normalize_platform_id(Some("android"), None), "android");
    assert_eq!(normalize_platform_id(None, None), "unknown");
}

#[test]
fn test_resolve_node_command_allowlist() {
    let allow = vec!["custom.command".to_string()];
    let deny = vec!["canvas.present".to_string()];
    let allowlist = resolve_node_command_allowlist(&allow, &deny, Some("darwin"), None);
    assert!(allowlist.contains("system.run"));
    assert!(!allowlist.contains("sms.send"));
    assert!(allowlist.contains("custom.command"));
    assert!(!allowlist.contains("canvas.present"));
}

// ============== Method Authorization Tests ==============

fn make_conn(role: &str) -> ConnectionContext {
    make_conn_with_scopes(role, vec![])
}

fn make_conn_with_scopes(role: &str, scopes: Vec<String>) -> ConnectionContext {
    ConnectionContext {
        conn_id: "test-conn".to_string(),
        role: role.to_string(),
        scopes,
        client: ClientInfo {
            id: "test-client".to_string(),
            version: "1.0".to_string(),
            platform: "test".to_string(),
            mode: "test".to_string(),
            display_name: None,
            device_family: None,
            model_identifier: None,
            instance_id: None,
        },
        device_id: None,
    }
}

fn make_conn_with_id(role: &str, scopes: Vec<String>, conn_id: &str) -> ConnectionContext {
    ConnectionContext {
        conn_id: conn_id.to_string(),
        role: role.to_string(),
        scopes,
        client: ClientInfo {
            id: "test-client".to_string(),
            version: "1.0".to_string(),
            platform: "test".to_string(),
            mode: "test".to_string(),
            display_name: None,
            device_family: None,
            model_identifier: None,
            instance_id: None,
        },
        device_id: None,
    }
}

#[test]
fn test_broadcast_event_scope_guard() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx_denied, mut rx_denied) = mpsc::channel(256);
    let (tx_allowed, mut rx_allowed) = mpsc::channel(256);
    let denied = make_conn_with_id("operator", vec![], "conn-denied");
    let allowed = make_conn_with_id(
        "operator",
        vec!["operator.pairing".to_string()],
        "conn-allowed",
    );

    state.register_connection(&denied, tx_denied, None);
    state.register_connection(&allowed, tx_allowed, None);

    // Clear presence broadcasts that are sent on connection registration
    let _ = rx_denied.try_recv();
    let _ = rx_denied.try_recv();
    let _ = rx_allowed.try_recv();
    let _ = rx_allowed.try_recv();

    broadcast_event(
        &state,
        "device.pair.requested",
        json!({ "requestId": "req-1" }),
    );

    assert!(rx_allowed.try_recv().is_ok());
    assert!(rx_denied.try_recv().is_err());
}

/// Default-closed broadcast scope for any future `matrix.*` event.
/// The current Matrix events are explicitly admin-scoped; the
/// fall-through pin guards against silent regressions where a new
/// `matrix.*` event added without an explicit `event_required_scope`
/// arm defaults to wide broadcast — which would surface sensitive
/// peer/device/room state to non-admin operator connections. The
/// fix routes any `matrix.*` event missing an explicit arm to
/// `operator.admin`; this test pins that contract using a synthetic
/// event name that will never collide with a real arm.
#[test]
fn test_broadcast_event_matrix_prefix_defaults_to_admin() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx_no_admin, mut rx_no_admin) = mpsc::channel(256);
    let (tx_admin_role, mut rx_admin_role) = mpsc::channel(256);
    let (tx_admin_scope, mut rx_admin_scope) = mpsc::channel(256);
    let no_admin = make_conn_with_id("operator", vec![], "matrix-no-admin");
    let admin_role = make_conn_with_id("admin", vec![], "matrix-admin-role");
    let admin_scope = make_conn_with_id(
        "operator",
        vec!["operator.admin".to_string()],
        "matrix-admin-scope",
    );

    state.register_connection(&no_admin, tx_no_admin, None);
    state.register_connection(&admin_role, tx_admin_role, None);
    state.register_connection(&admin_scope, tx_admin_scope, None);

    // Drain registration-time presence broadcasts.
    while rx_no_admin.try_recv().is_ok() {}
    while rx_admin_role.try_recv().is_ok() {}
    while rx_admin_scope.try_recv().is_ok() {}

    broadcast_event(
        &state,
        "matrix.synthetic.future_event",
        json!({ "deviceId": "DEVICE" }),
    );

    assert!(
        rx_no_admin.try_recv().is_err(),
        "operator without admin scope must not receive any matrix.* event missing an explicit arm"
    );
    assert!(
        rx_admin_role.try_recv().is_ok(),
        "admin-role connection must receive matrix.* default-admin events"
    );
    assert!(
        rx_admin_scope.try_recv().is_ok(),
        "operator with operator.admin scope must receive matrix.* default-admin events"
    );
}

#[test]
fn test_broadcast_event_emits_state_drop_for_payload_over_size_cap() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx, mut rx) = mpsc::channel(256);
    let admin = make_conn_with_id("admin", vec![], "oversize-admin");
    state.register_connection(&admin, tx, None);
    while rx.try_recv().is_ok() {}

    broadcast_event(
        &state,
        "chat",
        json!({ "body": "x".repeat(WS_BROADCAST_PAYLOAD_MAX_BYTES + 1) }),
    );

    let Message::Text(text) = rx
        .try_recv()
        .expect("oversized operator-visible broadcast should emit state.drop")
    else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "state.drop");
    assert_eq!(event["payload"]["event"], "chat");
    assert_eq!(event["payload"]["payloadClass"], "broadcast");
}

#[test]
fn test_broadcast_agent_event_truncates_oversized_payload() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx, mut rx) = mpsc::channel(256);
    let admin = make_conn_with_id("admin", vec![], "oversize-agent-admin");
    state.register_connection(&admin, tx, None);
    while rx.try_recv().is_ok() {}

    broadcast_event(
        &state,
        "agent",
        json!({
            "runId": "run-1",
            "seq": 7,
            "stream": "tool_result",
            "data": { "body": "x".repeat(WS_BROADCAST_PAYLOAD_MAX_BYTES + 1) }
        }),
    );

    let Message::Text(text) = rx
        .try_recv()
        .expect("oversized agent event should emit a truncation marker")
    else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "agent");
    assert_eq!(event["payload"]["runId"], "run-1");
    assert_eq!(event["payload"]["seq"], 7);
    assert_eq!(event["payload"]["stream"], "tool_result");
    assert_eq!(event["payload"]["truncated"], true);
    assert_eq!(event["payload"]["data"]["truncated"], true);
    assert_eq!(
        event["payload"]["reason"],
        "agent broadcast payload exceeded websocket frame cap"
    );
}

#[test]
fn test_matrix_verification_requested_rate_limited_per_peer_device() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx, mut rx) = mpsc::channel(256);
    let admin = make_conn_with_id("admin", vec![], "rate-admin");
    state.register_connection(&admin, tx, None);
    while rx.try_recv().is_ok() {}

    let payload = json!({
        "verification": {
            "flowId": "flow",
            "protocolFlowId": "txn",
            "userId": "@alice:example.com",
            "deviceId": "DEVICE",
            "state": "requested",
            "createdAt": 1,
            "updatedAt": 1
        },
        "ts": 1
    });
    for _ in 0..MATRIX_VERIFICATION_REQUEST_RATE_BURST {
        broadcast_event(&state, "matrix.verification.requested", payload.clone());
        assert!(
            rx.try_recv().is_ok(),
            "broadcast within burst limit should be delivered"
        );
    }

    broadcast_event(&state, "matrix.verification.requested", payload);
    assert!(
        rx.try_recv().is_err(),
        "broadcast exceeding per-peer/device burst limit must be dropped"
    );
}

#[test]
fn test_matrix_verification_rate_key_preserves_typed_components() {
    let left = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "flowId": "flow",
                "userId": "@alice:example.com\u{0}DEVICE",
                "deviceId": "A"
            }
        }),
    );
    let right = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "flowId": "flow",
                "userId": "@alice:example.com",
                "deviceId": "\u{0}DEVICE:A"
            }
        }),
    );
    assert_ne!(
        left, right,
        "typed key components must not collapse through delimiter concatenation"
    );
}

#[test]
fn test_matrix_verification_rate_key_uses_raw_device_not_flow_when_device_exists() {
    let flow_a = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "flowId": "flow-a",
                "userId": "@alice:example.com",
                "deviceId": "DEVICE"
            }
        }),
    );
    let flow_b = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "flowId": "flow-b",
                "userId": "@alice:example.com",
                "deviceId": "DEVICE"
            }
        }),
    );

    assert_eq!(
        flow_a, flow_b,
        "device-bearing verification events must share a raw user/device bucket"
    );
}

#[test]
fn test_matrix_verification_rate_peer_class_is_low_cardinality() {
    let device = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "flowId": "flow-a",
                "userId": "@alice:example.com",
                "deviceId": "DEVICE"
            }
        }),
    );
    assert_eq!(
        matrix_verification_rate_peer_class(&device, MatrixVerificationRequestRateClass::Normal),
        "peer_device"
    );

    let flow = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "flowId": "flow-a",
                "userId": "@alice:example.com"
            }
        }),
    );
    assert_eq!(
        matrix_verification_rate_peer_class(&flow, MatrixVerificationRequestRateClass::Normal),
        "peer_flow"
    );

    let malformed = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({"verification": {"flowId": "flow-a"}}),
    );
    assert_eq!(
        matrix_verification_rate_peer_class(
            &malformed,
            MatrixVerificationRequestRateClass::Malformed
        ),
        "malformed"
    );
}

/// The typed rate-class helper must route through
/// `MatrixVerificationState::is_terminal()` so every terminal variant
/// (Done, Cancelled, Mismatched) lands in `Finished` and every non-
/// terminal variant lands in `Normal`. The Value-based helper at
/// matrix_verification_request_rate_class duplicates this set as a
/// hand-maintained string match — a documented past bug shipped
/// because `mismatched` was missing from that list. This test pins
/// the typed path so future `MatrixVerificationState` variants are
/// classified by the enum directly.
#[test]
fn test_matrix_verification_request_rate_class_from_info_routes_through_is_terminal() {
    use crate::channels::matrix::{MatrixVerificationInfo, MatrixVerificationState};
    let info_with_state = |state: MatrixVerificationState| MatrixVerificationInfo {
        flow_id: "flow".to_string(),
        protocol_flow_id: "proto-flow".to_string(),
        raw_protocol_flow_id: "proto-flow".to_string(),
        user_id: "@alice:example.com".parse().expect("user id"),
        raw_user_id: "@alice:example.com".parse().expect("user id"),
        device_id: Some("DEVICE".into()),
        state,
        sas: None,
        created_at: 0,
        updated_at: 0,
    };
    for state in [
        MatrixVerificationState::Created,
        MatrixVerificationState::Requested,
        MatrixVerificationState::Ready,
        MatrixVerificationState::Transitioned,
        MatrixVerificationState::Started,
        MatrixVerificationState::Accepted,
        MatrixVerificationState::KeysExchanged,
        MatrixVerificationState::Confirmed,
    ] {
        let info = info_with_state(state.clone());
        let class = matrix_verification_request_rate_class_from_info(&info);
        assert_eq!(
            class,
            MatrixVerificationRequestRateClass::Normal,
            "non-terminal state {state:?} must route to Normal"
        );
    }
    for state in [
        MatrixVerificationState::Done,
        MatrixVerificationState::Cancelled,
        MatrixVerificationState::Mismatched,
    ] {
        let info = info_with_state(state.clone());
        let class = matrix_verification_request_rate_class_from_info(&info);
        assert_eq!(
            class,
            MatrixVerificationRequestRateClass::Finished,
            "terminal state {state:?} must route to Finished (incl. Mismatched, \
             which the Value-based variant historically missed)"
        );
    }

    let empty_device_empty_flow = MatrixVerificationInfo {
        flow_id: String::new(),
        protocol_flow_id: String::new(),
        raw_protocol_flow_id: String::new(),
        user_id: "@alice:example.com".parse().expect("user id"),
        raw_user_id: "@alice:example.com".parse().expect("user id"),
        device_id: Some(matrix_sdk::ruma::OwnedDeviceId::from("")),
        state: MatrixVerificationState::Requested,
        sas: None,
        created_at: 0,
        updated_at: 0,
    };
    assert_eq!(
        matrix_verification_request_rate_class_from_info(&empty_device_empty_flow),
        MatrixVerificationRequestRateClass::Malformed,
        "empty typed device_id with no flow must be treated as missing, not as a valid device"
    );
}

/// The typed key helper must yield the same key as the Value-based
/// helper for the same logical input, so the two gating paths stay
/// in lock-step. (The typed path is the production gate; the Value
/// path remains as defense-in-depth for any future caller bypassing
/// the typed broadcasters.)
#[test]
fn test_matrix_verification_request_rate_key_from_info_matches_value_path() {
    use crate::channels::matrix::{MatrixVerificationInfo, MatrixVerificationState};
    let cases: &[(&str, Option<&str>, &str)] = &[
        ("@alice:example.com", Some("DEVICE"), "flow-x"),
        ("@bob:example.com", None, "flow-y"),
        ("@carol:example.com", Some(""), "flow-w"),
    ];
    let event = "matrix.verification.requested";
    for (user_id, device_id, flow_id) in cases {
        let info = MatrixVerificationInfo {
            flow_id: flow_id.to_string(),
            protocol_flow_id: format!("{flow_id}-protocol"),
            raw_protocol_flow_id: format!("{flow_id}-protocol"),
            user_id: (*user_id).parse().expect("user id"),
            raw_user_id: (*user_id).parse().expect("user id"),
            device_id: device_id.map(matrix_sdk::ruma::OwnedDeviceId::from),
            state: MatrixVerificationState::Requested,
            sas: None,
            created_at: 0,
            updated_at: 0,
        };
        let mut value_payload = json!({
            "verification": {
                "flowId": flow_id,
                "userId": user_id,
            }
        });
        if let Some(d) = device_id {
            value_payload["verification"]["deviceId"] = json!(d);
        }
        let typed_key = matrix_verification_request_rate_key_from_info(event, &info);
        let value_key = matrix_verification_request_rate_key(event, &value_payload);
        assert_eq!(
            typed_key, value_key,
            "typed and value rate-key paths must match for ({user_id:?}, {device_id:?}, {flow_id:?})"
        );
    }
}

/// Regression: a hostile peer that spreads Normal-class claims
/// across N distinct user_ids (each ≤ MAX_NORMAL_KEYS_PER_PEER) can
/// fill the entire bucket table with Normal-class entries. Prior
/// to the most-claimed-peer eviction fix, `evict_low_value_bucket`
/// could only evict Malformed/Finished entries — when those didn't
/// exist, a legitimate new peer with zero existing buckets was
/// silently denied. The per-peer reserve also doesn't help the
/// legitimate peer because it requires their own pre-existing bucket
/// count > 0.
///
/// The fix evicts the OLDEST Normal-class bucket belonging to the
/// peer with the MOST Normal-class buckets — draining the attacker's
/// foothold rather than rejecting the legitimate new peer.
#[test]
fn test_matrix_verification_rate_table_evicts_most_claimed_peer_on_starvation() {
    let mut table = MatrixVerificationRequestRateTable::default();
    let now = Instant::now();

    // Saturate the table with Normal-class claims from a hostile
    // peer spreading across many user_ids. Phase 1: fill the
    // sub-reserve region (≤ MAX_KEYS - NEW_PEER_RESERVE) by stacking
    // up to MAX_NORMAL_KEYS_PER_PEER buckets under a few user_ids.
    // Phase 2: fill the final reserve slots with single-bucket
    // entries under distinct user_ids each (the per-peer reserve
    // only fires for peers with >0 existing buckets, so a fresh
    // user_id can always slot in one bucket).
    let mut peer_index = 0usize;
    let mut device_index = 0usize;
    while table.len() < MATRIX_VERIFICATION_REQUEST_RATE_MAX_KEYS {
        let key = MatrixVerificationRequestRateKey {
            user_id: format!("matrix.verification.requested:@hostile-{peer_index}:e.com"),
            device: MatrixVerificationRequestRateDevice::DeviceId {
                device_id: format!("DEVICE-{peer_index}-{device_index}"),
            },
        };
        let before = table.len();
        let _ = table.allow(key, MatrixVerificationRequestRateClass::Normal, now);
        let after = table.len();
        if after == before {
            // Refused — advance to a fresh user_id for the next claim
            // so the per-peer reserve doesn't keep refusing us.
            peer_index += 1;
            device_index = 0;
        } else {
            device_index += 1;
            if device_index >= MATRIX_VERIFICATION_REQUEST_RATE_MAX_NORMAL_KEYS_PER_PEER {
                peer_index += 1;
                device_index = 0;
            }
        }
        // Safety bound on the loop in case logic ever fails to
        // saturate; the test will then fail the assertion below
        // rather than spin forever.
        if peer_index > MATRIX_VERIFICATION_REQUEST_RATE_MAX_KEYS * 2 {
            break;
        }
    }
    assert_eq!(
        table.len(),
        MATRIX_VERIFICATION_REQUEST_RATE_MAX_KEYS,
        "test setup must saturate the table to exercise the eviction path"
    );

    // Legitimate new peer with zero existing buckets — prior to the
    // fix, this was silently rejected (`Limited`) because all buckets
    // are Normal-class (not evictable by `evict_low_value_bucket`) and
    // the per-peer reserve at 617-625 doesn't fire for a zero-bucket
    // peer.
    let legitimate_key = MatrixVerificationRequestRateKey {
        user_id: "matrix.verification.requested:@alice:legit.com".to_string(),
        device: MatrixVerificationRequestRateDevice::DeviceId {
            device_id: "ALICE-DEVICE".to_string(),
        },
    };
    let decision = table.allow(
        legitimate_key,
        MatrixVerificationRequestRateClass::Normal,
        now,
    );
    assert_eq!(
        decision,
        MatrixVerificationRequestRateDecision::Allowed,
        "legitimate new peer must NOT be starved by a hostile peer's distributed flood; \
         evict_oldest_normal_from_most_claimed_peer should have made room"
    );
}

#[test]
fn test_matrix_verification_rate_table_caps_unique_key_flood() {
    let mut table = MatrixVerificationRequestRateTable::default();
    let now = Instant::now();
    for index in 0..(MATRIX_VERIFICATION_REQUEST_RATE_MAX_KEYS + 64) {
        let key = MatrixVerificationRequestRateKey {
            user_id: format!("@peer-{index}:example.com"),
            device: MatrixVerificationRequestRateDevice::DeviceId {
                device_id: format!("DEVICE-{index}"),
            },
        };
        table.allow(key, MatrixVerificationRequestRateClass::Normal, now);
        assert!(
            table.len() <= MATRIX_VERIFICATION_REQUEST_RATE_MAX_KEYS,
            "rate table must stay bounded under unique-key flood"
        );
    }
    // Once the table is saturated, a new peer is admitted by evicting
    // an old bucket from the most-claimed peer (see
    // `test_matrix_verification_rate_table_evicts_most_claimed_peer_on_starvation`
    // for the threat model). The table size remains bounded at the
    // cap — eviction frees one slot, the new bucket takes it.
    let overflow_key = MatrixVerificationRequestRateKey {
        user_id: "@overflow:example.com".to_string(),
        device: MatrixVerificationRequestRateDevice::DeviceId {
            device_id: "OVERFLOW".to_string(),
        },
    };
    let decision = table.allow(
        overflow_key.clone(),
        MatrixVerificationRequestRateClass::Normal,
        now,
    );
    assert_eq!(
        decision,
        MatrixVerificationRequestRateDecision::Allowed,
        "saturated table must admit legitimate new peers by evicting oldest from most-claimed peer"
    );
    assert!(
        table.buckets.contains_key(&overflow_key),
        "new peer's bucket must be in the table after eviction-and-admit"
    );
    assert_eq!(
        table.len(),
        MATRIX_VERIFICATION_REQUEST_RATE_MAX_KEYS,
        "table size remains bounded at the cap (eviction freed one slot, new bucket took it)"
    );
}

#[test]
fn test_matrix_verification_updated_broadcast_uses_separate_rate_limiter() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx, mut rx) = mpsc::channel(256);
    let conn = make_conn_with_id("operator", vec!["operator.admin".to_string()], "conn-1");
    state.register_connection(&conn, tx, None);
    let _ = rx.try_recv();

    let payload = json!({
        "verification": {
            "flowId": "same-flow",
            "protocolFlowId": "same-flow",
            "userId": "@alice:example.com",
            "deviceId": "DEVICE",
            "state": "requested",
            "createdAt": 1,
            "updatedAt": 1
        },
        "ts": 1
    });
    for _ in 0..MATRIX_VERIFICATION_REQUEST_RATE_BURST {
        broadcast_event(&state, "matrix.verification.requested", payload.clone());
        assert!(rx.try_recv().is_ok());
    }

    broadcast_event(&state, "matrix.verification.updated", payload);
    assert!(
        rx.try_recv().is_ok(),
        "updated emissions must have budget separate from requested bursts"
    );
}

#[test]
fn test_matrix_verification_rate_missing_device_uses_flow_bucket() {
    let mut table = MatrixVerificationRequestRateTable::default();
    let now = Instant::now();
    let flow_a = MatrixVerificationRequestRateKey {
        user_id: "@alice:example.com".to_string(),
        device: MatrixVerificationRequestRateDevice::MissingDevice {
            flow_id: "flow-a".to_string(),
        },
    };
    let flow_b = MatrixVerificationRequestRateKey {
        user_id: "@alice:example.com".to_string(),
        device: MatrixVerificationRequestRateDevice::MissingDevice {
            flow_id: "flow-b".to_string(),
        },
    };

    for _ in 0..MATRIX_VERIFICATION_REQUEST_RATE_BURST {
        assert_ne!(
            table.allow(
                flow_a.clone(),
                MatrixVerificationRequestRateClass::Normal,
                now
            ),
            MatrixVerificationRequestRateDecision::Limited
        );
    }
    assert_eq!(
        table.allow(flow_a, MatrixVerificationRequestRateClass::Normal, now),
        MatrixVerificationRequestRateDecision::Limited,
        "same missing-device flow must share a bounded bucket"
    );
    assert_ne!(
        table.allow(flow_b, MatrixVerificationRequestRateClass::Normal, now),
        MatrixVerificationRequestRateDecision::Limited,
        "different missing-device flows must not collapse into one peer bucket"
    );
}

#[test]
fn test_matrix_verification_rate_refill_is_continuous_not_window_burst() {
    let mut table = MatrixVerificationRequestRateTable::default();
    let now = Instant::now();
    let key = MatrixVerificationRequestRateKey {
        user_id: "@alice:example.com".to_string(),
        device: MatrixVerificationRequestRateDevice::DeviceId {
            device_id: "DEVICE".to_string(),
        },
    };

    for _ in 0..MATRIX_VERIFICATION_REQUEST_RATE_BURST {
        assert_ne!(
            table.allow(key.clone(), MatrixVerificationRequestRateClass::Normal, now),
            MatrixVerificationRequestRateDecision::Limited
        );
    }
    assert_eq!(
        table.allow(key.clone(), MatrixVerificationRequestRateClass::Normal, now),
        MatrixVerificationRequestRateDecision::Limited
    );

    let almost_full_window =
        now + MATRIX_VERIFICATION_REQUEST_RATE_WINDOW - Duration::from_millis(1);
    let mut refilled = 0;
    for _ in 0..MATRIX_VERIFICATION_REQUEST_RATE_BURST {
        if table.allow(
            key.clone(),
            MatrixVerificationRequestRateClass::Normal,
            almost_full_window,
        ) != MatrixVerificationRequestRateDecision::Limited
        {
            refilled += 1;
        }
    }
    assert!(
        refilled < MATRIX_VERIFICATION_REQUEST_RATE_BURST,
        "rate limiter must not grant a full burst before the window elapses"
    );
}

#[test]
fn test_matrix_verification_rate_limited_requests_do_not_refresh_last_seen() {
    let mut table = MatrixVerificationRequestRateTable::default();
    let now = Instant::now();
    let key = MatrixVerificationRequestRateKey {
        user_id: "@alice:example.com".to_string(),
        device: MatrixVerificationRequestRateDevice::DeviceId {
            device_id: "DEVICE".to_string(),
        },
    };

    for _ in 0..MATRIX_VERIFICATION_REQUEST_RATE_BURST {
        assert_ne!(
            table.allow(key.clone(), MatrixVerificationRequestRateClass::Normal, now),
            MatrixVerificationRequestRateDecision::Limited
        );
    }
    let limited_at = now + Duration::from_secs(1);
    assert_eq!(
        table.allow(
            key.clone(),
            MatrixVerificationRequestRateClass::Normal,
            limited_at
        ),
        MatrixVerificationRequestRateDecision::Limited
    );

    table.prune_expired(
        limited_at + MATRIX_VERIFICATION_REQUEST_RATE_WINDOW * 2 - Duration::from_millis(1),
    );
    assert_eq!(table.len(), 0);
}

#[test]
fn test_matrix_verification_rate_missing_flow_does_not_collide_with_literal_flow_id() {
    let missing = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "userId": "@alice:example.com"
            }
        }),
    );
    let literal = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "userId": "@alice:example.com",
                "flowId": "<missing-flow>"
            }
        }),
    );
    assert_ne!(
        missing, literal,
        "missing flow must be a typed bucket, not a spoofable sentinel string"
    );
}

#[test]
fn test_matrix_verification_rate_malformed_payload_churn_uses_bounded_bucket() {
    let mut table = MatrixVerificationRequestRateTable::default();
    let now = Instant::now();
    for index in 0..MATRIX_VERIFICATION_REQUEST_RATE_BURST {
        let key = matrix_verification_request_rate_key(
            "matrix.verification.requested",
            &json!({
                "verification": {
                    "irrelevant": format!("churn-{index}")
                }
            }),
        );
        assert_eq!(
            table.allow(key, MatrixVerificationRequestRateClass::Malformed, now),
            MatrixVerificationRequestRateDecision::Allowed
        );
    }
    let churned_key = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "irrelevant": "new-payload"
            }
        }),
    );
    assert_eq!(
        table.allow(
            churned_key,
            MatrixVerificationRequestRateClass::Malformed,
            now
        ),
        MatrixVerificationRequestRateDecision::Limited,
        "malformed payload churn must not mint fresh buckets"
    );
    assert_eq!(table.len(), 1);
}

#[test]
fn test_matrix_verification_rate_malformed_claimed_user_churn_uses_bounded_bucket() {
    let mut table = MatrixVerificationRequestRateTable::default();
    let now = Instant::now();
    for index in 0..MATRIX_VERIFICATION_REQUEST_RATE_BURST {
        let key = matrix_verification_request_rate_key(
            "matrix.verification.requested",
            &json!({
                "verification": {
                    "userId": format!("@attacker-{index}:example.com")
                }
            }),
        );
        assert_eq!(
            table.allow(key, MatrixVerificationRequestRateClass::Malformed, now),
            MatrixVerificationRequestRateDecision::Allowed
        );
    }
    let churned_key = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "userId": "@attacker-new:example.com"
            }
        }),
    );
    assert_eq!(
        table.allow(
            churned_key,
            MatrixVerificationRequestRateClass::Malformed,
            now
        ),
        MatrixVerificationRequestRateDecision::Limited,
        "claimed userId churn must not mint fresh malformed buckets"
    );
    assert_eq!(table.len(), 1);
}

#[test]
fn test_matrix_verification_rate_normal_per_peer_cap_preserves_other_peers() {
    let mut table = MatrixVerificationRequestRateTable::default();
    let now = Instant::now();
    for index in 0..MATRIX_VERIFICATION_REQUEST_RATE_MAX_NORMAL_KEYS_PER_PEER {
        let key = MatrixVerificationRequestRateKey {
            user_id: "matrix.verification.requested:@hostile:example.com".to_string(),
            device: MatrixVerificationRequestRateDevice::MissingDevice {
                flow_id: format!("flow-{index}"),
            },
        };
        assert_eq!(
            table.allow(key, MatrixVerificationRequestRateClass::Normal, now),
            MatrixVerificationRequestRateDecision::Allowed
        );
    }
    let hostile_overflow = MatrixVerificationRequestRateKey {
        user_id: "matrix.verification.requested:@hostile:example.com".to_string(),
        device: MatrixVerificationRequestRateDevice::MissingDevice {
            flow_id: "overflow".to_string(),
        },
    };
    assert_eq!(
        table.allow(
            hostile_overflow,
            MatrixVerificationRequestRateClass::Normal,
            now
        ),
        MatrixVerificationRequestRateDecision::Limited,
        "one peer must not exceed its normal-flow bucket cap"
    );
    let legitimate = MatrixVerificationRequestRateKey {
        user_id: "matrix.verification.requested:@legit:example.com".to_string(),
        device: MatrixVerificationRequestRateDevice::MissingDevice {
            flow_id: "flow-a".to_string(),
        },
    };
    assert_eq!(
        table.allow(legitimate, MatrixVerificationRequestRateClass::Normal, now),
        MatrixVerificationRequestRateDecision::Allowed,
        "a capped peer must not block another peer's normal flow"
    );
}

#[test]
fn test_matrix_verification_rate_normal_reserves_capacity_for_new_peers() {
    let mut table = MatrixVerificationRequestRateTable::default();
    let now = Instant::now();
    let reservable = MATRIX_VERIFICATION_REQUEST_RATE_MAX_KEYS
        - MATRIX_VERIFICATION_REQUEST_RATE_NEW_PEER_RESERVE;
    for index in 0..reservable {
        let peer = index / MATRIX_VERIFICATION_REQUEST_RATE_MAX_NORMAL_KEYS_PER_PEER;
        let flow = index % MATRIX_VERIFICATION_REQUEST_RATE_MAX_NORMAL_KEYS_PER_PEER;
        let key = MatrixVerificationRequestRateKey {
            user_id: format!("matrix.verification.requested:@hostile-{peer}:example.com"),
            device: MatrixVerificationRequestRateDevice::MissingDevice {
                flow_id: format!("flow-{flow}"),
            },
        };
        assert_eq!(
            table.allow(key, MatrixVerificationRequestRateClass::Normal, now),
            MatrixVerificationRequestRateDecision::Allowed
        );
    }

    let same_peer_overflow = MatrixVerificationRequestRateKey {
        user_id: "matrix.verification.requested:@hostile-0:example.com".to_string(),
        device: MatrixVerificationRequestRateDevice::MissingDevice {
            flow_id: "reserved-overflow".to_string(),
        },
    };
    assert_eq!(
        table.allow(
            same_peer_overflow,
            MatrixVerificationRequestRateClass::Normal,
            now
        ),
        MatrixVerificationRequestRateDecision::Limited,
        "existing peers must not consume the reserve held for first flows from new peers"
    );

    let new_peer = MatrixVerificationRequestRateKey {
        user_id: "matrix.verification.requested:@new-peer:example.com".to_string(),
        device: MatrixVerificationRequestRateDevice::MissingDevice {
            flow_id: "first-flow".to_string(),
        },
    };
    assert_eq!(
        table.allow(new_peer, MatrixVerificationRequestRateClass::Normal, now),
        MatrixVerificationRequestRateDecision::Allowed,
        "a small hostile peer set must not fill every normal bucket"
    );
}

#[test]
fn test_matrix_verification_rate_missing_device_and_flow_uses_malformed_bucket() {
    let key = matrix_verification_request_rate_key(
        "matrix.verification.requested",
        &json!({
            "verification": {
                "userId": "@alice:example.com"
            }
        }),
    );
    assert_eq!(
        key.device,
        MatrixVerificationRequestRateDevice::MalformedMissingDevice
    );
}

#[test]
fn test_matrix_verification_rate_prunes_expired_buckets() {
    let mut table = MatrixVerificationRequestRateTable::default();
    let now = Instant::now();
    let key = MatrixVerificationRequestRateKey {
        user_id: "@alice:example.com".to_string(),
        device: MatrixVerificationRequestRateDevice::DeviceId {
            device_id: "DEVICE".to_string(),
        },
    };
    table.allow(key, MatrixVerificationRequestRateClass::Normal, now);
    assert_eq!(table.len(), 1);

    table.prune_expired(
        now + MATRIX_VERIFICATION_REQUEST_RATE_WINDOW * 2 + Duration::from_millis(1),
    );

    assert_eq!(table.len(), 0);
}

#[test]
fn test_ws_broadcast_backpressure_uses_connection_cleanup() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx, _rx) = mpsc::channel(1);
    let admin = make_conn_with_id("admin", vec![], "backpressure-admin");
    state.register_connection(&admin, tx, None);

    broadcast_event(&state, "health", json!({ "status": "degraded" }));

    assert!(
        !state.connections.lock().contains_key("backpressure-admin"),
        "backpressured client must be unregistered through the lifecycle cleanup path"
    );
    assert!(
        state.get_presence_list().is_empty(),
        "presence must be removed with the backpressured connection"
    );
    let snapshot = state.get_health_snapshot();
    let ws = snapshot
        .ws
        .expect("health snapshot should expose WS counters");
    assert_eq!(ws.broadcast_drop_total, 1);
}

#[test]
fn test_connection_tx_enforces_max_buffered_bytes_and_closes() {
    let (raw_tx, mut rx) = mpsc::channel::<QueuedWsMessage>(4);
    let (close_tx, close_rx) = tokio::sync::watch::channel(false);
    let tx = ConnectionTx::Metered(MeteredConnectionTx {
        tx: raw_tx,
        queued_bytes: Arc::new(AtomicUsize::new(0)),
        closed: Arc::new(AtomicBool::new(false)),
        close_tx,
        max_buffered_bytes: 5,
    });

    assert!(send_text(&tx, "1234".to_string()).is_ok());
    assert!(send_text(&tx, "12".to_string()).is_err());
    assert!(
        *close_rx.borrow(),
        "byte-budget overflow must force-close the connection lifecycle"
    );
    let queued = rx.try_recv().expect("first frame should remain queued");
    assert_eq!(queued.bytes, 4);
}

#[test]
fn test_health_snapshot_exposes_ws_drop_counters() {
    let state = WsServerState::new(WsServerConfig::default());
    state.record_ws_broadcast_drop();
    state.record_matrix_verification_rate_limit_drop();

    let snapshot = state.get_health_snapshot();
    let ws = snapshot
        .ws
        .expect("health snapshot should expose WS counters");
    assert_eq!(ws.broadcast_drop_total, 1);
    assert_eq!(ws.matrix_verification_rate_limit_drop_total, 1);
    assert_eq!(ws.max_buffered_bytes, MAX_BUFFERED_BYTES);

    let metrics = crate::server::metrics::METRICS.render();
    assert!(metrics.contains("carapace_ws_broadcast_drops_total"));
    assert!(metrics.contains("carapace_matrix_verification_rate_limit_drops_total"));
}

#[test]
fn test_role_satisfies() {
    // Any role satisfies read
    assert!(role_satisfies("read", "read"));
    assert!(role_satisfies("write", "read"));
    assert!(role_satisfies("admin", "read"));
    assert!(role_satisfies("operator", "read"));

    // Only write, admin, operator satisfy write
    assert!(!role_satisfies("read", "write"));
    assert!(role_satisfies("write", "write"));
    assert!(role_satisfies("admin", "write"));
    assert!(role_satisfies("operator", "write"));

    // Only admin satisfies admin
    assert!(!role_satisfies("read", "admin"));
    assert!(!role_satisfies("write", "admin"));
    assert!(role_satisfies("admin", "admin"));
}

#[test]
fn test_method_authorization_read_methods() {
    // Read-only methods should be allowed by any role
    let read_methods = [
        "health",
        "status",
        "config.get",
        "sessions.list",
        "channels.status",
        "plugins.bins",
    ];

    for method in read_methods {
        for role in ["read", "write", "admin"] {
            let conn = make_conn(role);
            let result = check_method_authorization(method, &conn);
            assert!(
                result.is_ok(),
                "Method '{}' should be allowed for role '{}'",
                method,
                role
            );
        }
    }
}

#[test]
fn test_method_authorization_write_methods() {
    // Write methods should not be allowed by read role
    let write_methods = ["config.set", "agent", "chat.send", "cron.add"];

    for method in write_methods {
        let read_conn = make_conn("read");
        let result = check_method_authorization(method, &read_conn);
        assert!(
            result.is_err(),
            "Method '{}' should NOT be allowed for role 'read'",
            method
        );

        // But allowed for write and admin
        for role in ["write", "admin"] {
            let conn = make_conn(role);
            let result = check_method_authorization(method, &conn);
            assert!(
                result.is_ok(),
                "Method '{}' should be allowed for role '{}'",
                method,
                role
            );
        }
    }
}

#[test]
fn test_method_authorization_admin_methods() {
    // Admin methods should only be allowed by admin role
    let admin_methods = [
        "device.pair.approve",
        "device.token.rotate",
        "exec.approvals.set",
        "node.pair.approve",
    ];

    for method in admin_methods {
        // Not allowed for read or write
        for role in ["read", "write"] {
            let conn = make_conn(role);
            let result = check_method_authorization(method, &conn);
            assert!(
                result.is_err(),
                "Method '{}' should NOT be allowed for role '{}'",
                method,
                role
            );
        }

        // Allowed for admin
        let admin_conn = make_conn("admin");
        let result = check_method_authorization(method, &admin_conn);
        assert!(
            result.is_ok(),
            "Method '{}' should be allowed for role 'admin'",
            method
        );
    }
}

#[test]
fn test_method_authorization_unknown_method_requires_admin() {
    // Unknown methods should require admin (fail secure)
    let conn = make_conn("write");
    let result = check_method_authorization("unknown.method.xyz", &conn);
    assert!(result.is_err(), "Unknown method should require admin role");

    let admin_conn = make_conn("admin");
    let result = check_method_authorization("unknown.method.xyz", &admin_conn);
    assert!(result.is_ok(), "Unknown method should be allowed for admin");
}

#[tokio::test]
async fn test_legacy_ws_method_aliases_are_unknown_on_wire() {
    let state = Arc::new(WsServerState::new(WsServerConfig::default()));
    let conn = make_conn("admin");
    let aliases = [
        "agent.run",
        "agent.cancel",
        "session.list",
        "session.preview",
        "session.create",
        "session.load",
        "session.fork",
        "session.rename",
        "session.switch",
        "session.patch",
        "session.reset",
        "session.delete",
        "session.compact",
        "session.archive",
        "session.restore",
        "session.archives",
        "session.archive.delete",
        "session.export_user",
        "session.purge_user",
        "config.update",
        "exec.list",
        "exec.approvals.list",
        "exec.approve",
        "exec.deny",
    ];

    for alias in aliases {
        let err = dispatch_method(alias, Some(&json!({})), &state, &conn)
            .await
            .expect_err("legacy method alias should be unavailable");
        assert_eq!(err.code, ERROR_UNAVAILABLE, "{alias}");
        assert_eq!(err.message, "method unavailable", "{alias}");
        assert_eq!(
            err.details
                .as_ref()
                .and_then(|details| details.get("method"))
                .and_then(|method| method.as_str()),
            Some(alias),
            "{alias}"
        );
        assert!(
            !GATEWAY_METHODS.contains(&alias),
            "legacy alias should not be advertised as a current method"
        );

        let (tx, mut rx) = tokio::sync::mpsc::channel(256);
        send_dispatch_result(
            &ConnectionTx::from(tx),
            "legacy-alias-test",
            alias,
            false,
            Err(err),
        );
        let Some(Message::Text(text)) = rx.recv().await else {
            panic!("expected response frame for legacy alias {alias}");
        };
        let frame: Value = serde_json::from_str(&text).unwrap();
        assert_eq!(frame["ok"], false, "{alias}");
        assert_eq!(frame["error"]["code"], ERROR_INVALID_REQUEST, "{alias}");
        assert_eq!(frame["error"]["message"], "unknown method", "{alias}");
        assert_eq!(frame["error"]["details"]["method"], alias, "{alias}");
    }
}

#[test]
fn test_method_authorization_error_contains_details() {
    let conn = make_conn("read");
    let result = check_method_authorization("config.set", &conn);

    assert!(result.is_err());
    let err = result.unwrap_err();
    assert_eq!(err.code, "invalid_request");
    assert!(err.message.contains("config.set"));
    assert!(err.message.contains("write"));
    assert!(err.message.contains("read"));
}

// ============== Node Role Allowlist Tests ==============

#[test]
fn test_node_role_only_allows_specific_methods() {
    let conn = make_conn("node");

    // Allowed methods for node role
    for method in NODE_ONLY_METHODS {
        let result = check_method_authorization(method, &conn);
        assert!(
            result.is_ok(),
            "Method '{}' should be allowed for node role",
            method
        );
    }

    // Read methods should NOT be allowed for node role
    let blocked_methods = ["health", "status", "config.get", "sessions.list"];
    for method in blocked_methods {
        let result = check_method_authorization(method, &conn);
        assert!(
            result.is_err(),
            "Method '{}' should NOT be allowed for node role",
            method
        );
    }
}

// ============== Operator Scope Tests ==============

#[test]
fn test_operator_without_scopes_cannot_write() {
    // Operator with no scopes should not be able to call write methods
    let conn = make_conn_with_scopes("operator", vec![]);

    let result = check_method_authorization("config.set", &conn);
    assert!(
        result.is_err(),
        "Operator without scopes should not be able to write"
    );
}

#[test]
fn test_operator_with_write_scope_can_write() {
    let conn = make_conn_with_scopes("operator", vec!["operator.write".to_string()]);

    // config.set requires operator.admin (it's in OPERATOR_ADMIN_REQUIRED_METHODS)
    let result = check_method_authorization("config.set", &conn);
    assert!(
        result.is_err(),
        "config.set requires operator.admin, not just write scope"
    );

    // sessions.patch also requires operator.admin
    let result = check_method_authorization("sessions.patch", &conn);
    assert!(
        result.is_err(),
        "sessions.patch requires operator.admin, not just write scope"
    );

    // agent/chat are write-level methods that DO work with operator.write
    let result = check_method_authorization("agent", &conn);
    assert!(
        result.is_ok(),
        "agent is a write-level method that works with operator.write"
    );

    let result = check_method_authorization("chat.send", &conn);
    assert!(
        result.is_ok(),
        "chat.send is a write-level method that works with operator.write"
    );
}

#[test]
fn test_operator_with_read_scope_can_read() {
    let conn = make_conn_with_scopes("operator", vec!["operator.read".to_string()]);

    // Read methods should work
    let result = check_method_authorization("health", &conn);
    assert!(
        result.is_ok(),
        "Operator with read scope should be able to read"
    );

    // Write methods should not work
    let result = check_method_authorization("config.set", &conn);
    assert!(
        result.is_err(),
        "Operator with only read scope should not be able to write"
    );
}

#[test]
fn test_operator_needs_pairing_scope_for_pairing() {
    // Per Node.js gateway: operator.pairing allows pairing methods WITHOUT needing operator.admin
    // This enables granular access control where operators can be granted just pairing rights

    // Operator with admin scope - can pair (admin covers all)
    let conn = make_conn_with_scopes("operator", vec!["operator.admin".to_string()]);
    let result = check_method_authorization("device.pair.approve", &conn);
    assert!(
        result.is_ok(),
        "Operator with admin scope should be able to pair"
    );

    // Operator with only write scope - cannot pair (needs pairing or admin)
    let conn_write = make_conn_with_scopes("operator", vec!["operator.write".to_string()]);
    let result = check_method_authorization("device.pair.approve", &conn_write);
    assert!(
        result.is_err(),
        "Operator with only write scope should not be able to pair"
    );

    // Operator with just pairing scope - CAN pair (per Node.js gateway)
    let conn_pairing = make_conn_with_scopes("operator", vec!["operator.pairing".to_string()]);
    let result = check_method_authorization("device.pair.approve", &conn_pairing);
    assert!(
        result.is_ok(),
        "Operator with pairing scope should be able to pair (Node.js parity)"
    );

    // Operator with read scope only - cannot pair
    let conn_read = make_conn_with_scopes("operator", vec!["operator.read".to_string()]);
    let result = check_method_authorization("device.pair.approve", &conn_read);
    assert!(
        result.is_err(),
        "Operator with only read scope should not be able to pair"
    );
}

#[test]
fn test_operator_needs_approvals_scope_for_exec_approvals() {
    // Per Node.js gateway: operator.approvals allows exec approval methods WITHOUT needing operator.admin

    let conn = make_conn_with_scopes("operator", vec!["operator.write".to_string()]);
    let result = check_method_authorization("exec.approvals.set", &conn);
    assert!(
        result.is_err(),
        "Operator without approvals scope should not set approvals"
    );

    // Admin scope covers all
    let conn_admin = make_conn_with_scopes("operator", vec!["operator.admin".to_string()]);
    let result = check_method_authorization("exec.approvals.set", &conn_admin);
    assert!(
        result.is_ok(),
        "Operator with admin scope should set approvals"
    );

    // Approvals scope alone allows exec approval methods (per Node.js gateway)
    let conn_approvals = make_conn_with_scopes("operator", vec!["operator.approvals".to_string()]);
    let result = check_method_authorization("exec.approvals.set", &conn_approvals);
    assert!(
        result.is_ok(),
        "Operator with approvals scope should set approvals (Node.js parity)"
    );
}

#[test]
fn test_operator_wildcard_scope() {
    let conn = make_conn_with_scopes("operator", vec!["operator.*".to_string()]);

    // Wildcard should cover all operations (covers operator.admin, operator.pairing, etc.)
    assert!(
        check_method_authorization("config.set", &conn).is_ok(),
        "wildcard covers operator.admin for config.set"
    );
    assert!(
        check_method_authorization("device.pair.approve", &conn).is_ok(),
        "wildcard covers operator.pairing"
    );
    assert!(
        check_method_authorization("exec.approvals.set", &conn).is_ok(),
        "wildcard covers operator.approvals"
    );
    assert!(
        check_method_authorization("agent", &conn).is_ok(),
        "wildcard covers operator.write"
    );
    assert!(
        check_method_authorization("health", &conn).is_ok(),
        "wildcard covers operator.read"
    );
}

#[test]
fn test_scope_satisfies() {
    // Exact match
    assert!(scope_satisfies(
        &["operator.write".to_string()],
        "operator.write"
    ));
    assert!(!scope_satisfies(
        &["operator.read".to_string()],
        "operator.write"
    ));

    // Wildcard
    assert!(scope_satisfies(
        &["operator.*".to_string()],
        "operator.pairing"
    ));
    assert!(scope_satisfies(
        &["operator.*".to_string()],
        "operator.admin"
    ));

    // Admin covers all
    assert!(scope_satisfies(
        &["operator.admin".to_string()],
        "operator.pairing"
    ));
    assert!(scope_satisfies(
        &["operator.admin".to_string()],
        "operator.approvals"
    ));

    // Write covers read
    assert!(scope_satisfies(
        &["operator.write".to_string()],
        "operator.read"
    ));
}

// ============== Node Pairing Handler Tests ==============

#[test]
fn test_handle_node_pair_request() {
    let state = WsServerState::new(WsServerConfig::default());

    let params = json!({
        "nodeId": "test-node-1",
        "displayName": "Test Node",
        "platform": "darwin",
        "commands": ["system.run", "camera.snap"]
    });

    let result = handle_node_pair_request(Some(&params), &state);
    assert!(result.is_ok());

    let response = result.unwrap();
    assert_eq!(response["status"], "pending");
    assert_eq!(response["created"], true);
    assert_eq!(response["request"]["nodeId"], "test-node-1");
    assert!(response["request"]["requestId"].as_str().is_some());
}

#[test]
fn test_handle_node_pair_request_requires_node_id() {
    let state = WsServerState::new(WsServerConfig::default());

    let params = json!({
        "displayName": "Test Node"
    });

    let result = handle_node_pair_request(Some(&params), &state);
    assert!(result.is_err());
    assert_eq!(result.unwrap_err().code, ERROR_INVALID_REQUEST);
}

#[test]
fn test_handle_node_pair_list() {
    let state = WsServerState::new(WsServerConfig::default());

    // Create a pairing request
    let params = json!({
        "nodeId": "test-node-1",
        "displayName": "Test Node"
    });
    handle_node_pair_request(Some(&params), &state).unwrap();

    // List should show the pending request
    let result = handle_node_pair_list(&state);
    assert!(result.is_ok());

    let response = result.unwrap();
    assert_eq!(response["paired"].as_array().unwrap().len(), 0);
    assert_eq!(response["pending"].as_array().unwrap().len(), 1);
    assert_eq!(response["pending"][0]["nodeId"], "test-node-1");
}

#[test]
fn test_handle_node_pair_approve_and_verify() {
    let state = WsServerState::new(WsServerConfig::default());

    // Create a pairing request
    let request_params = json!({
        "nodeId": "test-node-1",
        "displayName": "Test Node"
    });
    let request_response = handle_node_pair_request(Some(&request_params), &state).unwrap();
    let request_id = request_response["request"]["requestId"].as_str().unwrap();

    // Approve the request
    let approve_params = json!({ "requestId": request_id });
    let approve_result = handle_node_pair_approve(Some(&approve_params), &state);
    assert!(approve_result.is_ok());

    let approve_response = approve_result.unwrap();
    assert_eq!(approve_response["requestId"], request_id);
    assert_eq!(approve_response["node"]["nodeId"], "test-node-1");
    let token = approve_response["node"]["token"].as_str().unwrap();
    assert!(!token.is_empty());

    // Verify the token
    let verify_params = json!({
        "nodeId": "test-node-1",
        "token": token
    });
    let verify_result = handle_node_pair_verify(Some(&verify_params), &state);
    assert!(verify_result.is_ok());
    assert_eq!(verify_result.unwrap()["ok"], true);
}

#[test]
fn test_handle_node_pair_reject() {
    let state = WsServerState::new(WsServerConfig::default());

    // Create a pairing request
    let request_params = json!({
        "nodeId": "test-node-1"
    });
    let request_response = handle_node_pair_request(Some(&request_params), &state).unwrap();
    let request_id = request_response["request"]["requestId"].as_str().unwrap();

    // Reject the request
    let reject_params = json!({
        "requestId": request_id,
        "reason": "Not authorized"
    });
    let reject_result = handle_node_pair_reject(Some(&reject_params), &state);
    assert!(reject_result.is_ok());

    let reject_response = reject_result.unwrap();
    assert_eq!(reject_response["requestId"], request_id);
    assert_eq!(reject_response["nodeId"], "test-node-1");

    // Node should not be paired
    assert!(!state.node_pairing.is_paired("test-node-1"));
}

#[test]
fn test_handle_node_pair_verify_requires_pairing() {
    let state = WsServerState::new(WsServerConfig::default());

    // Try to verify without pairing
    let verify_params = json!({
        "nodeId": "unpaired-node",
        "token": "some-token"
    });
    let verify_result = handle_node_pair_verify(Some(&verify_params), &state);
    assert!(verify_result.is_ok());
    assert_eq!(verify_result.unwrap()["ok"], false);
}

#[test]
fn test_handle_node_list() {
    let state = WsServerState::new(WsServerConfig::default());

    // Initially empty
    let result = handle_node_list(&state);
    assert!(result.is_ok());
    assert_eq!(result.unwrap()["nodes"].as_array().unwrap().len(), 0);

    // Pair a node
    let request_params = json!({
        "nodeId": "test-node-1",
        "displayName": "Test Node",
        "platform": "darwin"
    });
    let request_response = handle_node_pair_request(Some(&request_params), &state).unwrap();
    let request_id = request_response["request"]["requestId"].as_str().unwrap();
    handle_node_pair_approve(Some(&json!({ "requestId": request_id })), &state).unwrap();

    // Should now have one node
    let result = handle_node_list(&state);
    assert!(result.is_ok());
    let binding = result.unwrap();
    let nodes = binding["nodes"].as_array().unwrap();
    assert_eq!(nodes.len(), 1);
    assert_eq!(nodes[0]["nodeId"], "test-node-1");
    assert_eq!(nodes[0]["displayName"], "Test Node");
    assert_eq!(nodes[0]["platform"], "darwin");
    assert_eq!(nodes[0]["paired"], true);
    assert_eq!(nodes[0]["connected"], false);
}

#[test]
fn test_handle_node_rename() {
    let state = WsServerState::new(WsServerConfig::default());

    // Pair a node
    let request_params = json!({
        "nodeId": "test-node-1",
        "displayName": "Old Name"
    });
    let request_response = handle_node_pair_request(Some(&request_params), &state).unwrap();
    let request_id = request_response["request"]["requestId"].as_str().unwrap();
    handle_node_pair_approve(Some(&json!({ "requestId": request_id })), &state).unwrap();

    // Rename the node
    let rename_params = json!({
        "nodeId": "test-node-1",
        "name": "New Name"
    });
    let result = handle_node_rename(Some(&rename_params), &state);
    assert!(result.is_ok());
    assert_eq!(result.unwrap()["name"], "New Name");

    // Verify the name changed
    let node = state.node_pairing.get_paired_node("test-node-1").unwrap();
    assert_eq!(node.display_name, Some("New Name".to_string()));
}

#[test]
fn test_handle_node_describe() {
    let state = WsServerState::new(WsServerConfig::default());

    // Pair a node
    let request_params = json!({
        "nodeId": "test-node-1",
        "displayName": "Test Node",
        "platform": "darwin",
        "commands": ["system.run"]
    });
    let request_response = handle_node_pair_request(Some(&request_params), &state).unwrap();
    let request_id = request_response["request"]["requestId"].as_str().unwrap();
    handle_node_pair_approve(Some(&json!({ "requestId": request_id })), &state).unwrap();

    // Describe the node
    let describe_params = json!({ "nodeId": "test-node-1" });
    let result = handle_node_describe(Some(&describe_params), &state);
    assert!(result.is_ok());

    let response = result.unwrap();
    assert_eq!(response["nodeId"], "test-node-1");
    assert_eq!(response["displayName"], "Test Node");
    assert_eq!(response["platform"], "darwin");
    assert_eq!(
        response["commands"].as_array().unwrap(),
        &vec![json!("system.run")]
    );
    assert_eq!(response["paired"], true);
    assert_eq!(response["connected"], false);
}

#[test]
fn test_handle_node_describe_requires_pairing() {
    let state = WsServerState::new(WsServerConfig::default());

    let describe_params = json!({ "nodeId": "unpaired-node" });
    let result = handle_node_describe(Some(&describe_params), &state);
    assert!(result.is_err());
    assert_eq!(result.unwrap_err().code, ERROR_INVALID_REQUEST);
}

#[test]
fn test_handle_node_pair_request_with_extended_fields() {
    let state = WsServerState::new(WsServerConfig::default());

    let params = json!({
        "nodeId": "test-node-extended",
        "displayName": "Extended Node",
        "platform": "ios",
        "version": "2.0.0",
        "coreVersion": "1.5.0",
        "uiVersion": "2.0.0-beta",
        "deviceFamily": "iPhone",
        "modelIdentifier": "iPhone14,2",
        "caps": ["audio", "camera", "location"],
        "commands": ["system.run", "camera.snap"],
        "permissions": { "camera": true, "location": false },
        "remoteIp": "192.168.1.100",
        "silent": true
    });

    let result = handle_node_pair_request(Some(&params), &state);
    assert!(result.is_ok());

    let response = result.unwrap();
    assert_eq!(response["status"], "pending");
    assert_eq!(response["created"], true);

    let request = &response["request"];
    assert_eq!(request["nodeId"], "test-node-extended");
    assert_eq!(request["displayName"], "Extended Node");
    assert_eq!(request["platform"], "ios");
    assert_eq!(request["version"], "2.0.0");
    assert_eq!(request["coreVersion"], "1.5.0");
    assert_eq!(request["uiVersion"], "2.0.0-beta");
    assert_eq!(request["deviceFamily"], "iPhone");
    assert_eq!(request["modelIdentifier"], "iPhone14,2");
    assert_eq!(request["caps"], json!(["audio", "camera", "location"]));
    assert_eq!(request["commands"], json!(["system.run", "camera.snap"]));
    assert_eq!(
        request["permissions"],
        json!({ "camera": true, "location": false })
    );
    assert_eq!(request["remoteIp"], "192.168.1.100");
    assert_eq!(request["silent"], true);
    assert_eq!(request["isRepair"], false);
}

#[test]
fn test_handle_node_pair_approve_preserves_extended_fields() {
    let state = WsServerState::new(WsServerConfig::default());

    // Create request with extended fields
    let request_params = json!({
        "nodeId": "test-node-extended-approve",
        "displayName": "Extended Approve Node",
        "platform": "darwin",
        "version": "3.0.0",
        "coreVersion": "2.5.0",
        "uiVersion": "3.0.0",
        "deviceFamily": "Mac",
        "modelIdentifier": "MacBookPro18,3",
        "caps": ["exec", "filesystem"],
        "commands": ["system.run", "file.read"],
        "permissions": { "filesystem": true },
        "remoteIp": "10.0.0.50"
    });
    let request_response = handle_node_pair_request(Some(&request_params), &state).unwrap();
    let request_id = request_response["request"]["requestId"].as_str().unwrap();

    // Approve the request
    let approve_params = json!({ "requestId": request_id });
    let approve_result = handle_node_pair_approve(Some(&approve_params), &state);
    assert!(approve_result.is_ok());

    let node = &approve_result.unwrap()["node"];
    assert_eq!(node["nodeId"], "test-node-extended-approve");
    assert_eq!(node["displayName"], "Extended Approve Node");
    assert_eq!(node["platform"], "darwin");
    assert_eq!(node["version"], "3.0.0");
    assert_eq!(node["coreVersion"], "2.5.0");
    assert_eq!(node["uiVersion"], "3.0.0");
    assert_eq!(node["deviceFamily"], "Mac");
    assert_eq!(node["modelIdentifier"], "MacBookPro18,3");
    assert_eq!(node["caps"], json!(["exec", "filesystem"]));
    assert_eq!(node["permissions"], json!({ "filesystem": true }));
    assert_eq!(node["remoteIp"], "10.0.0.50");
    assert!(node["createdAtMs"].as_u64().is_some());
    assert!(node["approvedAtMs"].as_u64().is_some());
}

#[test]
fn test_handle_node_list_includes_extended_fields() {
    let state = WsServerState::new(WsServerConfig::default());

    // Pair a node with extended fields
    let request_params = json!({
        "nodeId": "test-node-list-extended",
        "displayName": "List Extended Node",
        "platform": "ios",
        "version": "4.0.0",
        "coreVersion": "3.0.0",
        "uiVersion": "4.0.0",
        "deviceFamily": "iPad",
        "modelIdentifier": "iPad13,8",
        "caps": ["audio", "video"],
        "commands": ["media.play"],
        "permissions": { "audio": true },
        "remoteIp": "172.16.0.1"
    });
    let request_response = handle_node_pair_request(Some(&request_params), &state).unwrap();
    let request_id = request_response["request"]["requestId"].as_str().unwrap();
    handle_node_pair_approve(Some(&json!({ "requestId": request_id })), &state).unwrap();

    // List nodes should include extended fields
    let result = handle_node_list(&state);
    assert!(result.is_ok());

    let result_value = result.unwrap();
    let nodes = result_value["nodes"].as_array().unwrap();
    assert_eq!(nodes.len(), 1);

    let node = &nodes[0];
    assert_eq!(node["nodeId"], "test-node-list-extended");
    assert_eq!(node["platform"], "ios");
    assert_eq!(node["version"], "4.0.0");
    assert_eq!(node["coreVersion"], "3.0.0");
    assert_eq!(node["uiVersion"], "4.0.0");
    assert_eq!(node["deviceFamily"], "iPad");
    assert_eq!(node["modelIdentifier"], "iPad13,8");
    assert_eq!(node["remoteIp"], "172.16.0.1");
    assert_eq!(node["paired"], true);
}

#[test]
fn test_handle_node_describe_includes_extended_fields() {
    let state = WsServerState::new(WsServerConfig::default());

    // Pair a node with extended fields
    let request_params = json!({
        "nodeId": "test-node-describe-extended",
        "displayName": "Describe Extended Node",
        "platform": "android",
        "version": "5.0.0",
        "coreVersion": "4.0.0",
        "uiVersion": "5.0.0",
        "deviceFamily": "Pixel",
        "modelIdentifier": "Pixel 7 Pro",
        "caps": ["notification", "sms"],
        "commands": ["notification.send"],
        "permissions": { "notification": true, "sms": false },
        "remoteIp": "192.168.0.200"
    });
    let request_response = handle_node_pair_request(Some(&request_params), &state).unwrap();
    let request_id = request_response["request"]["requestId"].as_str().unwrap();
    handle_node_pair_approve(Some(&json!({ "requestId": request_id })), &state).unwrap();

    // Describe should include extended fields
    let describe_params = json!({ "nodeId": "test-node-describe-extended" });
    let result = handle_node_describe(Some(&describe_params), &state);
    assert!(result.is_ok());

    let node = result.unwrap();
    assert_eq!(node["nodeId"], "test-node-describe-extended");
    assert_eq!(node["platform"], "android");
    assert_eq!(node["version"], "5.0.0");
    assert_eq!(node["coreVersion"], "4.0.0");
    assert_eq!(node["uiVersion"], "5.0.0");
    assert_eq!(node["deviceFamily"], "Pixel");
    assert_eq!(node["modelIdentifier"], "Pixel 7 Pro");
    assert_eq!(node["remoteIp"], "192.168.0.200");
    assert_eq!(node["paired"], true);
    // caps and commands are merged from paired, should be sorted
    assert!(node["caps"].as_array().is_some());
    assert!(node["commands"].as_array().is_some());
}

#[test]
fn test_handle_node_pair_request_is_repair_flag() {
    let state = WsServerState::new(WsServerConfig::default());

    // First pairing - should not be a repair
    let params1 = json!({
        "nodeId": "test-node-repair",
        "displayName": "Repair Test Node"
    });
    let result1 = handle_node_pair_request(Some(&params1), &state).unwrap();
    assert_eq!(result1["request"]["isRepair"], false);

    // Approve the first request
    let request_id = result1["request"]["requestId"].as_str().unwrap();
    handle_node_pair_approve(Some(&json!({ "requestId": request_id })), &state).unwrap();

    // Second pairing request - should be a repair since node is already paired
    let params2 = json!({
        "nodeId": "test-node-repair",
        "displayName": "Repair Test Node Updated"
    });
    let result2 = handle_node_pair_request(Some(&params2), &state).unwrap();
    assert_eq!(result2["request"]["isRepair"], true);
}

#[test]
fn test_handle_node_pair_verify_includes_extended_fields() {
    let state = WsServerState::new(WsServerConfig::default());

    // Create and approve request with extended fields
    let request_params = json!({
        "nodeId": "test-node-verify-extended",
        "displayName": "Verify Extended Node",
        "platform": "linux",
        "version": "1.0.0",
        "coreVersion": "1.0.0",
        "deviceFamily": "Server",
        "caps": ["exec"],
        "commands": ["system.run"]
    });
    let request_response = handle_node_pair_request(Some(&request_params), &state).unwrap();
    let request_id = request_response["request"]["requestId"].as_str().unwrap();
    let approve_response =
        handle_node_pair_approve(Some(&json!({ "requestId": request_id })), &state).unwrap();
    let token = approve_response["node"]["token"].as_str().unwrap();

    // Verify should include extended fields in the node response
    let verify_params = json!({
        "nodeId": "test-node-verify-extended",
        "token": token
    });
    let result = handle_node_pair_verify(Some(&verify_params), &state);
    assert!(result.is_ok());

    let response = result.unwrap();
    assert_eq!(response["ok"], true);
    let node = &response["node"];
    assert_eq!(node["nodeId"], "test-node-verify-extended");
    assert_eq!(node["platform"], "linux");
    assert_eq!(node["version"], "1.0.0");
    assert_eq!(node["coreVersion"], "1.0.0");
    assert_eq!(node["deviceFamily"], "Server");
}

// ============== Event System Tests ==============

#[test]
fn test_state_version_tracking() {
    let state = WsServerState::new(WsServerConfig::default());

    // Initial state version should be 0,0
    let version = state.current_state_version();
    assert_eq!(version.presence, 0);
    assert_eq!(version.health, 0);

    // Create a connection to trigger presence update
    let (tx, _rx) = mpsc::channel(256);
    let conn = make_conn_with_id("operator", vec![], "conn-1");
    state.register_connection(&conn, tx, None);

    // Presence version should be incremented
    let version = state.current_state_version();
    assert_eq!(version.presence, 1);
    assert_eq!(version.health, 0);

    // Unregister should also increment presence version
    state.unregister_connection("conn-1");
    let version = state.current_state_version();
    assert_eq!(version.presence, 2);
    assert_eq!(version.health, 0);
}

#[test]
fn test_state_version_and_event_seq_allocate_monotonically_together() {
    let state = WsServerState::new(WsServerConfig::default());

    let (presence_seq_1, presence_version_1) = state.next_presence_event_ordering();
    let unrelated_seq = state.next_event_seq();
    let (presence_seq_2, presence_version_2) = state.next_presence_event_ordering();

    assert!(presence_seq_1 < unrelated_seq);
    assert!(unrelated_seq < presence_seq_2);
    assert!(presence_version_1.presence < presence_version_2.presence);
    assert!(
        presence_seq_2 > presence_seq_1,
        "later stateVersion allocations must receive later event seq values"
    );
}

#[test]
fn test_state_versioned_presence_broadcasts_enqueue_in_seq_order_under_contention() {
    const WORKERS: usize = 32;

    let state = Arc::new(WsServerState::new(WsServerConfig::default()));
    let (tx, mut rx) = mpsc::channel(128);
    let conn = make_conn_with_id("admin", vec![], "ordered-presence-admin");
    state.register_connection(&conn, tx, None);
    while rx.try_recv().is_ok() {}

    let barrier = Arc::new(std::sync::Barrier::new(WORKERS));
    let mut workers = Vec::new();
    for _ in 0..WORKERS {
        let state = Arc::clone(&state);
        let barrier = Arc::clone(&barrier);
        workers.push(std::thread::spawn(move || {
            barrier.wait();
            state.broadcast_next_presence_event();
        }));
    }
    for worker in workers {
        worker.join().expect("presence broadcast worker panicked");
    }

    let mut frames = Vec::new();
    while let Ok(message) = rx.try_recv() {
        let Message::Text(text) = message else {
            panic!("expected text presence broadcast");
        };
        let frame: Value = serde_json::from_str(text.as_str()).unwrap();
        assert_eq!(frame["event"], "presence");
        frames.push((
            frame["seq"].as_u64().unwrap(),
            frame["stateVersion"]["presence"].as_u64().unwrap(),
        ));
    }

    assert_eq!(frames.len(), WORKERS);
    for window in frames.windows(2) {
        assert!(
            window[0].0 < window[1].0,
            "wire enqueue order must follow allocated event seq"
        );
        assert!(
            window[0].1 < window[1].1,
            "wire enqueue order must follow allocated presence stateVersion"
        );
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_system_event_presence_broadcasts_allocate_under_ordering_lock() {
    const WORKERS: usize = 16;

    let state = Arc::new(WsServerState::new(WsServerConfig::default()));
    let (tx, mut rx) = mpsc::channel(128);
    let admin = make_conn_with_id("admin", vec![], "system-event-admin");
    state.register_connection(&admin, tx, None);
    while rx.try_recv().is_ok() {}

    let barrier = Arc::new(tokio::sync::Barrier::new(WORKERS));
    let mut workers = Vec::new();
    for index in 0..WORKERS {
        let state = Arc::clone(&state);
        let conn = admin.clone();
        let barrier = Arc::clone(&barrier);
        workers.push(tokio::spawn(async move {
            barrier.wait().await;
            let params = json!({ "text": format!("host-{index} (127.0.0.1) reason heartbeat") });
            handlers::dispatch_method("system-event", Some(&params), &state, &conn)
                .await
                .expect("system-event should succeed");
        }));
    }
    for worker in workers {
        worker.await.expect("system-event worker panicked");
    }

    let mut frames = Vec::new();
    while let Ok(message) = rx.try_recv() {
        let Message::Text(text) = message else {
            panic!("expected text presence broadcast");
        };
        let frame: Value = serde_json::from_str(text.as_str()).unwrap();
        if frame["event"] == "presence" {
            frames.push((
                frame["seq"].as_u64().unwrap(),
                frame["stateVersion"]["presence"].as_u64().unwrap(),
            ));
        }
    }

    assert_eq!(frames.len(), WORKERS);
    for window in frames.windows(2) {
        assert!(window[0].0 < window[1].0);
        assert!(window[0].1 < window[1].1);
    }
}

#[test]
fn test_state_drop_wire_shape_is_pinned() {
    let state = WsServerState::new(WsServerConfig::default());
    let (seq, state_version) = state.next_presence_event_ordering();
    let frame = serialize_state_drop_marker_event(
        &state,
        "presence",
        StateDropPayloadClass::Admin,
        seq,
        state_version,
    )
    .expect("state.drop frame should serialize");
    let value: Value = serde_json::from_str(&frame).unwrap();

    assert_eq!(value["type"], "event");
    assert_eq!(value["event"], "state.drop");
    assert_eq!(value["seq"], seq);
    assert_eq!(value["payload"]["dropped"], true);
    assert_eq!(value["payload"]["event"], "presence");
    assert_eq!(value["payload"]["payloadClass"], "admin");
    assert_eq!(value["payload"]["reason"], "payload_too_large");
    assert_eq!(value["payload"]["reasonTruncated"], false);
    assert_eq!(value["payload"]["resyncRequired"], true);
    assert!(value["payload"]["ts"].as_i64().is_some());
    let payload = value["payload"].as_object().unwrap();
    let mut keys: Vec<_> = payload.keys().map(String::as_str).collect();
    keys.sort_unstable();
    assert_eq!(
        keys,
        [
            "dropped",
            "event",
            "payloadClass",
            "reason",
            "reasonTruncated",
            "resyncRequired",
            "ts"
        ]
    );
    assert!(value["stateVersion"]["presence"].as_u64().is_some());
}

fn ws_golden_trace(name: &str) -> Value {
    let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("tests")
        .join("golden")
        .join("ws")
        .join(name);
    let content = std::fs::read_to_string(&path)
        .unwrap_or_else(|err| panic!("read WS golden trace {}: {err}", path.display()));
    serde_json::from_str(&content)
        .unwrap_or_else(|err| panic!("parse WS golden trace {}: {err}", path.display()))
}

fn sorted_object_keys(value: &Value) -> Vec<String> {
    let mut keys: Vec<String> = value
        .as_object()
        .expect("expected JSON object")
        .keys()
        .cloned()
        .collect();
    keys.sort();
    keys
}

/// Mirror of `test_state_drop_runtime_shape_matches_ws_golden_schema`
/// for the shutdown event. The reasonTruncated flag is the operator's
/// only signal that the wire value was silently clipped by the
/// WS_SHUTDOWN_REASON_MAX_CHARS cap; if it ever leaves the payload
/// (or the golden), clients can no longer distinguish a verbatim
/// short reason from a multi-megabyte reason cut at the cap.
#[test]
fn test_shutdown_runtime_shape_matches_ws_golden_schema() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx, mut rx) = mpsc::channel(8);
    let conn = make_conn_with_id("admin", vec![], "shutdown-shape-conn");
    state.register_connection(&conn, tx, None);
    // Drain the initial presence frame so the only message left is the
    // shutdown frame we are about to broadcast.
    while rx.try_recv().is_ok() {}

    broadcast_shutdown(&state, "scheduled-update", Some(5000));
    let Message::Text(text) = rx
        .try_recv()
        .expect("shutdown broadcast should deliver to the admin connection")
    else {
        panic!("expected text message");
    };
    let frame: Value = serde_json::from_str(&text).expect("shutdown frame parses as JSON");
    assert_eq!(frame["event"], "shutdown");
    let runtime_payload = frame.get("payload").expect("shutdown payload");

    let golden = ws_golden_trace("events.json");
    let schema = &golden["events"]["shutdown"]["payload_schema"];
    let mut golden_required: Vec<String> = schema["required"]
        .as_array()
        .expect("shutdown required fields")
        .iter()
        .map(|value| value.as_str().expect("required field").to_string())
        .collect();
    golden_required.sort();
    let mut runtime_keys = sorted_object_keys(runtime_payload);
    runtime_keys.retain(|key| golden_required.contains(key));
    assert_eq!(runtime_keys, golden_required);
    // Required and properties must stay in lockstep on the schema side
    // so a future contributor cannot add a `required` entry without
    // also documenting its type/shape.
    let golden_property_keys = sorted_object_keys(&schema["properties"]);
    for key in &golden_required {
        assert!(
            golden_property_keys.contains(key),
            "shutdown golden required field {key:?} must have a matching property entry"
        );
    }
}

/// `reasonTruncated` must be `false` when the operator-supplied reason
/// fits inside `WS_SHUTDOWN_REASON_MAX_CHARS` and `true` when the cap
/// silently clips the wire value. Without this round-trip pin the
/// flag is structurally present in the payload but semantically
/// meaningless — exactly the R56-era state.drop-misfire shape this
/// fix is supposed to retire.
#[test]
fn test_shutdown_reason_truncated_flag_reflects_truncation() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx, mut rx) = mpsc::channel(8);
    let conn = make_conn_with_id("admin", vec![], "shutdown-truncate-conn");
    state.register_connection(&conn, tx, None);
    while rx.try_recv().is_ok() {}

    broadcast_shutdown(&state, "short", None);
    let Message::Text(text) = rx.try_recv().expect("short reason should broadcast") else {
        panic!("expected text message");
    };
    let frame: Value = serde_json::from_str(&text).expect("frame parses");
    assert_eq!(
        frame["payload"]["reasonTruncated"],
        serde_json::json!(false)
    );
    assert_eq!(frame["payload"]["reason"], serde_json::json!("short"));

    // Long reason: provide one full cap + 1 extra character.
    let oversize_reason: String = "x".repeat(WS_SHUTDOWN_REASON_MAX_CHARS + 1);
    broadcast_shutdown(&state, &oversize_reason, None);
    let Message::Text(text) = rx
        .try_recv()
        .expect("oversize reason should still broadcast (clipped, not dropped)")
    else {
        panic!("expected text message");
    };
    let frame: Value = serde_json::from_str(&text).expect("frame parses");
    assert_eq!(frame["payload"]["reasonTruncated"], serde_json::json!(true));
    let payload_reason = frame["payload"]["reason"]
        .as_str()
        .expect("reason must be a string");
    assert_eq!(payload_reason.chars().count(), WS_SHUTDOWN_REASON_MAX_CHARS);
}

/// Defuse R57-H14: when an oversized `matrix.verification.requested`
/// event falls back to a `state.drop` marker, the marker frame must
/// only reach connections that satisfied the original event's
/// `operator.admin` scope. A naive reading of the wire (the outer
/// frame `event` field is `"state.drop"`) suggests the marker would
/// broadcast to all non-node connections, but the actual fan-out path
/// queries `event_required_scope` with the ORIGINAL event name, so
/// admin-scope event names never leak to non-admin operators. Pin the
/// invariant here so a future refactor that loses the original-event
/// scope routing fails fast.
#[test]
fn test_state_drop_for_admin_scope_event_does_not_reach_non_admin() {
    let state = WsServerState::new(WsServerConfig::default());

    let (admin_tx, mut admin_rx) = mpsc::channel(32);
    let admin_conn = make_conn_with_id("admin", vec![], "state-drop-admin-conn");
    state.register_connection(&admin_conn, admin_tx, None);

    let (operator_tx, mut operator_rx) = mpsc::channel(32);
    let operator_conn = make_conn_with_id(
        "operator",
        vec!["operator.approvals".to_string()],
        "state-drop-non-admin-conn",
    );
    state.register_connection(&operator_conn, operator_tx, None);
    while admin_rx.try_recv().is_ok() {}
    while operator_rx.try_recv().is_ok() {}

    // Build a `matrix.verification.requested` payload large enough to
    // force the state.drop fallback. The display-name field is
    // operator-controlled and not validated against the cap, so it is
    // the simplest way to overshoot WS_BROADCAST_PAYLOAD_MAX_BYTES.
    let huge_field = "x".repeat(WS_BROADCAST_PAYLOAD_MAX_BYTES + 1);
    let payload = json!({
        "device_id": huge_field,
        "user_id": "@admin:example.org",
        "flow_id": "flow-state-drop",
        "method": "m.sas.v1",
    });
    broadcast_event(&state, "matrix.verification.requested", payload);

    let Message::Text(admin_text) = admin_rx
        .try_recv()
        .expect("admin connection must receive the state.drop marker")
    else {
        panic!("expected text message");
    };
    let admin_frame: Value = serde_json::from_str(&admin_text).expect("admin frame parses as JSON");
    assert_eq!(
        admin_frame["event"], "state.drop",
        "admin receives the state.drop marker for an oversized admin-scope event"
    );
    assert_eq!(
        admin_frame["payload"]["event"], "matrix.verification.requested",
        "marker payload preserves the original event name"
    );

    assert!(
        operator_rx.try_recv().is_err(),
        "non-admin operator must NOT receive the state.drop marker for an admin-scope event"
    );
}

#[test]
fn test_state_drop_runtime_shape_matches_ws_golden_schema() {
    let state = WsServerState::new(WsServerConfig::default());
    let (seq, state_version) = state.next_presence_event_ordering();
    let frame = serialize_state_drop_marker_event(
        &state,
        "presence",
        StateDropPayloadClass::Admin,
        seq,
        state_version,
    )
    .expect("state.drop frame should serialize");
    let value: Value = serde_json::from_str(&frame).unwrap();
    let runtime_payload = value.get("payload").expect("state.drop payload");

    // Structural round-trip: deserializing the runtime payload back
    // into the typed `StateDropPayload` struct asserts every field
    // name AND every field type (`bool`, `i64`, enums for
    // `payloadClass`/`reason`). A future change from `bool` to
    // `Option<bool>`, an unknown payloadClass value, or an unknown
    // reason all fail here — the pre-H12 key-set comparison would
    // have silently let those through.
    let typed: StateDropPayload = serde_json::from_value(runtime_payload.clone()).expect(
        "runtime state.drop payload must deserialize into the typed StateDropPayload \
             struct (deny_unknown_fields rejects shape drift)",
    );
    let reserialized =
        serde_json::to_value(&typed).expect("typed payload must re-serialize back to JSON");
    assert_eq!(
        reserialized,
        runtime_payload.clone(),
        "typed serialize/deserialize round-trip must produce byte-identical JSON"
    );

    let golden = ws_golden_trace("events.json");
    let schema = &golden["events"]["state.drop"]["payload_schema"];
    let mut golden_required: Vec<String> = schema["required"]
        .as_array()
        .expect("state.drop required fields")
        .iter()
        .map(|value| value.as_str().expect("required field").to_string())
        .collect();
    golden_required.sort();
    assert_eq!(sorted_object_keys(runtime_payload), golden_required);
    assert_eq!(
        sorted_object_keys(&schema["properties"]),
        golden_required,
        "state.drop golden properties and required fields must stay in lockstep"
    );

    // The golden example must itself deserialize into the typed
    // struct so a future contributor cannot widen the golden's
    // accepted values past what the runtime can produce.
    let golden_example = &golden["events"]["state.drop"]["example"]["payload"];
    serde_json::from_value::<StateDropPayload>(golden_example.clone())
        .expect("state.drop golden example payload must match the typed StateDropPayload contract");
}

/// Pin the closed set of `payloadClass` values and the
/// `deny_unknown_fields` posture so a new payload-class value cannot
/// land in production without an explicit Rust enum addition.
#[test]
fn test_state_drop_payload_class_rejects_unknown_values() {
    use serde_json::json;
    let valid: StateDropPayload = serde_json::from_value(json!({
        "dropped": true,
        "event": "presence",
        "payloadClass": "admin",
        "reason": "payload_too_large",
        "reasonTruncated": false,
        "resyncRequired": true,
        "ts": 0_u64,
    }))
    .expect("admin payload class must parse");
    assert_eq!(valid.payload_class, StateDropPayloadClass::Admin);

    for class in ["operator", "broadcast"] {
        let parsed: StateDropPayload = serde_json::from_value(json!({
            "dropped": true,
            "event": "presence",
            "payloadClass": class,
            "reason": "payload_too_large",
            "reasonTruncated": false,
            "resyncRequired": true,
            "ts": 0_u64,
        }))
        .expect("known payload class must parse");
        match (class, parsed.payload_class) {
            ("operator", StateDropPayloadClass::Operator)
            | ("broadcast", StateDropPayloadClass::Broadcast) => {}
            (other, got) => panic!("unexpected payload class {other:?} -> {got:?}"),
        }
    }

    // Unknown payloadClass must be rejected.
    let bad = json!({
        "dropped": true,
        "event": "presence",
        "payloadClass": "future-class",
        "reason": "payload_too_large",
        "reasonTruncated": false,
        "resyncRequired": true,
        "ts": 0_u64,
    });
    assert!(
        serde_json::from_value::<StateDropPayload>(bad).is_err(),
        "unknown payloadClass values must fail deserialization"
    );

    // Unknown reason must be rejected.
    let bad = json!({
        "dropped": true,
        "event": "presence",
        "payloadClass": "admin",
        "reason": "future-reason",
        "reasonTruncated": false,
        "resyncRequired": true,
        "ts": 0_u64,
    });
    assert!(
        serde_json::from_value::<StateDropPayload>(bad).is_err(),
        "unknown reason values must fail deserialization"
    );

    // Extra fields must be rejected (deny_unknown_fields).
    let bad = json!({
        "dropped": true,
        "event": "presence",
        "payloadClass": "admin",
        "reason": "payload_too_large",
        "reasonTruncated": false,
        "resyncRequired": true,
        "ts": 0_u64,
        "futureField": "anything",
    });
    assert!(
        serde_json::from_value::<StateDropPayload>(bad).is_err(),
        "unknown fields must fail deserialization (deny_unknown_fields)"
    );
}

#[test]
fn test_hello_policy_runtime_shape_matches_ws_golden_example() {
    let state = WsServerState::new(WsServerConfig::default());
    let payload = serde_json::to_value(build_hello_response(&state, "conn-1", None))
        .expect("hello-ok payload should serialize");
    let runtime_policy = payload.get("policy").expect("hello-ok policy");

    let golden = ws_golden_trace("handshake.json");
    let scenarios = golden["scenarios"].as_array().expect("handshake scenarios");
    let local_success = scenarios
        .iter()
        .find(|scenario| scenario["name"] == "connect_success_local_with_token")
        .expect("local success scenario");
    let receive_step = local_success["steps"]
        .as_array()
        .expect("scenario steps")
        .iter()
        .find(|step| {
            step["action"] == "receive" && step["expected"]["payload"]["type"] == "hello-ok"
        })
        .expect("hello-ok receive step");
    let golden_policy = &receive_step["expected"]["payload"]["policy"];

    assert_eq!(
        sorted_object_keys(runtime_policy),
        sorted_object_keys(golden_policy)
    );
    assert_eq!(
        runtime_policy["maxBroadcastPayload"],
        serde_json::json!(WS_BROADCAST_PAYLOAD_MAX_BYTES)
    );
    assert_eq!(
        golden_policy["maxBroadcastPayload"],
        serde_json::json!(WS_BROADCAST_PAYLOAD_MAX_BYTES)
    );
}

#[test]
fn test_oversized_presence_broadcast_emits_state_drop_wire_event() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx, mut rx) = mpsc::channel(256);
    let mut conn = make_conn_with_id("admin", vec![], "oversized-presence");
    conn.client.display_name = Some("x".repeat(WS_BROADCAST_PAYLOAD_MAX_BYTES + 1));

    state.register_connection(&conn, tx, None);

    let Message::Text(text) = rx
        .try_recv()
        .expect("oversized presence should send a state.drop marker")
    else {
        panic!("expected text message");
    };
    let value: Value = serde_json::from_str(&text).expect("state.drop frame must deserialize");
    assert_eq!(value["event"], "state.drop");
    assert_eq!(value["payload"]["event"], "presence");
    assert_eq!(value["payload"]["payloadClass"], "admin");
    assert_eq!(value["payload"]["reason"], "payload_too_large");
    assert_eq!(value["payload"]["resyncRequired"], true);
    assert!(value["stateVersion"]["presence"].as_u64().is_some());
}

#[test]
fn test_set_matrix_runtime_second_install_preserves_existing_runtime() {
    let state = WsServerState::new(WsServerConfig::default());
    let first = crate::channels::matrix::MatrixRuntimeHandle::for_test();
    let second = crate::channels::matrix::MatrixRuntimeHandle::for_test();

    state
        .set_matrix_runtime(Some(first.clone()))
        .expect("first runtime install should succeed");
    let rejected = state
        .set_matrix_runtime(Some(second.clone()))
        .expect_err("second runtime install must be rejected");

    assert!(Arc::ptr_eq(&rejected, &second));
    assert!(Arc::ptr_eq(
        &state
            .matrix_runtime()
            .expect("existing runtime should remain"),
        &first
    ));
}

#[test]
fn test_presence_tracking() {
    let state = WsServerState::new(WsServerConfig::default());

    // Initially empty
    let presence = state.get_presence_list();
    assert!(presence.is_empty());

    // Register a connection
    let (tx, _rx) = mpsc::channel(256);
    let conn = ConnectionContext {
        conn_id: "conn-1".to_string(),
        role: "operator".to_string(),
        scopes: vec!["operator.admin".to_string()],
        client: ClientInfo {
            id: "test-client".to_string(),
            version: "1.0.0".to_string(),
            platform: "darwin".to_string(),
            mode: "ui".to_string(),
            display_name: Some("Test Mac".to_string()),
            device_family: Some("MacBookPro".to_string()),
            model_identifier: Some("Mac14,5".to_string()),
            instance_id: Some("inst-1".to_string()),
        },
        device_id: Some("device-1".to_string()),
    };
    state.register_connection(&conn, tx, Some("192.168.1.100".to_string()));

    // Should now have one presence entry
    let presence = state.get_presence_list();
    assert_eq!(presence.len(), 1);

    let entry = &presence[0];
    assert_eq!(entry["host"], "Test Mac");
    assert!(entry.get("ip").is_none());
    assert_eq!(entry["version"], "1.0.0");
    assert_eq!(entry["platform"], "darwin");
    assert_eq!(entry["mode"], "ui");
    assert_eq!(entry["reason"], "connect");
    assert!(entry.get("deviceFamily").is_none());
    assert!(entry.get("modelIdentifier").is_none());
    assert!(entry.get("deviceId").is_none());
    assert!(entry.get("roles").is_none());
    assert!(entry.get("scopes").is_none());
    assert!(entry.get("instanceId").is_none());
    // connId and clientId are internal fields, not serialized per Node schema
    assert!(entry.get("connId").is_none() || entry["connId"].is_null());
    assert!(entry.get("clientId").is_none() || entry["clientId"].is_null());

    let admin_presence = state.get_presence_list_for_conn("conn-1");
    let admin_entry = &admin_presence[0];
    assert_eq!(admin_entry["ip"], "192.168.1.100");
    assert_eq!(admin_entry["deviceId"], "device-1");
    assert_eq!(admin_entry["roles"][0], "operator");
    assert_eq!(admin_entry["scopes"][0], "operator.admin");
    assert_eq!(admin_entry["instanceId"], "inst-1");
    // deviceFamily and modelIdentifier are admin-only hardware
    // identifiers (Node PresenceEntrySchema parity) — non-admin
    // path strips them (assertions above), admin path includes them.
    assert_eq!(admin_entry["deviceFamily"], "MacBookPro");
    assert_eq!(admin_entry["modelIdentifier"], "Mac14,5");

    // Unregister should remove presence
    state.unregister_connection("conn-1");
    let presence = state.get_presence_list();
    assert!(presence.is_empty());
}

#[test]
fn test_health_snapshot_cache() {
    let state = WsServerState::new(WsServerConfig::default());

    // Initial health should be "healthy"
    let health = state.get_health_snapshot();
    assert_eq!(health.status, "healthy");
    assert!(health.ts > 0);

    // Update health
    state.update_health("degraded", Some(json!({"whatsapp": "disconnected"})), None);

    let health = state.get_health_snapshot();
    assert_eq!(health.status, "degraded");
    assert_eq!(health.channels, Some(json!({"whatsapp": "disconnected"})));

    // State version should be incremented for health
    let version = state.current_state_version();
    assert_eq!(version.health, 1);
}

#[test]
fn test_presence_broadcast_on_connect() {
    let state = WsServerState::new(WsServerConfig::default());

    // Register first connection (will receive broadcasts)
    let (tx1, mut rx1) = mpsc::channel(256);
    let conn1 = make_conn_with_id("operator", vec!["operator.admin".to_string()], "conn-1");
    state.register_connection(&conn1, tx1, None);

    // First connection should receive presence update about itself
    let msg = rx1.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["type"], "event");
    assert_eq!(event["event"], "presence");
    assert!(event["seq"].as_u64().is_some());
    assert!(event["stateVersion"]["presence"].as_u64().unwrap() >= 1);
    assert!(!event["payload"]["presence"].as_array().unwrap().is_empty());

    // Register second connection
    let (tx2, _rx2) = mpsc::channel(256);
    let conn2 = make_conn_with_id("operator", vec![], "conn-2");
    state.register_connection(&conn2, tx2, None);

    // First connection should receive another presence update
    let msg = rx1.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "presence");
    assert_eq!(event["payload"]["presence"].as_array().unwrap().len(), 2);
}

#[test]
fn test_presence_broadcast_excludes_nodes() {
    let state = WsServerState::new(WsServerConfig::default());

    // Register a node connection
    let (tx_node, mut rx_node) = mpsc::channel(256);
    let node_conn = make_conn_with_id("node", vec![], "node-conn");
    state.register_connection(&node_conn, tx_node, None);

    // Node should not receive presence broadcast
    let msg = rx_node.try_recv();
    assert!(msg.is_err(), "Node should not receive presence broadcasts");

    // Register an operator
    let (tx_op, mut rx_op) = mpsc::channel(256);
    let op_conn = make_conn_with_id("operator", vec![], "op-conn");
    state.register_connection(&op_conn, tx_op, None);

    // Operator should receive presence broadcast
    let msg = rx_op.try_recv();
    assert!(msg.is_ok(), "Operator should receive presence broadcasts");

    // Node still should not have received anything
    let msg = rx_node.try_recv();
    assert!(msg.is_err(), "Node should not receive presence broadcasts");
}

#[test]
fn test_broadcast_agent_event() {
    let state = WsServerState::new(WsServerConfig::default());

    let (tx, mut rx) = mpsc::channel(256);
    let conn = make_conn_with_id("operator", vec![], "conn-1");
    state.register_connection(&conn, tx, None);

    // Clear the presence broadcast
    let _ = rx.try_recv();

    // Broadcast an agent event
    broadcast_agent_event(&state, "run-123", 1, "text", json!({"text": "Hello!"}));

    let msg = rx.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["type"], "event");
    assert_eq!(event["event"], "agent");
    assert_eq!(event["payload"]["runId"], "run-123");
    assert_eq!(event["payload"]["seq"], 1);
    assert_eq!(event["payload"]["stream"], "text");
    assert_eq!(event["payload"]["data"]["text"], "Hello!");
    assert!(event["payload"]["ts"].as_u64().is_some());
}

#[test]
fn test_broadcast_chat_event() {
    let state = WsServerState::new(WsServerConfig::default());

    let (tx, mut rx) = mpsc::channel(256);
    let conn = make_conn_with_id("operator", vec![], "conn-1");
    state.register_connection(&conn, tx, None);

    // Clear the presence broadcast
    let _ = rx.try_recv();

    // Broadcast a chat event
    broadcast_chat_event(
        &state,
        "msg-123",
        "main",
        5,
        "final",
        Some(json!({"role": "assistant", "content": "Hello!"})),
        None,
        Some(json!({"input": 10, "output": 20})),
        Some("end_turn"),
    );

    let msg = rx.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "chat");
    assert_eq!(event["payload"]["runId"], "msg-123");
    assert_eq!(event["payload"]["sessionKey"], "main");
    assert_eq!(event["payload"]["seq"], 5);
    assert_eq!(event["payload"]["state"], "final");
    assert_eq!(event["payload"]["message"]["role"], "assistant");
    assert_eq!(event["payload"]["usage"]["input"], 10);
    assert_eq!(event["payload"]["stopReason"], "end_turn");
}

#[test]
fn test_broadcast_cron_event() {
    let state = WsServerState::new(WsServerConfig::default());

    let (tx, mut rx) = mpsc::channel(256);
    let conn = make_conn_with_id("operator", vec![], "conn-1");
    state.register_connection(&conn, tx, None);

    // Clear the presence broadcast
    let _ = rx.try_recv();

    // Broadcast a cron event
    broadcast_cron_event(
        &state,
        "job-1",
        "completed",
        Some("run-456"),
        Some(json!({"ok": true})),
    );

    let msg = rx.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "cron");
    assert_eq!(event["payload"]["jobId"], "job-1");
    assert_eq!(event["payload"]["status"], "completed");
    assert_eq!(event["payload"]["runId"], "run-456");
    assert_eq!(event["payload"]["result"]["ok"], true);
}

#[test]
fn test_broadcast_voicewake_changed() {
    let state = WsServerState::new(WsServerConfig::default());

    // voicewake.changed should go to both operators AND nodes
    let (tx_op, mut rx_op) = mpsc::channel(256);
    let op_conn = make_conn_with_id("operator", vec![], "op-conn");
    state.register_connection(&op_conn, tx_op, None);

    let (tx_node, mut rx_node) = mpsc::channel(256);
    let node_conn = make_conn_with_id("node", vec![], "node-conn");
    state.register_connection(&node_conn, tx_node, None);

    // Clear presence broadcasts for operator
    let _ = rx_op.try_recv();
    let _ = rx_op.try_recv();

    // Broadcast voicewake change
    broadcast_voicewake_changed(
        &state,
        vec!["hey claude".to_string(), "ok claude".to_string()],
    );

    // Operator should receive it
    let msg = rx_op.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "voicewake.changed");
    assert_eq!(
        event["payload"]["triggers"],
        json!(["hey claude", "ok claude"])
    );

    // Node should also receive it
    let msg = rx_node.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "voicewake.changed");
}

#[test]
fn test_broadcast_exec_approval_events() {
    let state = WsServerState::new(WsServerConfig::default());

    // Create connection with approvals scope
    let (tx_with_scope, mut rx_with_scope) = mpsc::channel(256);
    let conn_with_scope = make_conn_with_id(
        "operator",
        vec!["operator.approvals".to_string()],
        "conn-with-scope",
    );
    state.register_connection(&conn_with_scope, tx_with_scope, None);

    // Create connection without approvals scope
    let (tx_without_scope, mut rx_without_scope) = mpsc::channel(256);
    let conn_without_scope = make_conn_with_id("operator", vec![], "conn-without-scope");
    state.register_connection(&conn_without_scope, tx_without_scope, None);

    // Clear presence broadcasts
    let _ = rx_with_scope.try_recv();
    let _ = rx_with_scope.try_recv();
    let _ = rx_without_scope.try_recv();
    let _ = rx_without_scope.try_recv();

    // Broadcast exec approval requested
    broadcast_exec_approval_requested(
        &state,
        "req-1",
        "rm",
        vec!["-rf".to_string(), "/tmp/test".to_string()],
        Some("/home/user"),
        Some("agent-1"),
        Some("main"),
    );

    // Connection with scope should receive it
    let msg = rx_with_scope.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "exec.approval.requested");
    assert_eq!(event["payload"]["requestId"], "req-1");
    assert_eq!(event["payload"]["command"], "rm");
    assert_eq!(event["payload"]["args"], json!(["-rf", "/tmp/test"]));
    assert_eq!(event["payload"]["cwd"], "/home/user");
    assert_eq!(event["payload"]["agentId"], "agent-1");
    assert_eq!(event["payload"]["sessionKey"], "main");

    // Connection without scope should NOT receive it
    let msg = rx_without_scope.try_recv();
    assert!(
        msg.is_err(),
        "Connection without scope should not receive exec.approval events"
    );

    // Broadcast exec approval resolved
    broadcast_exec_approval_resolved(&state, "req-1", "approved");

    // Connection with scope should receive it
    let msg = rx_with_scope.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "exec.approval.resolved");
    assert_eq!(event["payload"]["requestId"], "req-1");
    assert_eq!(event["payload"]["decision"], "approved");

    // Connection without scope should NOT receive it
    let msg = rx_without_scope.try_recv();
    assert!(msg.is_err());
}

#[test]
fn test_broadcast_shutdown() {
    let state = WsServerState::new(WsServerConfig::default());

    // Shutdown goes to ALL connections including nodes
    let (tx_op, mut rx_op) = mpsc::channel(256);
    let op_conn = make_conn_with_id("operator", vec![], "op-conn");
    state.register_connection(&op_conn, tx_op, None);

    let (tx_node, mut rx_node) = mpsc::channel(256);
    let node_conn = make_conn_with_id("node", vec![], "node-conn");
    state.register_connection(&node_conn, tx_node, None);

    // Clear presence broadcasts
    let _ = rx_op.try_recv();
    let _ = rx_op.try_recv();

    // Broadcast shutdown
    broadcast_shutdown(&state, "update", Some(5000));

    // Operator should receive it
    let msg = rx_op.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "shutdown");
    assert_eq!(event["payload"]["reason"], "update");
    assert_eq!(event["payload"]["restartExpectedMs"], 5000);

    // Node should also receive it
    let msg = rx_node.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "shutdown");
}

#[test]
fn test_broadcast_shutdown_preserves_shutdown_event_for_oversized_reason() {
    let state = WsServerState::new(WsServerConfig::default());
    let (tx, mut rx) = mpsc::channel(256);
    let admin = make_conn_with_id("admin", vec![], "shutdown-oversized-reason");
    state.register_connection(&admin, tx, None);
    while rx.try_recv().is_ok() {}

    broadcast_shutdown(
        &state,
        &"x".repeat(WS_BROADCAST_PAYLOAD_MAX_BYTES + 1),
        None,
    );

    let Message::Text(text) = rx
        .try_recv()
        .expect("shutdown should remain a shutdown event after truncation")
    else {
        panic!("expected shutdown text message");
    };
    let event: Value = serde_json::from_str(&text).expect("shutdown frame must deserialize");
    assert_eq!(event["event"], "shutdown");
    assert_eq!(
        event["payload"]["reason"].as_str().unwrap().len(),
        WS_SHUTDOWN_REASON_MAX_CHARS
    );
}

#[test]
fn test_broadcast_heartbeat() {
    let state = WsServerState::new(WsServerConfig::default());

    let (tx, mut rx) = mpsc::channel(256);
    let conn = make_conn_with_id("operator", vec![], "conn-1");
    state.register_connection(&conn, tx, None);

    // Clear presence broadcast
    let _ = rx.try_recv();

    broadcast_heartbeat(&state);

    let msg = rx.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "heartbeat");
    assert!(event["payload"]["ts"].as_u64().is_some());
}

#[test]
fn test_broadcast_talk_mode() {
    let state = WsServerState::new(WsServerConfig::default());

    let (tx, mut rx) = mpsc::channel(256);
    let conn = make_conn_with_id("operator", vec![], "conn-1");
    state.register_connection(&conn, tx, None);

    // Clear presence broadcast
    let _ = rx.try_recv();

    broadcast_talk_mode(&state, true, Some("whatsapp"));

    let msg = rx.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "talk.mode");
    assert_eq!(event["payload"]["enabled"], true);
    assert_eq!(event["payload"]["channel"], "whatsapp");
}

#[test]
fn test_event_seq_increments() {
    let state = WsServerState::new(WsServerConfig::default());

    let (tx, mut rx) = mpsc::channel(256);
    let conn = make_conn_with_id("operator", vec![], "conn-1");
    state.register_connection(&conn, tx, None);

    // Get seq from presence event
    let msg = rx.try_recv().unwrap();
    let Message::Text(text) = msg else { panic!() };
    let event: Value = serde_json::from_str(&text).unwrap();
    let seq1 = event["seq"].as_u64().unwrap();

    // Broadcast another event
    broadcast_heartbeat(&state);
    let msg = rx.try_recv().unwrap();
    let Message::Text(text) = msg else { panic!() };
    let event: Value = serde_json::from_str(&text).unwrap();
    let seq2 = event["seq"].as_u64().unwrap();

    // Seq should be monotonically increasing
    assert!(seq2 > seq1, "Event seq should be monotonically increasing");
}

#[test]
fn test_health_broadcast_on_status_change() {
    let state = WsServerState::new(WsServerConfig::default());

    let (tx, mut rx) = mpsc::channel(256);
    let conn = make_conn_with_id("operator", vec![], "conn-1");
    state.register_connection(&conn, tx, None);

    // Clear presence broadcast
    let _ = rx.try_recv();

    // Update health - should broadcast since status changes
    state.update_health("degraded", None, None);

    let msg = rx.try_recv();
    assert!(
        msg.is_ok(),
        "Should receive health broadcast on status change"
    );
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "health");
    assert_eq!(event["payload"]["status"], "degraded");
    assert!(event["stateVersion"]["health"].as_u64().unwrap() >= 1);

    // Update health with same status - should NOT broadcast
    state.update_health("degraded", Some(json!({"more": "data"})), None);
    let msg = rx.try_recv();
    assert!(msg.is_err(), "Should not broadcast when status unchanged");
}

// ============================================================================
// Cron Scheduler Integration Tests
// ============================================================================

#[test]
fn test_cron_scheduler_integration() {
    use crate::cron::{
        CronJobCreate, CronJobPatch, CronPayload, CronRunMode, CronSchedule, CronSessionTarget,
        CronWakeMode,
    };

    let state = WsServerState::new(WsServerConfig::default());

    // Test scheduler status
    let status = state.cron_scheduler.status();
    assert!(status.enabled);
    assert_eq!(status.jobs, 0);

    // Add a job
    let job = state
        .cron_scheduler
        .add(CronJobCreate {
            name: "Test Job".to_string(),
            agent_id: None,
            description: None,
            enabled: true,
            delete_after_run: None,
            schedule: CronSchedule::Every {
                every_ms: 60000,
                anchor_ms: None,
            },
            session_target: CronSessionTarget::Main,
            wake_mode: CronWakeMode::Now,
            payload: CronPayload::SystemEvent {
                text: "Hello!".to_string(),
            },
            isolation: None,
        })
        .unwrap();
    assert_eq!(job.name, "Test Job");
    assert!(job.state.next_run_at_ms.is_some());

    // List jobs
    let jobs = state.cron_scheduler.list(true);
    assert_eq!(jobs.len(), 1);

    // Check status after adding
    let status = state.cron_scheduler.status();
    assert_eq!(status.jobs, 1);

    // Update the job
    let updated = state
        .cron_scheduler
        .update(
            &job.id,
            CronJobPatch {
                name: Some("Updated Job".to_string()),
                enabled: Some(false),
                ..Default::default()
            },
        )
        .unwrap();
    assert_eq!(updated.name, "Updated Job");
    assert!(!updated.enabled);

    // Force run the job
    let result = state
        .cron_scheduler
        .run(&job.id, Some(CronRunMode::Force))
        .unwrap();
    assert!(result.ok);
    assert!(result.ran);
    assert!(result.payload.is_some());

    // Run no longer records log entry — mark_run_finished does.
    state
        .cron_scheduler
        .mark_run_finished(&job.id, crate::cron::CronJobStatus::Ok, 0, None)
        .unwrap();

    // Check runs
    let runs = state.cron_scheduler.runs(Some(&job.id), None);
    assert_eq!(runs.len(), 1);

    // Remove the job
    let remove_result = state.cron_scheduler.remove(&job.id);
    assert!(remove_result.ok);
    assert!(remove_result.removed);

    // Verify removed
    let jobs = state.cron_scheduler.list(true);
    assert_eq!(jobs.len(), 0);
}

#[test]
fn test_cron_scheduler_update_job_not_found() {
    use crate::cron::CronJobPatch;

    let state = WsServerState::new(WsServerConfig::default());

    let result = state.cron_scheduler.update(
        "non-existent-job",
        CronJobPatch {
            name: Some("Updated".to_string()),
            ..Default::default()
        },
    );
    assert!(result.is_err());
}

#[test]
fn test_cron_scheduler_run_not_found() {
    use crate::cron::CronRunMode;

    let state = WsServerState::new(WsServerConfig::default());

    let result = state
        .cron_scheduler
        .run("non-existent-job", Some(CronRunMode::Force));
    assert!(result.is_err());
}

#[test]
fn test_cron_scheduler_event_integration() {
    use crate::cron::{
        CronJobCreate, CronPayload, CronRunMode, CronSchedule, CronSessionTarget, CronWakeMode,
    };

    let state = WsServerState::new(WsServerConfig::default());

    let (tx, mut rx) = mpsc::channel(256);
    let conn = make_conn_with_id("operator", vec![], "conn-1");
    state.register_connection(&conn, tx, None);

    // Clear the presence broadcast
    let _ = rx.try_recv();

    // Add a job and manually broadcast event (simulating what handlers do)
    let job = state
        .cron_scheduler
        .add(CronJobCreate {
            name: "Test Job".to_string(),
            agent_id: None,
            description: None,
            enabled: true,
            delete_after_run: None,
            schedule: CronSchedule::Every {
                every_ms: 60000,
                anchor_ms: None,
            },
            session_target: CronSessionTarget::Main,
            wake_mode: CronWakeMode::Now,
            payload: CronPayload::SystemEvent {
                text: "Hello!".to_string(),
            },
            isolation: None,
        })
        .unwrap();

    // Broadcast event manually (normally done by handlers)
    broadcast_cron_event(
        &state,
        &job.id,
        "scheduled",
        None,
        Some(json!({"name": job.name})),
    );

    // Should receive cron event
    let msg = rx.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["event"], "cron");
    assert_eq!(event["payload"]["jobId"], job.id);
    assert_eq!(event["payload"]["status"], "scheduled");

    // Run the job and broadcast events
    let result = state
        .cron_scheduler
        .run(&job.id, Some(CronRunMode::Force))
        .unwrap();
    assert!(result.ran);

    broadcast_cron_event(&state, &job.id, "started", None, None);
    broadcast_cron_event(
        &state,
        &job.id,
        "completed",
        None,
        Some(json!({"status": "ok"})),
    );

    // Should receive started event
    let msg = rx.try_recv().unwrap();
    let Message::Text(text) = msg else { panic!() };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["payload"]["status"], "started");

    // Should receive completed event
    let msg = rx.try_recv().unwrap();
    let Message::Text(text) = msg else { panic!() };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["payload"]["status"], "completed");
}

// ============== Node Event Routing Tests ==============

fn make_node_conn(node_id: &str, conn_id: &str) -> ConnectionContext {
    ConnectionContext {
        conn_id: conn_id.to_string(),
        role: "node".to_string(),
        scopes: vec![],
        client: ClientInfo {
            id: node_id.to_string(),
            version: "1.0".to_string(),
            platform: "test".to_string(),
            mode: "node".to_string(),
            display_name: None,
            device_family: None,
            model_identifier: None,
            instance_id: None,
        },
        device_id: Some(node_id.to_string()),
    }
}

fn pair_node(state: &WsServerState, node_id: &str) {
    let params = json!({ "nodeId": node_id, "displayName": node_id });
    let result = handle_node_pair_request(Some(&params), state).unwrap();
    let request_id = result["request"]["requestId"].as_str().unwrap();
    handle_node_pair_approve(Some(&json!({ "requestId": request_id })), state).unwrap();
}

#[test]
fn test_handle_node_event_requires_node_role() {
    let state = WsServerState::new(WsServerConfig::default());
    let conn = make_conn("admin");

    let params = json!({
        "nodeId": "node-1",
        "event": "some.event",
        "payload": { "key": "value" }
    });
    let result = handle_node_event(Some(&params), &state, &conn);
    assert!(result.is_err());
    assert_eq!(result.unwrap_err().code, ERROR_INVALID_REQUEST);
}

#[test]
fn test_handle_node_event_requires_pairing() {
    let state = WsServerState::new(WsServerConfig::default());
    let conn = make_node_conn("unpaired-node", "conn-node");

    let params = json!({
        "nodeId": "unpaired-node",
        "event": "some.event"
    });
    let result = handle_node_event(Some(&params), &state, &conn);
    assert!(result.is_err());
    assert_eq!(result.unwrap_err().code, ERROR_NOT_PAIRED);
}

#[test]
fn test_handle_node_event_requires_event_field() {
    let state = WsServerState::new(WsServerConfig::default());
    pair_node(&state, "node-1");
    let conn = make_node_conn("node-1", "conn-node-1");

    let params = json!({
        "nodeId": "node-1",
        "payload": { "key": "value" }
    });
    let result = handle_node_event(Some(&params), &state, &conn);
    assert!(result.is_err());
    assert_eq!(result.unwrap_err().code, ERROR_INVALID_REQUEST);
}

#[test]
fn test_handle_node_event_prevents_node_id_mismatch() {
    let state = WsServerState::new(WsServerConfig::default());
    pair_node(&state, "node-1");
    let conn = make_node_conn("node-1", "conn-node-1");

    let params = json!({
        "nodeId": "node-other",
        "event": "some.event"
    });
    let result = handle_node_event(Some(&params), &state, &conn);
    assert!(result.is_err());
    assert_eq!(result.unwrap_err().code, ERROR_INVALID_REQUEST);
}

#[test]
fn test_handle_node_event_broadcasts_to_operators() {
    let state = WsServerState::new(WsServerConfig::default());
    pair_node(&state, "node-1");

    // Register an operator connection to receive broadcasts
    let (tx_op, mut rx_op) = mpsc::channel(256);
    let op_conn = make_conn_with_id("operator", vec![], "op-conn");
    state.register_connection(&op_conn, tx_op, None);

    // Clear the presence broadcast
    let _ = rx_op.try_recv();

    // Send a node event
    let node_conn = make_node_conn("node-1", "conn-node-1");
    let params = json!({
        "nodeId": "node-1",
        "event": "status.changed",
        "payload": { "status": "idle", "battery": 85 }
    });
    let result = handle_node_event(Some(&params), &state, &node_conn);
    assert!(result.is_ok());

    let response = result.unwrap();
    assert_eq!(response["ok"], true);
    assert_eq!(response["nodeId"], "node-1");
    assert_eq!(response["event"], "status.changed");
    assert_eq!(response["hasPayload"], true);

    // Operator should receive the broadcast
    let msg = rx_op.try_recv();
    assert!(msg.is_ok(), "Operator should receive node.event broadcast");
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["type"], "event");
    assert_eq!(event["event"], "node.event");
    assert_eq!(event["payload"]["nodeId"], "node-1");
    assert_eq!(event["payload"]["event"], "status.changed");
    assert_eq!(event["payload"]["payload"]["status"], "idle");
    assert_eq!(event["payload"]["payload"]["battery"], 85);
    assert!(event["payload"]["ts"].as_u64().is_some());
}

#[test]
fn test_handle_node_event_does_not_broadcast_to_nodes() {
    let state = WsServerState::new(WsServerConfig::default());
    pair_node(&state, "node-1");
    pair_node(&state, "node-2");

    // Register a node connection (should NOT receive broadcasts)
    let (tx_node2, mut rx_node2) = mpsc::channel(256);
    let node2_conn = make_node_conn("node-2", "conn-node-2");
    state.register_connection(&node2_conn, tx_node2, None);

    // Register an operator connection (should receive broadcasts)
    let (tx_op, mut rx_op) = mpsc::channel(256);
    let op_conn = make_conn_with_id("operator", vec![], "op-conn");
    state.register_connection(&op_conn, tx_op, None);

    // Clear presence broadcasts
    let _ = rx_op.try_recv();
    let _ = rx_op.try_recv();

    // Send a node event from node-1
    let node1_conn = make_node_conn("node-1", "conn-node-1");
    let params = json!({
        "nodeId": "node-1",
        "event": "heartbeat",
        "payload": { "uptime": 12345 }
    });
    let result = handle_node_event(Some(&params), &state, &node1_conn);
    assert!(result.is_ok());

    // Operator should receive it
    assert!(rx_op.try_recv().is_ok(), "Operator should receive event");

    // Node-2 should NOT receive it
    assert!(
        rx_node2.try_recv().is_err(),
        "Other nodes should not receive node.event broadcasts"
    );
}

#[test]
fn test_handle_node_event_with_payload_json() {
    let state = WsServerState::new(WsServerConfig::default());
    pair_node(&state, "node-1");

    // Register an operator connection
    let (tx_op, mut rx_op) = mpsc::channel(256);
    let op_conn = make_conn_with_id("operator", vec![], "op-conn");
    state.register_connection(&op_conn, tx_op, None);

    // Clear the presence broadcast
    let _ = rx_op.try_recv();

    // Send a node event with payloadJSON (string-encoded JSON)
    let node_conn = make_node_conn("node-1", "conn-node-1");
    let params = json!({
        "nodeId": "node-1",
        "event": "data.update",
        "payloadJSON": r#"{"temperature":22.5,"humidity":60}"#
    });
    let result = handle_node_event(Some(&params), &state, &node_conn);
    assert!(result.is_ok());
    assert_eq!(result.unwrap()["hasPayload"], true);

    // Operator should receive the broadcast with parsed payloadJSON
    let msg = rx_op.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["payload"]["event"], "data.update");
    assert_eq!(event["payload"]["payload"]["temperature"], 22.5);
    assert_eq!(event["payload"]["payload"]["humidity"], 60);
}

#[test]
fn test_handle_node_event_without_payload() {
    let state = WsServerState::new(WsServerConfig::default());
    pair_node(&state, "node-1");

    // Register an operator connection
    let (tx_op, mut rx_op) = mpsc::channel(256);
    let op_conn = make_conn_with_id("operator", vec![], "op-conn");
    state.register_connection(&op_conn, tx_op, None);

    // Clear the presence broadcast
    let _ = rx_op.try_recv();

    // Send a node event without any payload
    let node_conn = make_node_conn("node-1", "conn-node-1");
    let params = json!({
        "nodeId": "node-1",
        "event": "ready"
    });
    let result = handle_node_event(Some(&params), &state, &node_conn);
    assert!(result.is_ok());
    assert_eq!(result.unwrap()["hasPayload"], false);

    // Operator should receive the broadcast with null payload
    let msg = rx_op.try_recv();
    assert!(msg.is_ok());
    let Message::Text(text) = msg.unwrap() else {
        panic!("expected text message");
    };
    let event: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(event["payload"]["event"], "ready");
    assert_eq!(event["payload"]["payload"], Value::Null);
}

#[test]
fn test_handle_node_event_defaults_node_id_from_connection() {
    let state = WsServerState::new(WsServerConfig::default());
    pair_node(&state, "node-1");

    let node_conn = make_node_conn("node-1", "conn-node-1");

    // Send without explicit nodeId -- should default to connection's node id
    let params = json!({
        "event": "ping"
    });
    let result = handle_node_event(Some(&params), &state, &node_conn);
    assert!(result.is_ok());
    assert_eq!(result.unwrap()["nodeId"], "node-1");
}

#[test]
fn test_node_event_in_gateway_events() {
    // Verify that node.event is registered as a known gateway event
    assert!(
        GATEWAY_EVENTS.contains(&"node.event"),
        "node.event should be in GATEWAY_EVENTS"
    );
}

#[test]
fn test_node_list_merges_paired_metadata_with_live_data() {
    let state = WsServerState::new(WsServerConfig::default());

    // Pair a node with extended fields
    let params = json!({
        "nodeId": "node-merge",
        "displayName": "Merge Node",
        "platform": "darwin",
        "version": "1.0.0",
        "coreVersion": "0.9.0",
        "uiVersion": "1.0.0",
        "deviceFamily": "Mac",
        "modelIdentifier": "MacBookPro18,3",
        "caps": ["exec", "audio"],
        "commands": ["system.run"],
        "permissions": { "filesystem": true },
        "remoteIp": "192.168.1.10"
    });
    let result = handle_node_pair_request(Some(&params), &state).unwrap();
    let request_id = result["request"]["requestId"].as_str().unwrap();
    handle_node_pair_approve(Some(&json!({ "requestId": request_id })), &state).unwrap();

    // Register a live session with updated version
    {
        let mut registry = state.node_registry.lock();
        registry.register(NodeSession {
            node_id: "node-merge".to_string(),
            conn_id: "conn-merge".to_string(),
            display_name: Some("Merge Node Live".to_string()),
            platform: Some("darwin".to_string()),
            version: Some("2.0.0".to_string()),
            device_family: Some("Mac".to_string()),
            model_identifier: Some("MacBookPro18,3".to_string()),
            remote_ip: Some("10.0.0.5".to_string()),
            caps: vec!["exec".to_string(), "video".to_string()],
            commands: HashSet::from(["system.run".to_string(), "screen.capture".to_string()]),
            permissions: Some({
                let mut m = HashMap::new();
                m.insert("filesystem".to_string(), true);
                m.insert("screen".to_string(), true);
                m
            }),
            path_env: Some("/usr/local/bin".to_string()),
            connected_at_ms: now_ms(),
        });
    }

    // node.list should merge paired + live data
    let result = handle_node_list(&state).unwrap();
    let nodes = result["nodes"].as_array().unwrap();
    assert_eq!(nodes.len(), 1);

    let node = &nodes[0];
    // Live data should take precedence for version and display name
    assert_eq!(node["version"], "2.0.0");
    assert_eq!(node["displayName"], "Merge Node Live");
    assert_eq!(node["remoteIp"], "10.0.0.5");
    // Paired-only fields should still be present
    assert_eq!(node["coreVersion"], "0.9.0");
    assert_eq!(node["uiVersion"], "1.0.0");
    // Caps should be merged (union of both)
    let caps = node["caps"].as_array().unwrap();
    let cap_strs: Vec<&str> = caps.iter().map(|v| v.as_str().unwrap()).collect();
    assert!(cap_strs.contains(&"exec"));
    assert!(cap_strs.contains(&"audio"));
    assert!(cap_strs.contains(&"video"));
    // Commands should be merged
    let commands = node["commands"].as_array().unwrap();
    let cmd_strs: Vec<&str> = commands.iter().map(|v| v.as_str().unwrap()).collect();
    assert!(cmd_strs.contains(&"system.run"));
    assert!(cmd_strs.contains(&"screen.capture"));
    // Should be connected and paired
    assert_eq!(node["connected"], true);
    assert_eq!(node["paired"], true);
    // pathEnv from live data
    assert_eq!(node["pathEnv"], "/usr/local/bin");
}

#[test]
fn test_node_describe_merges_paired_metadata_with_live_data() {
    let state = WsServerState::new(WsServerConfig::default());

    // Pair a node with extended fields
    let params = json!({
        "nodeId": "node-describe-merge",
        "displayName": "Describe Merge",
        "platform": "ios",
        "version": "1.0.0",
        "coreVersion": "0.8.0",
        "uiVersion": "1.0.0",
        "deviceFamily": "iPhone",
        "modelIdentifier": "iPhone14,2",
        "caps": ["camera"],
        "commands": ["camera.snap"],
        "permissions": { "camera": true },
        "remoteIp": "172.16.0.1"
    });
    let result = handle_node_pair_request(Some(&params), &state).unwrap();
    let request_id = result["request"]["requestId"].as_str().unwrap();
    handle_node_pair_approve(Some(&json!({ "requestId": request_id })), &state).unwrap();

    // Register a live session with additional caps
    {
        let mut registry = state.node_registry.lock();
        registry.register(NodeSession {
            node_id: "node-describe-merge".to_string(),
            conn_id: "conn-describe-merge".to_string(),
            display_name: Some("Describe Merge Live".to_string()),
            platform: Some("ios".to_string()),
            version: Some("2.0.0".to_string()),
            device_family: Some("iPhone".to_string()),
            model_identifier: Some("iPhone14,2".to_string()),
            remote_ip: Some("172.16.0.2".to_string()),
            caps: vec!["camera".to_string(), "location".to_string()],
            commands: HashSet::from(["camera.snap".to_string(), "location.get".to_string()]),
            permissions: None,
            path_env: None,
            connected_at_ms: now_ms(),
        });
    }

    // node.describe should merge paired + live data
    let describe_params = json!({ "nodeId": "node-describe-merge" });
    let result = handle_node_describe(Some(&describe_params), &state).unwrap();

    // Live data should take precedence
    assert_eq!(result["version"], "2.0.0");
    assert_eq!(result["displayName"], "Describe Merge Live");
    assert_eq!(result["remoteIp"], "172.16.0.2");
    // Paired-only fields should still be present
    assert_eq!(result["coreVersion"], "0.8.0");
    assert_eq!(result["uiVersion"], "1.0.0");
    // Caps merged
    let caps = result["caps"].as_array().unwrap();
    let cap_strs: Vec<&str> = caps.iter().map(|v| v.as_str().unwrap()).collect();
    assert!(cap_strs.contains(&"camera"));
    assert!(cap_strs.contains(&"location"));
    // Commands merged
    let commands = result["commands"].as_array().unwrap();
    let cmd_strs: Vec<&str> = commands.iter().map(|v| v.as_str().unwrap()).collect();
    assert!(cmd_strs.contains(&"camera.snap"));
    assert!(cmd_strs.contains(&"location.get"));
    // Permissions from paired data should be used since live has None
    assert_eq!(result["permissions"], json!({ "camera": true }));
    // Connected and paired
    assert_eq!(result["connected"], true);
    assert_eq!(result["paired"], true);
}

#[tokio::test]
async fn test_shutdown_watch_channel_propagates() {
    let (tx, mut rx) = tokio::sync::watch::channel(false);

    // Initially not shut down
    assert!(!*rx.borrow());

    // Simulate shutdown signal
    tx.send(true).unwrap();

    // Receiver should observe the change
    rx.changed().await.unwrap();
    assert!(*rx.borrow());
}

#[tokio::test]
async fn test_shutdown_watch_channel_multiple_receivers() {
    let (tx, rx1) = tokio::sync::watch::channel(false);
    let mut rx2 = rx1.clone();
    let mut rx3 = rx1.clone();

    // All receivers start as false
    assert!(!*rx1.borrow());
    assert!(!*rx2.borrow());
    assert!(!*rx3.borrow());

    // Trigger shutdown
    tx.send(true).unwrap();

    // All receivers see the update
    rx2.changed().await.unwrap();
    rx3.changed().await.unwrap();
    assert!(*rx2.borrow());
    assert!(*rx3.borrow());
}

#[test]
fn test_broadcast_shutdown_notifies_connected_clients() {
    let state = Arc::new(WsServerState::new(WsServerConfig::default()));

    // Register a mock connection
    let (tx, mut rx) = tokio::sync::mpsc::channel(256);
    let conn = ConnectionContext {
        conn_id: "conn-shutdown-test".to_string(),
        role: "operator".to_string(),
        scopes: vec![],
        client: ClientInfo {
            id: "test".to_string(),
            version: "1.0".to_string(),
            platform: "test".to_string(),
            mode: "test".to_string(),
            display_name: None,
            device_family: None,
            model_identifier: None,
            instance_id: None,
        },
        device_id: None,
    };
    state.register_connection(&conn, tx, None);

    // Drain any messages sent during registration (e.g. presence broadcast)
    while rx.try_recv().is_ok() {}

    // Broadcast shutdown
    broadcast_shutdown(&state, "SIGTERM", None);

    // The connected client should receive a shutdown event frame
    let msg = rx
        .try_recv()
        .expect("should have received shutdown message");
    let text = match msg {
        Message::Text(t) => t,
        other => panic!("expected text message, got {:?}", other),
    };
    let parsed: serde_json::Value = serde_json::from_str(&text).unwrap();
    assert_eq!(parsed["event"], "shutdown");
    assert_eq!(parsed["payload"]["reason"], "SIGTERM");
}

#[test]
fn test_broadcast_sends_identical_bytes_to_all_connections() {
    let state = WsServerState::new(WsServerConfig::default());

    // Register three operator connections
    let (tx1, mut rx1) = mpsc::channel(256);
    let (tx2, mut rx2) = mpsc::channel(256);
    let (tx3, mut rx3) = mpsc::channel(256);

    let conn1 = make_conn_with_id("operator", vec![], "conn-1");
    let conn2 = make_conn_with_id("operator", vec![], "conn-2");
    let conn3 = make_conn_with_id("operator", vec![], "conn-3");

    state.register_connection(&conn1, tx1, None);
    state.register_connection(&conn2, tx2, None);
    state.register_connection(&conn3, tx3, None);

    // Drain all presence broadcasts from registration
    while rx1.try_recv().is_ok() {}
    while rx2.try_recv().is_ok() {}
    while rx3.try_recv().is_ok() {}

    // Broadcast a general event
    broadcast_event(
        &state,
        "agent",
        json!({"runId": "run-1", "seq": 1, "stream": "text", "data": {"text": "Hello!"}}),
    );

    let msg1 = rx1.try_recv().expect("conn-1 should receive broadcast");
    let msg2 = rx2.try_recv().expect("conn-2 should receive broadcast");
    let msg3 = rx3.try_recv().expect("conn-3 should receive broadcast");

    // Extract text from each message
    let text1 = match msg1 {
        Message::Text(t) => t,
        other => panic!("expected text message, got {:?}", other),
    };
    let text2 = match msg2 {
        Message::Text(t) => t,
        other => panic!("expected text message, got {:?}", other),
    };
    let text3 = match msg3 {
        Message::Text(t) => t,
        other => panic!("expected text message, got {:?}", other),
    };

    // All three connections must receive byte-identical JSON strings.
    // This verifies that broadcast serializes once and clones the string,
    // rather than re-serializing per connection (which could produce
    // different key orderings or whitespace).
    assert_eq!(
        text1, text2,
        "conn-1 and conn-2 should receive identical bytes"
    );
    assert_eq!(
        text2, text3,
        "conn-2 and conn-3 should receive identical bytes"
    );

    // Sanity: verify the content is valid JSON with expected fields
    let parsed: Value = serde_json::from_str(&text1).unwrap();
    assert_eq!(parsed["type"], "event");
    assert_eq!(parsed["event"], "agent");
    assert_eq!(parsed["payload"]["runId"], "run-1");

    // Also verify shutdown broadcast sends identical bytes (different code path)
    while rx1.try_recv().is_ok() {}
    while rx2.try_recv().is_ok() {}
    while rx3.try_recv().is_ok() {}

    broadcast_shutdown(&state, "update", Some(3000));

    let s1 = match rx1.try_recv().unwrap() {
        Message::Text(t) => t,
        other => panic!("expected text, got {:?}", other),
    };
    let s2 = match rx2.try_recv().unwrap() {
        Message::Text(t) => t,
        other => panic!("expected text, got {:?}", other),
    };
    let s3 = match rx3.try_recv().unwrap() {
        Message::Text(t) => t,
        other => panic!("expected text, got {:?}", other),
    };

    assert_eq!(
        s1, s2,
        "shutdown bytes must be identical across connections"
    );
    assert_eq!(
        s2, s3,
        "shutdown bytes must be identical across connections"
    );

    // Also verify voicewake broadcast (sends to all including nodes)
    let (tx_node, mut rx_node) = mpsc::channel(256);
    let node_conn = make_conn_with_id("node", vec![], "node-conn");
    state.register_connection(&node_conn, tx_node, None);

    while rx1.try_recv().is_ok() {}
    while rx_node.try_recv().is_ok() {}

    broadcast_voicewake_changed(&state, vec!["hey claude".to_string()]);

    let v1 = match rx1.try_recv().unwrap() {
        Message::Text(t) => t,
        other => panic!("expected text, got {:?}", other),
    };
    let v_node = match rx_node.try_recv().unwrap() {
        Message::Text(t) => t,
        other => panic!("expected text, got {:?}", other),
    };

    assert_eq!(
        v1, v_node,
        "voicewake broadcast bytes must be identical for operator and node"
    );
}

#[test]
fn test_every_gateway_method_has_explicit_role_assignment() {
    // Methods that are explicitly assigned "admin" in get_method_required_role.
    // If a method returns "admin" but is NOT in this list, it means the method
    // is falling through to the default `_ => "admin"` arm, which is a bug —
    // it should be explicitly listed.
    let explicit_admin_methods: std::collections::HashSet<&str> = [
        "device.pair.approve",
        "device.pair.reject",
        "device.token.rotate",
        "device.token.revoke",
        "node.pair.request",
        "node.pair.approve",
        "node.pair.reject",
        "node.pair.verify",
        "node.rename",
        "exec.approvals.set",
        "exec.approvals.node.set",
        "exec.approval.request",
        "exec.approval.resolve",
        "sessions.export_user",
        "sessions.purge_user",
        "system-event",
        "config.reload",
    ]
    .into_iter()
    .collect();

    // Node-only methods are dispatched before role checks, so they
    // legitimately have no explicit role assignment.
    let node_only: std::collections::HashSet<&str> = NODE_ONLY_METHODS.iter().copied().collect();

    for method in &GATEWAY_METHODS {
        if node_only.contains(method) {
            continue;
        }
        let role = get_method_required_role(method);
        assert!(
            role == "read" || role == "write" || explicit_admin_methods.contains(method),
            "GATEWAY_METHODS entry '{}' returns role '{}' via default fallthrough — \
             add it to an explicit arm in get_method_required_role()",
            method,
            role,
        );
    }
}

// ---------------------------------------------------------------------------
// JSON depth validation tests
// ---------------------------------------------------------------------------

/// Helper: build a JSON value nested to exactly `depth` levels using objects.
fn nested_object(depth: usize) -> Value {
    let mut value = json!("leaf");
    for _ in 0..depth.saturating_sub(1) {
        value = json!({ "nested": value });
    }
    value
}

/// Helper: build a JSON value nested to exactly `depth` levels using arrays.
fn nested_array(depth: usize) -> Value {
    let mut value = json!("leaf");
    for _ in 0..depth.saturating_sub(1) {
        value = json!([value]);
    }
    value
}

/// Helper: build mixed nesting (alternating objects and arrays) to `depth`.
fn mixed_nesting(depth: usize) -> Value {
    let mut value = json!("leaf");
    for i in 0..depth.saturating_sub(1) {
        if i % 2 == 0 {
            value = json!({ "nested": value });
        } else {
            value = json!([value]);
        }
    }
    value
}

#[test]
fn test_shallow_json_passes() {
    // Depth 1: simple scalar values
    assert!(validate_json_depth(&json!("hello"), MAX_JSON_DEPTH).is_ok());
    assert!(validate_json_depth(&json!(42), MAX_JSON_DEPTH).is_ok());

    // Depth 2: one level of nesting
    assert!(validate_json_depth(&json!({"a": 1}), MAX_JSON_DEPTH).is_ok());

    // Depth 3-5: a few levels
    assert!(validate_json_depth(&nested_object(3), MAX_JSON_DEPTH).is_ok());
    assert!(validate_json_depth(&nested_object(5), MAX_JSON_DEPTH).is_ok());
}

#[test]
fn test_exact_limit_passes() {
    let value = nested_object(MAX_JSON_DEPTH);
    assert!(validate_json_depth(&value, MAX_JSON_DEPTH).is_ok());
}

#[test]
fn test_exceeds_limit_fails() {
    let value = nested_object(MAX_JSON_DEPTH + 1);
    let result = validate_json_depth(&value, MAX_JSON_DEPTH);
    assert!(result.is_err());
    assert!(result
        .unwrap_err()
        .contains("JSON nesting depth exceeds maximum"));
}

#[test]
fn test_deeply_nested_array_fails() {
    let value = nested_array(MAX_JSON_DEPTH + 1);
    let result = validate_json_depth(&value, MAX_JSON_DEPTH);
    assert!(result.is_err());
    assert!(result
        .unwrap_err()
        .contains("JSON nesting depth exceeds maximum"));
}

#[test]
fn test_mixed_nesting_fails() {
    let value = mixed_nesting(MAX_JSON_DEPTH + 1);
    let result = validate_json_depth(&value, MAX_JSON_DEPTH);
    assert!(result.is_err());
    assert!(result
        .unwrap_err()
        .contains("JSON nesting depth exceeds maximum"));
}

#[test]
fn test_flat_large_object_passes() {
    // 1000 keys at depth 1 -- size != depth
    let mut map = serde_json::Map::new();
    for i in 0..1000 {
        map.insert(format!("key_{i}"), json!(i));
    }
    let value = Value::Object(map);
    assert!(validate_json_depth(&value, MAX_JSON_DEPTH).is_ok());
}

#[test]
fn test_empty_value_passes() {
    assert!(validate_json_depth(&json!(null), MAX_JSON_DEPTH).is_ok());
    assert!(validate_json_depth(&json!(true), MAX_JSON_DEPTH).is_ok());
    assert!(validate_json_depth(&json!(false), MAX_JSON_DEPTH).is_ok());
    assert!(validate_json_depth(&json!(0), MAX_JSON_DEPTH).is_ok());
    assert!(validate_json_depth(&json!(3.15), MAX_JSON_DEPTH).is_ok());
    assert!(validate_json_depth(&json!(""), MAX_JSON_DEPTH).is_ok());
}

#[test]
fn test_validate_json_depth_at_small_limit() {
    // Test with a smaller limit.
    // Depth model: root value starts at depth 1.
    // {"a": {"b": 1}} has leaf at depth 3, so it needs max_depth >= 3.
    assert!(validate_json_depth(&json!({"a": {"b": 1}}), 3).is_ok());
    // {"a": {"b": {"c": 1}}} has leaf at depth 4, so max_depth 3 rejects it.
    assert!(validate_json_depth(&json!({"a": {"b": {"c": 1}}}), 3).is_err());
}

#[test]
fn test_validate_json_depth_zero_rejects_nested() {
    // Depth 0 rejects everything because even a scalar starts at depth 1.
    assert!(validate_json_depth(&json!("hello"), 0).is_err());
    assert!(validate_json_depth(&json!(42), 0).is_err());
    // Nested structures are also rejected.
    assert!(validate_json_depth(&json!({"a": 1}), 0).is_err());
    assert!(validate_json_depth(&json!([1]), 0).is_err());
}

#[test]
fn test_validate_json_depth_limit_1() {
    // Depth 1 allows only scalar values (root at depth 1, no children).
    assert!(validate_json_depth(&json!("hello"), 1).is_ok());
    assert!(validate_json_depth(&json!(42), 1).is_ok());
    // Objects and arrays have children at depth 2, which exceeds limit 1.
    assert!(validate_json_depth(&json!({"a": 1}), 1).is_err());
    assert!(validate_json_depth(&json!([1, 2, 3]), 1).is_err());
    // Nested structures are also rejected.
    assert!(validate_json_depth(&json!({"a": {"b": 1}}), 1).is_err());
    assert!(validate_json_depth(&json!([[1]]), 1).is_err());
}

/// `NewVerificationFlow::from_upsert` returns `Some` only when the
/// upsert actually inserted the record. Without this gate,
/// `start_matrix_verification` would broadcast
/// `matrix.verification.requested` for an already-known flow that the
/// inbound handler had already announced, duplicating UI notifications.
#[test]
fn test_new_verification_flow_from_upsert_gates_on_inserted() {
    use crate::channels::matrix::{MatrixVerificationInfo, MatrixVerificationState};
    let info = MatrixVerificationInfo {
        flow_id: "flow".to_string(),
        protocol_flow_id: "txn".to_string(),
        raw_protocol_flow_id: "txn".to_string(),
        user_id: "@a:x".parse().expect("user id"),
        raw_user_id: "@a:x".parse().expect("user id"),
        device_id: None,
        state: MatrixVerificationState::Requested,
        sas: None,
        created_at: 0,
        updated_at: 0,
    };
    assert!(
        NewVerificationFlow::from_upsert(&info, true).is_some(),
        "inserted=true must produce a witness"
    );
    assert!(
        NewVerificationFlow::from_upsert(&info, false).is_none(),
        "inserted=false must produce no witness; refresh-tick rebuilds must NOT re-emit `requested`"
    );
}

/// Pin that `broadcast_matrix_verification_request` accepts an Option
/// witness — the typed call site `Some(NewVerificationFlow::from_upsert(...))`
/// vs the `None` no-op. The test exists primarily to compile-fail any
/// future regression that drops the Option wrapper.
#[test]
fn test_broadcast_matrix_verification_request_takes_optional_witness() {
    fn _accepts_none(state: &WsServerState) {
        broadcast_matrix_verification_request(state, None);
    }
    // We don't actually invoke the function (no real server state in
    // this test scope); the function signature is what's pinned.
    let _ = _accepts_none;
}

/// End-to-end delivery + scope-filtering for
/// `matrix.verification.requested`. Constructs a real
/// `MatrixVerificationInfo`, calls `broadcast_matrix_verification_request`
/// against scope-gated connections, and asserts only admin-scoped
/// receivers see the event. Complements the synthetic-event pin
/// (`test_broadcast_event_matrix_prefix_defaults_to_admin`) by
/// covering the actual production payload + helper.
#[test]
fn test_broadcast_matrix_verification_request_delivers_only_to_admin_scope() {
    use crate::channels::matrix::{MatrixVerificationInfo, MatrixVerificationState};

    let state = WsServerState::new(WsServerConfig::default());
    let (tx_no_admin, mut rx_no_admin) = mpsc::channel(256);
    let (tx_admin_role, mut rx_admin_role) = mpsc::channel(256);
    let (tx_admin_scope, mut rx_admin_scope) = mpsc::channel(256);
    let (tx_node, mut rx_node) = mpsc::channel(256);
    let no_admin = make_conn_with_id("operator", vec![], "verify-no-admin");
    let admin_role = make_conn_with_id("admin", vec![], "verify-admin-role");
    let admin_scope = make_conn_with_id(
        "operator",
        vec!["operator.admin".to_string()],
        "verify-admin-scope",
    );
    let node = make_conn_with_id("node", vec![], "verify-node");

    state.register_connection(&no_admin, tx_no_admin, None);
    state.register_connection(&admin_role, tx_admin_role, None);
    state.register_connection(&admin_scope, tx_admin_scope, None);
    state.register_connection(&node, tx_node, None);

    while rx_no_admin.try_recv().is_ok() {}
    while rx_admin_role.try_recv().is_ok() {}
    while rx_admin_scope.try_recv().is_ok() {}
    while rx_node.try_recv().is_ok() {}

    let info = MatrixVerificationInfo {
        flow_id: "flow-abc".to_string(),
        protocol_flow_id: "txn-abc".to_string(),
        raw_protocol_flow_id: "txn-abc".to_string(),
        user_id: "@alice:example.com".parse().expect("user id"),
        raw_user_id: "@alice:example.com".parse().expect("user id"),
        device_id: Some("DEVICE".into()),
        state: MatrixVerificationState::Requested,
        sas: None,
        created_at: 1,
        updated_at: 1,
    };
    broadcast_matrix_verification_request(
        &state,
        crate::server::ws::NewVerificationFlow::from_upsert(&info, true),
    );

    assert!(
        rx_no_admin.try_recv().is_err(),
        "operator without operator.admin scope must NOT receive matrix.verification.requested"
    );
    assert!(
        rx_admin_role.try_recv().is_ok(),
        "admin-role connection must receive matrix.verification.requested"
    );
    assert!(
        rx_admin_scope.try_recv().is_ok(),
        "operator with operator.admin scope must receive matrix.verification.requested"
    );
    assert!(
        rx_node.try_recv().is_err(),
        "node-role connection must NOT receive matrix.verification.requested \
         (broadcast_event skips role==node)"
    );
}
