//! A2A cluster-0.22 tenant-binding integration tests.
//!
//! Five replicas-on-one-bus scenarios pin the
//! `A2aDispatcherBuilder::with_tenant_binding` wiring + the ADR-022
//! gate on `dispatch_subscribe_to_task`:
//!
//! 1. Owner match → resume proceeds, no `-32001` envelope, body
//!    carries an SSE task-event frame.
//! 2. Owner mismatch → response shape matches the nonexistent-task
//!    response (both `-32000 internal server error` envelopes); no
//!    `-32001` (would leak existence info).
//! 3. No authenticator → tenant binding skipped, anonymous request
//!    succeeds even with a pre-claimed entry (partial-deployment
//!    posture).
//! 4. Legacy stream (no ownership entry) → `lookup` returns `None`
//!    → resume proceeds unbound.
//! 5. KV unavailable at `lookup` → fail-open per ADR-022 partial
//!    posture; subscribe still succeeds.
//!
//! Identity wiring uses the OAuth fixture (PEM + JWKS + mint_token)
//! copied verbatim from `tests/oauth.rs`; `mint_token` is parameterised
//! over `sub` so two principals (alice + bob) can be minted from one
//! shared JWKS endpoint.

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

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine;
use bytes::Bytes;
use http_body_util::BodyExt;
use jsonwebtoken::{encode, Algorithm, EncodingKey, Header};
use klieo_a2a::handler::EchoHandler;
use klieo_a2a::http::A2aHttpServer;
use klieo_a2a::server::A2aDispatcher;
use klieo_a2a::task_store::{A2aTaskStore, DEFAULT_BUCKET};
use klieo_a2a::types::{Task, TaskStatus};
use klieo_auth_common::{AllowAnonymous, Authenticator};
use klieo_auth_oauth::OAuthAuthenticator;
use klieo_bus_memory::MemoryBus;
use klieo_core::{BusError, KvEntry, KvStore, Lease, Pubsub, Revision};
use rsa::pkcs8::DecodePrivateKey;
use rsa::traits::PublicKeyParts;
use rsa::RsaPrivateKey;
use serde::Serialize;
use serde_json::json;
use tokio_util::sync::CancellationToken;
use tower::ServiceExt;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};

// Fixture PEM — fresh 2048-bit RSA throwaway key. Copied verbatim
// from `tests/oauth.rs` for self-contained cross-crate integration.
// Committed to source on purpose; it protects nothing.
const PEM_A: &str = "-----BEGIN PRIVATE KEY-----\n\
MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDbRAbiN8CgEaBd\n\
FKMynC4IJK30RVa9zCYeNo4gs7qZ1cRELDifv+cdWfhXAaTFl8bPCnLC2wXZAaAk\n\
OBz66UOAIEvz2X/tdIelu0LHTGbUMqnr5LV69UQ8KfNRFL+JGkMdEWsIawUfm6ch\n\
YcPV5UAQUAqNrDdGfVW7osSIIYP8gxU3Yr6bM2KANhz56fg7ajPFHp3kUkxQ6Z/9\n\
uE3ZFWYKg9WPGahzSf2q9yO+3zu1xWO0AbP+weNzecLqNhpGB8Od2+4gdAxIC1g/\n\
R2kvjJ3Pa1WCNNGjYdwCX3z/csDqLuMCmemEInnqMlEcJbS6Ltpg2rCZTPUCuQ9+\n\
aX4busPLAgMBAAECggEADHhLc0JxdDOxwj7CDfGflTtPdf9wMJ4eItoTw4dahZTd\n\
3coCtGR2Zo8Z1l+TuTW/Cuz7XC/sSWGpc7XfQWPSa28qx/pRfLXZCxNhzQeAJ5xJ\n\
WU7qBU1Sl/YDeP6BlXCxasNKqjFlRV1q/V/b19vQWhJ7RIBhVQ3GukObrWmejICQ\n\
VVutCepj+a+EyFUMeEI6X6TQi+IV1BQWQ8ahjv/oQIgKNCKB9BLnzXGO6aaz/8mU\n\
uRiW+XlkKdIlqsuq56eoS13CAn80AHdALv/0nuKb5+QfhKhcMe4uIqWITb+Gj1Ow\n\
kGdUR3SBOE0gJmXjZAOXhPogIqCwr4bW45PlmKxvcQKBgQD1dvBlgnF/NEgHXzhs\n\
emiAx7JR6VXt2DKU6Fio+nOAzw84p/f3LUKZh+XReLNcUyg1aT8HrucwlcH8j5CL\n\
At2g0oIILzu9OGajNBXySCvjploPH46VccuUhklLPG73GfikcNZm0C879LBqBnef\n\
Dkf/VElhovkUByeStf83jUHwsQKBgQDkrTnIG6t+Hli4SIAyrtCAEaNHBYIqAgiJ\n\
h8OUmAPnJe/lxudlVjmtTuXvETrzNoBYClACoguDQ6uy1CCVV3rGg7tthSn89anE\n\
pv7NFHcTWQTcU/tnLpLOXu4NLeOFBitsLmJPFAJl3ebiXeBsG09StHJHLvV5DC1I\n\
L/jpI+e7OwKBgA+iKaFtc2kUnVnK2L8f9LPvBA5vfAeP83jEa5Enkl+r0F5dZE+b\n\
1FgO7bYvp8zDhxYPw/HrU9ImV5qMUjpoTlFF1v5Kc9WTWAYg9RzfN9xh5e58pgoi\n\
m0K0Ysql6LNYNhvTC/S0Ya3B4dJ+4vOAZ1l9tsa/KHn8EqsnaD/TgVWhAoGBAMZQ\n\
H9MVg/Tc1m4CmgLaLgFSynm2IjrSBptWvNtptcDI2rI3e8cbLuvIh4l7NNdlRYqU\n\
buzJoRhWIFn2sscw1U3BQlLpcA9q8K6VcFVj0RxbT+bPgi/qaPBFzUZqrGxZfGDW\n\
HPWdUbML7MYVSaIXUQRVbuGirMksMTzwNXniZH2vAoGBALeG/u/alf7cIZxmnemE\n\
sq2N7wJz+JDHKujxwXl40P0dz9D9kYIVRhHA6TVhp/8oxXJTfMuTfWFfzBK5EMco\n\
MTQjW/kWlJQe/nS1TQ+IO+ZY5E4vc35S5hGZJtKUmTAfiBwQmswg/4KefjDRyRgV\n\
bBz+ZVw8j0jxC77vKnnplmPc\n\
-----END PRIVATE KEY-----\n";

const KID_A: &str = "test-key-a";
const ISSUER: &str = "https://issuer.example.test";
const AUDIENCE: &str = "klieo-a2a-test";
const TASK_ID: &str = "t-1";
const ALICE: &str = "alice@example.test";
const BOB: &str = "bob@example.test";
const UNAUTHENTICATED_CODE: i32 = -32001;
const SERVER_ERROR_CODE: i32 = -32000;
const TOKEN_TTL_SECS: u64 = 300;
const NBF_SKEW_SECS: u64 = 5;
const BODY_COLLECT_TIMEOUT_MS: u64 = 1500;
const TERMINAL_DRAIN_DELAY_MS: u64 = 40;

#[derive(Serialize)]
struct TestClaims {
    iss: String,
    sub: String,
    aud: String,
    exp: u64,
    nbf: u64,
    iat: u64,
    scope: String,
}

fn now_secs() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .expect("system time before unix epoch")
        .as_secs()
}

fn jwk_from_pem(pem: &str, kid: &str) -> serde_json::Value {
    let private_key = RsaPrivateKey::from_pkcs8_pem(pem).expect("parse RSA PEM");
    let n = URL_SAFE_NO_PAD.encode(private_key.n().to_bytes_be());
    let e = URL_SAFE_NO_PAD.encode(private_key.e().to_bytes_be());
    json!({
        "kty": "RSA",
        "use": "sig",
        "alg": "RS256",
        "kid": kid,
        "n": n,
        "e": e,
    })
}

async fn serve_jwks(jwks: serde_json::Value) -> (MockServer, String) {
    let server = MockServer::start().await;
    Mock::given(method("GET"))
        .and(path("/jwks"))
        .respond_with(ResponseTemplate::new(200).set_body_json(jwks))
        .mount(&server)
        .await;
    let uri = format!("{}/jwks", server.uri());
    (server, uri)
}

/// Mint a JWT for `sub` with the given `scopes`. Distinct from the
/// one in `tests/oauth.rs` only in that `sub` is a parameter so the
/// suite can mint two principals from the same JWKS endpoint.
fn mint_token(sub: &str, scopes: &[&str]) -> String {
    let now = now_secs();
    let claims = TestClaims {
        iss: ISSUER.into(),
        sub: sub.into(),
        aud: AUDIENCE.into(),
        exp: now + TOKEN_TTL_SECS,
        nbf: now - NBF_SKEW_SECS,
        iat: now,
        scope: scopes.join(" "),
    };
    let mut header = Header::new(Algorithm::RS256);
    header.kid = Some(KID_A.into());
    let encoding_key =
        EncodingKey::from_rsa_pem(PEM_A.as_bytes()).expect("EncodingKey::from_rsa_pem");
    encode(&header, &claims, &encoding_key).expect("sign JWT")
}

fn scope_map() -> HashMap<String, Vec<String>> {
    HashMap::from([
        ("SendMessage".to_string(), vec!["klieo:write".to_string()]),
        ("GetTask".to_string(), vec!["klieo:read".to_string()]),
        (
            "SubscribeToTask".to_string(),
            vec!["klieo:read".to_string()],
        ),
    ])
}

async fn build_oauth_authenticator(jwks_uri: String) -> OAuthAuthenticator {
    OAuthAuthenticator::builder()
        .issuer_url(ISSUER)
        .audience(AUDIENCE)
        .jwks_uri(jwks_uri)
        .required_scopes(scope_map())
        .build()
        .await
        .expect("OAuthAuthenticator builds against mock JWKS")
}

fn subscribe_body(task_id: &str) -> Vec<u8> {
    serde_json::to_vec(&json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "SubscribeToTask",
        "params": { "id": task_id }
    }))
    .expect("encode SubscribeToTask body")
}

fn seed_task(task_id: &str) -> Task {
    Task {
        id: task_id.into(),
        contextId: "ctx-1".into(),
        status: TaskStatus::Submitted,
        artifacts: vec![],
        history: vec![],
        metadata: None,
    }
}

fn terminal_task(task_id: &str) -> Task {
    Task {
        id: task_id.into(),
        contextId: "ctx-1".into(),
        status: TaskStatus::Completed,
        artifacts: vec![],
        history: vec![],
        metadata: None,
    }
}

async fn collect_body_text(resp: axum::response::Response) -> String {
    let bytes = tokio::time::timeout(
        Duration::from_millis(BODY_COLLECT_TIMEOUT_MS),
        resp.into_body().collect(),
    )
    .await
    .expect("body collect timeout")
    .expect("body collect failed")
    .to_bytes();
    String::from_utf8(bytes.to_vec()).expect("body is utf-8")
}

/// Build a dispatcher with both leader-election + tenant-binding wired
/// against the same KV bucket store; OAuth authenticator enforced.
fn build_dispatcher_with_oauth(
    pubsub: Arc<dyn Pubsub>,
    kv: Arc<dyn KvStore>,
    auth: OAuthAuthenticator,
) -> Arc<A2aDispatcher> {
    let handler: Arc<dyn klieo_a2a::handler::A2aHandler> = Arc::new(EchoHandler::default());
    let authenticator: Arc<dyn Authenticator> = Arc::new(auth);
    A2aDispatcher::builder()
        .handler(handler)
        .authenticator(authenticator)
        .pubsub(pubsub)
        .with_leader_election(kv.clone())
        .with_tenant_binding(kv)
        .build_arc()
        .expect("test builder")
}

/// Stand up an HTTP server wired around a dispatcher + a task store
/// backed by `task_kv`. Tenant-binding lives on the dispatcher's
/// `tenant_kv` (passed separately because the KV-unavailable test
/// uses different stores).
fn build_http_server(
    dispatcher: Arc<A2aDispatcher>,
    task_kv: Arc<dyn KvStore>,
) -> Arc<A2aHttpServer> {
    let task_store = Arc::new(
        A2aTaskStore::new(task_kv, DEFAULT_BUCKET.into()).with_event_sink(dispatcher.event_sink()),
    );
    Arc::new(A2aHttpServer::new(
        dispatcher,
        task_store,
        CancellationToken::new(),
    ))
}

fn bearer_subscribe_request(token: &str, task_id: &str) -> Request<Body> {
    Request::post("/a2a")
        .header("content-type", "application/json")
        .header("authorization", format!("Bearer {token}"))
        .body(Body::from(subscribe_body(task_id)))
        .expect("build request")
}

fn anonymous_subscribe_request(task_id: &str) -> Request<Body> {
    Request::post("/a2a")
        .header("content-type", "application/json")
        .body(Body::from(subscribe_body(task_id)))
        .expect("build request")
}

/// Schedule a terminal task transition so the SSE body closes cleanly
/// after the snapshot/tail loop sees a `final_event = true`. Without
/// this the body would block until the client disconnects.
fn schedule_terminal(task_store: Arc<A2aTaskStore>, task_id: &'static str) {
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(TERMINAL_DRAIN_DELAY_MS)).await;
        task_store.put(&terminal_task(task_id)).await.ok();
    });
}

fn assert_no_unauthenticated_envelope(text: &str, context: &str) {
    let marker = format!("\"code\":{UNAUTHENTICATED_CODE}");
    assert!(
        !text.contains(&marker),
        "{context}: response contains -32001 UNAUTHENTICATED envelope (would leak \
         existence info); body: {text}",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn owner_match_resume_succeeds() {
    let (_jwks_server, jwks_uri) =
        serve_jwks(json!({ "keys": [jwk_from_pem(PEM_A, KID_A)] })).await;
    let authn = build_oauth_authenticator(jwks_uri).await;

    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let kv: Arc<dyn KvStore> = bus.kv.clone();
    let dispatcher = build_dispatcher_with_oauth(pubsub, kv.clone(), authn);

    // Pre-claim ownership for alice on TASK_ID and keep the handle in
    // scope; without the binding the lookup returns None and the gate
    // becomes a no-op.
    let _ownership = dispatcher
        .ownership_registry()
        .expect("ownership registry wired")
        .claim(format!("a2a.{TASK_ID}"), ALICE.into())
        .await
        .expect("claim ownership");

    let server = build_http_server(dispatcher, kv);
    server.task_store().put(&seed_task(TASK_ID)).await.unwrap();
    schedule_terminal(server.task_store().clone(), TASK_ID);

    let token = mint_token(ALICE, &["klieo:read"]);
    let resp = server
        .router()
        .oneshot(bearer_subscribe_request(&token, TASK_ID))
        .await
        .expect("oneshot");
    assert_eq!(resp.status(), StatusCode::OK);

    let text = collect_body_text(resp).await;
    assert_no_unauthenticated_envelope(&text, "owner_match_resume_succeeds");
    let server_error_marker = format!("\"code\":{SERVER_ERROR_CODE}");
    assert!(
        !text.contains(&server_error_marker),
        "owner match must NOT surface a -32000 TaskNotFound envelope; body: {text}",
    );
    // Snapshot+tail emits at least one SSE `data:` frame for the
    // seeded Submitted task before the scheduled Completed transition
    // closes the stream.
    assert!(
        text.contains("data:") && text.contains(TASK_ID),
        "owner match must surface an SSE task-event frame; body: {text}",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn owner_mismatch_returns_task_not_found() {
    let (_jwks_server, jwks_uri) =
        serve_jwks(json!({ "keys": [jwk_from_pem(PEM_A, KID_A)] })).await;
    let authn_mismatch = build_oauth_authenticator(jwks_uri.clone()).await;
    let bus_mismatch = MemoryBus::new();
    let pubsub_mismatch: Arc<dyn Pubsub> = bus_mismatch.pubsub.clone();
    let kv_mismatch: Arc<dyn KvStore> = bus_mismatch.kv.clone();
    let dispatcher_mismatch =
        build_dispatcher_with_oauth(pubsub_mismatch, kv_mismatch.clone(), authn_mismatch);

    // Alice claims ownership of TASK_ID; bob will subscribe with the
    // same id and get denied.
    let _alice_handle = dispatcher_mismatch
        .ownership_registry()
        .expect("ownership registry wired")
        .claim(format!("a2a.{TASK_ID}"), ALICE.into())
        .await
        .expect("claim ownership for alice");

    let server_mismatch = build_http_server(dispatcher_mismatch, kv_mismatch);
    server_mismatch
        .task_store()
        .put(&seed_task(TASK_ID))
        .await
        .unwrap();

    let bob_token = mint_token(BOB, &["klieo:read"]);
    let mismatch_resp = server_mismatch
        .router()
        .oneshot(bearer_subscribe_request(&bob_token, TASK_ID))
        .await
        .expect("oneshot mismatch");
    assert_eq!(mismatch_resp.status(), StatusCode::OK);
    let mismatch_text = collect_body_text(mismatch_resp).await;

    // Independent dispatcher + bus for the nonexistent-task probe so
    // the two responses are wire-shape comparable without state leaks
    // between scenarios.
    let authn_nonexistent = build_oauth_authenticator(jwks_uri).await;
    let bus_nonexistent = MemoryBus::new();
    let pubsub_nonexistent: Arc<dyn Pubsub> = bus_nonexistent.pubsub.clone();
    let kv_nonexistent: Arc<dyn KvStore> = bus_nonexistent.kv.clone();
    let dispatcher_nonexistent = build_dispatcher_with_oauth(
        pubsub_nonexistent,
        kv_nonexistent.clone(),
        authn_nonexistent,
    );
    let server_nonexistent = build_http_server(dispatcher_nonexistent, kv_nonexistent);
    // No task seeded; no ownership entry.
    let nonexistent_resp = server_nonexistent
        .router()
        .oneshot(bearer_subscribe_request(&bob_token, TASK_ID))
        .await
        .expect("oneshot nonexistent");
    assert_eq!(nonexistent_resp.status(), StatusCode::OK);
    let nonexistent_text = collect_body_text(nonexistent_resp).await;

    // Crucial IDOR property: mismatch must NOT return -32001 (would
    // leak existence info; an attacker could distinguish "exists but
    // not mine" from "doesn't exist" by the auth-vs-not-found split).
    assert_no_unauthenticated_envelope(&mismatch_text, "owner_mismatch (mismatch)");
    assert_no_unauthenticated_envelope(&nonexistent_text, "owner_mismatch (nonexistent)");

    // Wire-shape parity: both paths must surface a -32000 SERVER_ERROR
    // envelope (TaskNotFound → http.rs sanitises to "internal server
    // error"). Request ids differ but the code + message do not.
    let server_error_marker = format!("\"code\":{SERVER_ERROR_CODE}");
    assert!(
        mismatch_text.contains(&server_error_marker),
        "mismatch must surface -32000 TaskNotFound envelope; body: {mismatch_text}",
    );
    assert!(
        nonexistent_text.contains(&server_error_marker),
        "nonexistent must surface -32000 TaskNotFound envelope; body: {nonexistent_text}",
    );
    assert!(
        mismatch_text.contains("internal server error"),
        "mismatch envelope must carry sanitised message; body: {mismatch_text}",
    );
    assert!(
        nonexistent_text.contains("internal server error"),
        "nonexistent envelope must carry sanitised message; body: {nonexistent_text}",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn no_authenticator_skips_binding() {
    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let kv: Arc<dyn KvStore> = bus.kv.clone();
    let handler: Arc<dyn klieo_a2a::handler::A2aHandler> = Arc::new(EchoHandler::default());
    let authenticator: Arc<dyn Authenticator> = Arc::new(AllowAnonymous);
    let dispatcher = A2aDispatcher::builder()
        .handler(handler)
        .authenticator(authenticator)
        .pubsub(pubsub)
        .with_tenant_binding(kv.clone())
        .build_arc()
        .expect("test builder");

    // Pre-claim ownership for alice; the anonymous request below must
    // still proceed because the gate skips when the caller is
    // anonymous (no principal to compare).
    let _ownership = dispatcher
        .ownership_registry()
        .expect("ownership registry wired")
        .claim(format!("a2a.{TASK_ID}"), ALICE.into())
        .await
        .expect("claim ownership");

    let server = build_http_server(dispatcher, kv);
    server.task_store().put(&seed_task(TASK_ID)).await.unwrap();
    schedule_terminal(server.task_store().clone(), TASK_ID);

    let resp = server
        .router()
        .oneshot(anonymous_subscribe_request(TASK_ID))
        .await
        .expect("oneshot");
    assert_eq!(resp.status(), StatusCode::OK);

    let text = collect_body_text(resp).await;
    assert_no_unauthenticated_envelope(&text, "no_authenticator_skips_binding");
    assert!(
        text.contains("data:") && text.contains(TASK_ID),
        "anonymous bypass must surface task-event SSE frame; body: {text}",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn legacy_stream_without_owner_entry_resumes_normally() {
    let (_jwks_server, jwks_uri) =
        serve_jwks(json!({ "keys": [jwk_from_pem(PEM_A, KID_A)] })).await;
    let authn = build_oauth_authenticator(jwks_uri).await;

    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let kv: Arc<dyn KvStore> = bus.kv.clone();
    let dispatcher = build_dispatcher_with_oauth(pubsub, kv.clone(), authn);
    // Deliberately NO ownership claim — simulating a pre-0.22 stream
    // already in flight when the operator upgraded.

    let server = build_http_server(dispatcher, kv);
    server.task_store().put(&seed_task(TASK_ID)).await.unwrap();
    schedule_terminal(server.task_store().clone(), TASK_ID);

    let token = mint_token(ALICE, &["klieo:read"]);
    let resp = server
        .router()
        .oneshot(bearer_subscribe_request(&token, TASK_ID))
        .await
        .expect("oneshot");
    assert_eq!(resp.status(), StatusCode::OK);

    let text = collect_body_text(resp).await;
    assert_no_unauthenticated_envelope(&text, "legacy_stream");
    let server_error_marker = format!("\"code\":{SERVER_ERROR_CODE}");
    assert!(
        !text.contains(&server_error_marker),
        "legacy stream must NOT surface -32000 TaskNotFound; body: {text}",
    );
    assert!(
        text.contains("data:") && text.contains(TASK_ID),
        "legacy stream must surface task-event SSE frame; body: {text}",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn kv_unavailable_at_lookup_degrades_to_pre_cluster_behavior() {
    let (_jwks_server, jwks_uri) =
        serve_jwks(json!({ "keys": [jwk_from_pem(PEM_A, KID_A)] })).await;
    let authn = build_oauth_authenticator(jwks_uri).await;

    // Tenant KV always errors; task store + bus use a healthy in-mem
    // KV so the snapshot/tail path runs unaffected.
    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let healthy_kv: Arc<dyn KvStore> = bus.kv.clone();
    let failing_kv: Arc<dyn KvStore> = Arc::new(FailingKv);

    let handler: Arc<dyn klieo_a2a::handler::A2aHandler> = Arc::new(EchoHandler::default());
    let authenticator: Arc<dyn Authenticator> = Arc::new(authn);
    let dispatcher = A2aDispatcher::builder()
        .handler(handler)
        .authenticator(authenticator)
        .pubsub(pubsub)
        .with_tenant_binding(failing_kv)
        .build_arc()
        .expect("test builder");

    let server = build_http_server(dispatcher, healthy_kv);
    server.task_store().put(&seed_task(TASK_ID)).await.unwrap();
    schedule_terminal(server.task_store().clone(), TASK_ID);

    let token = mint_token(ALICE, &["klieo:read"]);
    let resp = server
        .router()
        .oneshot(bearer_subscribe_request(&token, TASK_ID))
        .await
        .expect("oneshot");
    assert_eq!(resp.status(), StatusCode::OK);

    let text = collect_body_text(resp).await;
    assert_no_unauthenticated_envelope(&text, "kv_unavailable_at_lookup");
    let server_error_marker = format!("\"code\":{SERVER_ERROR_CODE}");
    assert!(
        !text.contains(&server_error_marker),
        "KV failure at lookup must fail open; -32000 must NOT surface; body: {text}",
    );
    assert!(
        text.contains("data:") && text.contains(TASK_ID),
        "fail-open path must surface task-event SSE frame; body: {text}",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn strict_kv_unavailable_at_lookup_denies_fail_closed() {
    let (_jwks_server, jwks_uri) =
        serve_jwks(json!({ "keys": [jwk_from_pem(PEM_A, KID_A)] })).await;
    let authn = build_oauth_authenticator(jwks_uri).await;

    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let healthy_kv: Arc<dyn KvStore> = bus.kv.clone();
    let failing_kv: Arc<dyn KvStore> = Arc::new(FailingKv);

    let handler: Arc<dyn klieo_a2a::handler::A2aHandler> = Arc::new(EchoHandler::default());
    let authenticator: Arc<dyn Authenticator> = Arc::new(authn);
    // `with_tenant_binding_strict`: a KV error at lookup must DENY, not proceed.
    let dispatcher = A2aDispatcher::builder()
        .handler(handler)
        .authenticator(authenticator)
        .pubsub(pubsub)
        .with_tenant_binding_strict(failing_kv)
        .build_arc()
        .expect("test builder");

    let server = build_http_server(dispatcher, healthy_kv);
    server.task_store().put(&seed_task(TASK_ID)).await.unwrap();
    schedule_terminal(server.task_store().clone(), TASK_ID);

    let token = mint_token(ALICE, &["klieo:read"]);
    let resp = server
        .router()
        .oneshot(bearer_subscribe_request(&token, TASK_ID))
        .await
        .expect("oneshot");

    let text = collect_body_text(resp).await;
    assert_no_unauthenticated_envelope(&text, "strict_kv_unavailable_at_lookup");
    let server_error_marker = format!("\"code\":{SERVER_ERROR_CODE}");
    assert!(
        text.contains(&server_error_marker),
        "strict tenant binding must FAIL CLOSED on KV error at lookup (-32000); body: {text}",
    );
    assert!(
        !text.contains(TASK_ID),
        "a strict deny must NOT surface any task-event SSE frame; body: {text}",
    );
}

/// `KvStore` test double whose every operation fails with
/// `BusError::Connection`. Used to verify ADR-022 fail-open on the
/// tenant-binding `lookup` path. Mirrors the `FailingKv` pattern in
/// `tests/leader_election.rs`; not extracted to a shared module
/// because the test scaffolds intentionally stay verbatim per the
/// crate's integration-test convention.
struct FailingKv;

#[async_trait]
impl KvStore for FailingKv {
    async fn get(&self, _bucket: &str, _key: &str) -> Result<Option<KvEntry>, BusError> {
        Err(BusError::Connection("simulated".into()))
    }
    async fn put(&self, _bucket: &str, _key: &str, _value: Bytes) -> Result<Revision, BusError> {
        Err(BusError::Connection("simulated".into()))
    }
    async fn cas(
        &self,
        _bucket: &str,
        _key: &str,
        _value: Bytes,
        _expected: Option<Revision>,
    ) -> Result<Revision, BusError> {
        Err(BusError::Connection("simulated".into()))
    }
    async fn delete(&self, _bucket: &str, _key: &str) -> Result<(), BusError> {
        Err(BusError::Connection("simulated".into()))
    }
    async fn lease(&self, _bucket: &str, _key: &str, _ttl: Duration) -> Result<Lease, BusError> {
        Err(BusError::Connection("simulated".into()))
    }
}
