//! HTTP/SSE transport integration tests for klieo-a2a.
//!
//! Driven via `tower::ServiceExt::oneshot` against the axum
//! Router. SSE bodies parsed by the shared `common::sse_parser`.

#![cfg(feature = "http")]

mod common;

use axum::body::{to_bytes, Body};
use axum::http::{header, Method, Request, StatusCode};
use common::sse_parser::parse_sse_frames;
use klieo_a2a::handler::EchoHandler;
use klieo_a2a::http::A2aHttpServer;
use klieo_a2a::task_store::{A2aTaskStore, DEFAULT_BUCKET};
use klieo_a2a::types::{Task, TaskStatus};
use klieo_a2a::A2aDispatcher;
use klieo_auth_common::{AllowAnonymous, AuthError, BearerTokenAuthenticator};
use klieo_bus_memory::MemoryBus;
use serde_json::{json, Value};
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use tower::ServiceExt;

fn make_store(dispatcher: &A2aDispatcher) -> Arc<A2aTaskStore> {
    let bus = Arc::new(MemoryBus::new());
    Arc::new(
        A2aTaskStore::new(bus.kv.clone(), DEFAULT_BUCKET.into())
            .with_event_sink(dispatcher.event_sink()),
    )
}

fn build_server() -> Arc<A2aHttpServer> {
    let dispatcher = Arc::new(A2aDispatcher::with_in_process_pubsub(
        Arc::new(EchoHandler::default()),
        Arc::new(AllowAnonymous),
    ));
    let store = make_store(&dispatcher);
    Arc::new(A2aHttpServer::new(
        dispatcher,
        store,
        CancellationToken::new(),
    ))
}

async fn post_json(server: &Arc<A2aHttpServer>, body: Value) -> (StatusCode, Value) {
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    let status = resp.status();
    let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
    let v: Value = serde_json::from_slice(&bytes).unwrap_or(Value::Null);
    (status, v)
}

#[tokio::test]
async fn post_json_send_message_returns_envelope() {
    let server = build_server();
    let body = json!({
        "jsonrpc":"2.0","id":1,"method":"SendMessage",
        "params":{"message":{"messageId":"m-1","role":"user","parts":[{"type":"text","content":"hi"}]}}
    });
    let (status, resp) = post_json(&server, body).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(resp["jsonrpc"], "2.0");
    assert_eq!(resp["id"], 1);
    assert!(resp.get("result").is_some());
}

#[tokio::test]
async fn post_json_unknown_method_returns_method_not_found_envelope() {
    let server = build_server();
    let body = json!({"jsonrpc":"2.0","id":2,"method":"wat","params":{}});
    let (status, resp) = post_json(&server, body).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(resp["error"]["code"], -32601);
}

#[tokio::test]
async fn post_unauthenticated_returns_jsonrpc_minus_32001() {
    // Build a server with bearer-token auth, no Authorization header.
    let dispatcher = Arc::new(A2aDispatcher::with_in_process_pubsub(
        Arc::new(EchoHandler::default()),
        Arc::new(BearerTokenAuthenticator::new(|_token: &str| {
            Err(AuthError::Rejected("wrong token".into()))
        })),
    ));
    let store = make_store(&dispatcher);
    let server = Arc::new(A2aHttpServer::new(
        dispatcher,
        store,
        CancellationToken::new(),
    ));

    let body = json!({"jsonrpc":"2.0","id":3,"method":"SendMessage","params":{"message":{"messageId":"m-1","role":"user","parts":[{"type":"text","content":"hi"}]}}});
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        // No Authorization header → AuthError::Missing → -32001
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
    let resp_json: Value = serde_json::from_slice(&bytes).unwrap();
    assert_eq!(resp_json["error"]["code"], -32001);
}

#[tokio::test]
async fn post_malformed_json_returns_400() {
    let server = build_server();
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .body(Body::from(&b"not-json"[..]))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn post_wrong_content_type_returns_415() {
    let server = build_server();
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "text/plain")
        .body(Body::from(&b"{}"[..]))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
}

#[tokio::test]
async fn post_body_over_limit_returns_413() {
    let server = build_server();
    let big = vec![b'a'; 2 * 1024 * 1024];
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .body(Body::from(big))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE);
}

#[tokio::test]
async fn post_non_post_returns_405() {
    let server = build_server();
    let req = Request::builder()
        .method(Method::GET)
        .uri("/a2a")
        .body(Body::empty())
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
}

#[tokio::test]
async fn post_send_streaming_message_returns_sse_event_stream() {
    let server = build_server();
    let body = json!({
        "jsonrpc":"2.0","id":10,"method":"SendStreamingMessage",
        "params":{"message":{"messageId":"m-1","role":"user","parts":[{"type":"text","content":"hi"}]}}
    });
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .header(header::ACCEPT, "text/event-stream")
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let ct = resp
        .headers()
        .get(header::CONTENT_TYPE)
        .unwrap()
        .to_str()
        .unwrap();
    assert!(
        ct.starts_with("text/event-stream"),
        "expected SSE content type; got: {ct}"
    );
    let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
    let frames = parse_sse_frames(&bytes);
    assert!(!frames.is_empty(), "expected ≥1 SSE frame");
    // Last frame should carry final=true.
    let last: Value = serde_json::from_str(&frames.last().unwrap().data).unwrap();
    assert_eq!(last["result"]["final"], true);
}

#[tokio::test]
async fn post_subscribe_to_task_replays_then_tails() {
    let server = build_server();
    // Pre-populate a completed task so SubscribeToTask can replay it.
    let task_id = "task-subscribe-test";
    let task = Task {
        id: task_id.into(),
        contextId: "ctx-sub-test".into(),
        status: TaskStatus::Completed,
        artifacts: vec![],
        history: vec![],
        metadata: None,
    };
    server.task_store().put(&task).await.unwrap();

    // SubscribeToTask: completed task should replay one terminal event.
    let sub_body =
        json!({"jsonrpc":"2.0","id":2,"method":"SubscribeToTask","params":{"id":task_id}});
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .header(header::ACCEPT, "text/event-stream")
        .body(Body::from(serde_json::to_vec(&sub_body).unwrap()))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
    let frames = parse_sse_frames(&bytes);
    assert!(!frames.is_empty(), "expected ≥1 SSE frame on subscribe");
    let first: Value = serde_json::from_str(&frames[0].data).unwrap();
    assert_eq!(first["result"]["task_id"], task_id);
}

#[tokio::test]
async fn sse_terminates_after_final_event() {
    let server = build_server();
    let body = json!({
        "jsonrpc":"2.0","id":1,"method":"SendStreamingMessage",
        "params":{"message":{"messageId":"m-1","role":"user","parts":[{"type":"text","content":"hi"}]}}
    });
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .header(header::ACCEPT, "text/event-stream")
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
    let frames = parse_sse_frames(&bytes);
    let final_idx = frames
        .iter()
        .position(|f| {
            serde_json::from_str::<Value>(&f.data)
                .map(|v| v["result"]["final"] == true)
                .unwrap_or(false)
        })
        .expect("no final event found");
    assert_eq!(
        final_idx,
        frames.len() - 1,
        "frames after final event: {:?}",
        &frames[final_idx + 1..]
    );
}

#[tokio::test]
async fn parent_cancel_returns_503() {
    let cancel = CancellationToken::new();
    let dispatcher = Arc::new(A2aDispatcher::with_in_process_pubsub(
        Arc::new(EchoHandler::default()),
        Arc::new(AllowAnonymous),
    ));
    let store = make_store(&dispatcher);
    let server = Arc::new(A2aHttpServer::new(dispatcher, store, cancel.clone()));
    cancel.cancel();

    let body = json!({"jsonrpc":"2.0","id":1,"method":"SendMessage","params":{"message":{"messageId":"m-1","role":"user","parts":[{"type":"text","content":"hi"}]}}});
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);

    // Reload response to check envelope shape.
    let cancel2 = CancellationToken::new();
    let dispatcher2 = Arc::new(A2aDispatcher::with_in_process_pubsub(
        Arc::new(EchoHandler::default()),
        Arc::new(AllowAnonymous),
    ));
    let store2 = make_store(&dispatcher2);
    let server2 = Arc::new(A2aHttpServer::new(dispatcher2, store2, cancel2.clone()));
    cancel2.cancel();
    let req2 = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap();
    let resp2 = server2.router().oneshot(req2).await.unwrap();
    let bytes = to_bytes(resp2.into_body(), 1 << 20).await.unwrap();
    let envelope: Value = serde_json::from_slice(&bytes).unwrap();
    assert_eq!(
        envelope["error"]["code"], -32000,
        "shutdown envelope must carry code -32000; got: {envelope}"
    );
    assert!(
        envelope["error"]["message"]
            .as_str()
            .unwrap_or("")
            .contains("server shutting down"),
        "expected 'server shutting down' in envelope; got: {envelope}",
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn subscribe_to_task_broadcast_fallback_tails_store_writes() {
    let server = build_server();
    let task_id = "t-broadcast";

    // Seed an in-progress task so SubscribeToTask returns a non-terminal
    // replay event, leaving the stream open to receive broadcast events.
    let task = Task {
        id: task_id.into(),
        contextId: "ctx-broadcast".into(),
        status: TaskStatus::Working,
        artifacts: vec![],
        history: vec![],
        metadata: None,
    };
    server.task_store().put(&task).await.unwrap();

    // Spawn a writer that transitions the task to Completed 200ms after
    // the subscriber stream has been opened. The 200ms gap ensures the
    // SSE subscriber is attached before the broadcast fires.
    let store = server.task_store().clone();
    tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        let completed = Task {
            id: task_id.into(),
            contextId: "ctx-broadcast".into(),
            status: TaskStatus::Completed,
            artifacts: vec![],
            history: vec![],
            metadata: None,
        };
        let _ = store.put(&completed).await;
    });

    let sub_body = json!({
        "jsonrpc":"2.0","id":1,"method":"SubscribeToTask",
        "params":{"id": task_id}
    });
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .header(header::ACCEPT, "text/event-stream")
        .body(Body::from(serde_json::to_vec(&sub_body).unwrap()))
        .unwrap();

    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let ct = resp
        .headers()
        .get(header::CONTENT_TYPE)
        .unwrap()
        .to_str()
        .unwrap();
    assert!(
        ct.starts_with("text/event-stream"),
        "expected SSE; got: {ct}"
    );

    let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
    let frames = parse_sse_frames(&bytes);
    assert!(
        frames.len() >= 2,
        "expected ≥2 frames (replay + broadcast); got {}",
        frames.len()
    );
    let first: Value = serde_json::from_str(&frames[0].data).unwrap();
    assert_eq!(first["result"]["task_id"], task_id);
    assert_eq!(
        first["result"]["final"], false,
        "replay of Working must not be final"
    );
    let last: Value = serde_json::from_str(&frames.last().unwrap().data).unwrap();
    assert_eq!(
        last["result"]["final"], true,
        "last frame must carry final=true"
    );
}

/// Handler whose `send_message` creates a `Working` (non-terminal) task,
/// writes it to a shared store (so pubsub events are emitted), and
/// returns it. `send_streaming_message` returns `MethodNotFound` so the
/// dispatcher falls back to the pubsub-derived stream.
///
/// `event_sink` is wired to the dispatcher's pubsub after
/// construction via `WorkingTaskHandler::wire_sink`.
struct WorkingTaskHandler {
    kv: Arc<dyn klieo_core::KvStore>,
    event_sink: std::sync::OnceLock<klieo_a2a::TaskEventSink>,
}

impl WorkingTaskHandler {
    fn new(kv: Arc<dyn klieo_core::KvStore>) -> Self {
        Self {
            kv,
            event_sink: std::sync::OnceLock::new(),
        }
    }

    fn wire_sink(&self, sink: klieo_a2a::TaskEventSink) {
        let _ = self.event_sink.set(sink);
    }

    fn task_store(&self) -> A2aTaskStore {
        let store = A2aTaskStore::new(self.kv.clone(), DEFAULT_BUCKET.into());
        if let Some(sink) = self.event_sink.get() {
            store.with_event_sink(sink.clone())
        } else {
            store
        }
    }
}

#[async_trait::async_trait]
impl klieo_a2a::A2aHandler for WorkingTaskHandler {
    async fn send_message(
        &self,
        _ctx: &klieo_a2a::auth::RequestContext,
        params: klieo_a2a::types::SendMessageParams,
    ) -> Result<klieo_a2a::types::SendMessageResult, klieo_a2a::A2aError> {
        let task_id = format!("task-{}", uuid::Uuid::new_v4());
        let context_id = params
            .message
            .contextId
            .clone()
            .unwrap_or_else(|| "ctx-streaming".into());
        let task = Task {
            id: task_id,
            contextId: context_id,
            status: TaskStatus::Working,
            artifacts: vec![],
            history: vec![params.message],
            metadata: None,
        };
        self.task_store().put(&task).await?;
        Ok(klieo_a2a::types::SendMessageResult::Task(task))
    }
}

/// Build an `A2aHttpServer` wired to a `WorkingTaskHandler`.
///
/// The handler writes tasks as `Working` through a store that shares the
/// dispatcher's event sink, so the dispatcher's broadcast channel carries
/// the task events — enabling the broadcast-derived stream path.
fn build_test_streaming_server() -> Arc<A2aHttpServer> {
    let bus = Arc::new(MemoryBus::new());
    let handler = Arc::new(WorkingTaskHandler::new(bus.kv.clone()));
    let dispatcher = Arc::new(A2aDispatcher::with_in_process_pubsub(
        handler.clone(),
        Arc::new(AllowAnonymous),
    ));
    // Wire the handler's store to emit on the dispatcher's pubsub.
    handler.wire_sink(dispatcher.event_sink());
    let server_store = Arc::new(
        A2aTaskStore::new(bus.kv.clone(), DEFAULT_BUCKET.into())
            .with_event_sink(dispatcher.event_sink()),
    );
    Arc::new(A2aHttpServer::new(
        dispatcher,
        server_store,
        CancellationToken::new(),
    ))
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn send_streaming_message_broadcast_fallback_tails_store_writes() {
    let server = build_test_streaming_server();
    let store = server.task_store().clone();

    // POST SendStreamingMessage — the dispatcher falls back to send_message
    // (WorkingTaskHandler), which creates a Working task and emits it to the
    // broadcast channel. The stream stays open. A concurrent writer then
    // transitions the task to Completed via the same store.
    let body = json!({
        "jsonrpc":"2.0","id":1,"method":"SendStreamingMessage",
        "params":{"message":{"messageId":"m-1","role":"user",
            "parts":[{"type":"text","content":"hi"}]}}
    });

    // Spawn a writer that polls until a task appears in the store, then
    // transitions it to Completed.
    tokio::spawn(async move {
        for _ in 0..40 {
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            let tasks = store.list(Some("ctx-streaming")).await.unwrap_or_default();
            if let Some(task) = tasks.into_iter().next() {
                let completed = Task {
                    status: TaskStatus::Completed,
                    ..task
                };
                let _ = store.put(&completed).await;
                return;
            }
        }
    });

    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .header(header::ACCEPT, "text/event-stream")
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
    let frames = parse_sse_frames(&bytes);
    assert!(
        frames.len() >= 2,
        "expected ≥2 frames (initial Working + Completed broadcast); got {}",
        frames.len()
    );
    let last: Value = serde_json::from_str(&frames.last().unwrap().data).unwrap();
    assert_eq!(
        last["result"]["final"], true,
        "last frame must carry final=true; got: {last}"
    );
}

#[tokio::test]
async fn duplicate_authorization_header_returns_400() {
    let server = build_server();
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .header(header::AUTHORIZATION, "Bearer alpha")
        .header(header::AUTHORIZATION, "Bearer beta")
        .body(Body::from(
            serde_json::to_vec(&json!({
                "jsonrpc": "2.0", "id": 1, "method": "GetAgentCard"
            }))
            .unwrap(),
        ))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
    let body: Value = serde_json::from_slice(&bytes).unwrap();
    assert_eq!(body["error"]["code"], -32600);
    assert!(
        body["error"]["message"]
            .as_str()
            .unwrap()
            .to_lowercase()
            .contains("authorization"),
        "expected 'authorization' in error message; got: {body}"
    );
}

#[tokio::test]
async fn serve_http_refuses_non_loopback_bind_with_allow_anonymous() {
    use klieo_a2a::A2aError;
    use std::net::SocketAddr;

    let dispatcher = Arc::new(A2aDispatcher::with_in_process_pubsub(
        Arc::new(EchoHandler::default()),
        Arc::new(AllowAnonymous),
    ));
    let store = make_store(&dispatcher);
    let server = Arc::new(A2aHttpServer::new(
        dispatcher,
        store,
        CancellationToken::new(),
    ));
    let addr: SocketAddr = "0.0.0.0:0".parse().unwrap();

    let result = server.serve_http(addr).await;
    match result {
        Err(A2aError::Misconfigured(msg)) => {
            assert!(
                msg.to_lowercase().contains("non-loopback"),
                "expected 'non-loopback' in error; got: {msg}"
            );
            assert!(
                msg.to_lowercase().contains("anonymous"),
                "expected 'anonymous' in error; got: {msg}"
            );
        }
        other => panic!("expected Err(Misconfigured); got: {other:?}"),
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn serve_http_allow_public_bind_opt_in_overrides_refusal() {
    use std::net::SocketAddr;

    let dispatcher = Arc::new(A2aDispatcher::with_in_process_pubsub(
        Arc::new(EchoHandler::default()),
        Arc::new(AllowAnonymous),
    ));
    let store = make_store(&dispatcher);
    let cancel = CancellationToken::new();
    let cancel_for_drop = cancel.clone();
    // allow_public_bind() consumes Self — build before Arc wrapping.
    let server = Arc::new(A2aHttpServer::new(dispatcher, store, cancel).allow_public_bind());
    // Bind loopback (safe in tests) to confirm refusal guard doesn't fire.
    let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
    let handle = tokio::spawn(async move { server.serve_http(addr).await });
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    cancel_for_drop.cancel();
    // Any result other than Misconfigured is success.
    if let Err(klieo_a2a::A2aError::Misconfigured(msg)) = handle.await.unwrap() {
        panic!("unexpected Misconfigured with allow_public_bind: {msg}");
    }
}

#[tokio::test]
async fn auth_failure_during_streaming_does_not_open_sse_body() {
    let dispatcher = Arc::new(A2aDispatcher::with_in_process_pubsub(
        Arc::new(EchoHandler::default()),
        Arc::new(BearerTokenAuthenticator::new(|_token: &str| {
            Err(AuthError::Rejected("bad".into()))
        })),
    ));
    let store = make_store(&dispatcher);
    let server = Arc::new(A2aHttpServer::new(
        dispatcher,
        store,
        CancellationToken::new(),
    ));
    let body = json!({"jsonrpc":"2.0","id":1,"method":"SendStreamingMessage","params":{"message":{"messageId":"m-1","role":"user","parts":[{"type":"text","content":"hi"}]}}});
    let req = Request::builder()
        .method(Method::POST)
        .uri("/a2a")
        .header(header::CONTENT_TYPE, "application/json")
        .header(header::ACCEPT, "text/event-stream")
        // no Authorization header → auth failure
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap();
    let resp = server.router().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let ct = resp
        .headers()
        .get(header::CONTENT_TYPE)
        .unwrap()
        .to_str()
        .unwrap();
    assert!(
        ct.starts_with("application/json"),
        "expected JSON 401-equivalent envelope, not SSE; got: {ct}"
    );
}
