klieo-a2a 0.4.0

//! A2A v1.0 conformance suite.
//!
//! Drives an [`A2aHandler`] (plus a few in-crate fixtures) through every
//! contract clause documented in the [A2A v1.0
//! specification](https://a2a-protocol.org/latest/specification/) that
//! `klieo-a2a` claims to implement, collecting a per-clause pass/fail
//! result.
//!
//! The intent is *both* a regression guard for `klieo-a2a` itself *and*
//! a publishable, machine-verifiable claim ("klieo-a2a passes N/N
//! conformance cases against the v1.0 spec"). Each case carries the spec
//! §-reference it asserts so the report doubles as an audit table.
//!
//! Bring your own `A2aHandler` impl, then call [`run_conformance_suite`]:
//!
//! ```ignore
//! # async fn run<H: klieo_a2a::A2aHandler + Send + Sync>(handler: &H) {
//! let report = klieo_a2a::conformance::run_conformance_suite(handler).await;
//! println!("{report}");
//! assert_eq!(report.failed, 0);
//! # }
//! ```

use crate::auth::{Identity, RequestContext};
use crate::envelope::{A2aHeaders, A2aMethod, JsonRpcRequest, JsonRpcResponse};
use crate::error::A2aError;
use crate::handler::A2aHandler;
use crate::task_store::A2aTaskStore;
use crate::types::{
    CancelTaskParams, Message, Part, Role, SendMessageParams, SendMessageResult,
    SubscribeToTaskParams, Task, TaskStatus,
};
use klieo_bus_memory::MemoryKv;
use klieo_core::Headers;
use serde_json::json;
use std::borrow::Cow;
use std::fmt;
use std::time::Duration;

/// Outcome of a single conformance case.
///
/// Marked `#[non_exhaustive]` so additional outcome variants (e.g. a
/// future `Inconclusive`) can be added without a SemVer-major break.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum CaseStatus {
    /// The handler/crate satisfied the contract.
    Pass,
    /// The handler/crate violated the contract.
    Fail,
    /// The case was deliberately skipped (e.g. capability not implemented in v0.1).
    Skip(Cow<'static, str>),
}

impl CaseStatus {
    fn label(&self) -> &'static str {
        match self {
            CaseStatus::Pass => "PASS",
            CaseStatus::Fail => "FAIL",
            CaseStatus::Skip(_) => "SKIP",
        }
    }
}

/// One conformance case + its outcome.
///
/// Marked `#[non_exhaustive]` so future fields (per-case timing, severity,
/// machine-readable error code) can be added additively.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct ConformanceCase {
    /// Stable case id (`"C-01"` … `"C-12"`).
    pub id: &'static str,
    /// Spec clause this case asserts.
    pub clause: &'static str,
    /// Outcome.
    pub status: CaseStatus,
    /// Optional human-readable detail (failure reason or skip rationale).
    pub detail: Option<String>,
}

/// Full conformance report.
///
/// Marked `#[non_exhaustive]` so future fields (spec version, run timestamp)
/// can be added without breaking pattern-matching downstream.
#[derive(Debug, Clone, Default)]
#[non_exhaustive]
pub struct ConformanceReport {
    /// Number of cases with [`CaseStatus::Pass`].
    pub passed: u32,
    /// Number of cases with [`CaseStatus::Fail`].
    pub failed: u32,
    /// All cases in order.
    pub cases: Vec<ConformanceCase>,
}

impl ConformanceReport {
    /// Number of cases with [`CaseStatus::Skip`].
    pub fn skipped(&self) -> u32 {
        self.cases
            .iter()
            .filter(|c| matches!(c.status, CaseStatus::Skip(_)))
            .count() as u32
    }

    fn record(&mut self, case: ConformanceCase) {
        match case.status {
            CaseStatus::Pass => self.passed += 1,
            CaseStatus::Fail => self.failed += 1,
            CaseStatus::Skip(_) => {}
        }
        self.cases.push(case);
    }
}

impl fmt::Display for ConformanceReport {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let total = self.passed + self.failed;
        writeln!(
            f,
            "{}/{} passed (skips: {})",
            self.passed,
            total,
            self.skipped()
        )?;
        writeln!(f, "{:<6} {:<6} {:<14} detail", "case", "status", "clause")?;
        for c in &self.cases {
            let detail = c.detail.as_deref().unwrap_or("—");
            writeln!(
                f,
                "{:<6} {:<6} {:<14} {}",
                c.id,
                c.status.label(),
                c.clause,
                detail
            )?;
        }
        Ok(())
    }
}

/// Result helper: build a `Pass` case.
fn pass(id: &'static str, clause: &'static str) -> ConformanceCase {
    ConformanceCase {
        id,
        clause,
        status: CaseStatus::Pass,
        detail: None,
    }
}

/// Result helper: build a `Fail` case with a reason.
fn fail(id: &'static str, clause: &'static str, detail: impl Into<String>) -> ConformanceCase {
    ConformanceCase {
        id,
        clause,
        status: CaseStatus::Fail,
        detail: Some(detail.into()),
    }
}

/// Result helper: build a `Skip` case with a reason.
fn skip(
    id: &'static str,
    clause: &'static str,
    reason: impl Into<Cow<'static, str>>,
) -> ConformanceCase {
    let reason = reason.into();
    let detail = reason.to_string();
    ConformanceCase {
        id,
        clause,
        status: CaseStatus::Skip(reason),
        detail: Some(detail),
    }
}

/// Run the full v1.0 conformance suite against `handler`.
///
/// Cases that target the dispatcher / wire format are exercised inline
/// here without needing the handler. Cases that target handler behaviour
/// (`send_message`, `get_task`, `cancel_task`) drive `handler` directly.
/// Cases that target the task-store + streaming surface synthesise a
/// minimal `MemoryBus`-free fixture or mark `Skip` when the crate
/// genuinely does not yet implement the clause.
pub async fn run_conformance_suite<H: A2aHandler + Send + Sync>(handler: &H) -> ConformanceReport {
    let mut report = ConformanceReport {
        cases: Vec::with_capacity(12),
        ..ConformanceReport::default()
    };

    report.record(case_c01_default_a2a_version().await);
    report.record(case_c02_method_names_parse().await);
    report.record(case_c03_unknown_method_rejects().await);
    report.record(case_c04_id_null_accepted().await);
    report.record(case_c05_id_int_and_string_round_trip().await);
    report.record(case_c06_error_code_mapping().await);
    report.record(case_c07_idempotency_key_header_round_trip().await);
    report.record(case_c08_idempotency_replay().await);
    report.record(case_c09_streaming_subscription_in_order().await);
    report.record(case_c10_cancel_completes_with_canceled_status(handler).await);
    report.record(case_c11_task_store_put_get_round_trip().await);
    report.record(case_c12_malformed_json_is_parse_error().await);

    report
}

// ---- C-01 : §8 default A2A-Version ----
async fn case_c01_default_a2a_version() -> ConformanceCase {
    let h = Headers::new();
    let decoded = A2aHeaders::decode_from(&h);
    if decoded.a2a_version == "1.0" {
        pass("C-01", "§8")
    } else {
        fail(
            "C-01",
            "§8",
            format!(
                "A2A-Version default was `{}`, expected `1.0`",
                decoded.a2a_version
            ),
        )
    }
}

// ---- C-02 : §9.4 all 11 method names parse ----
async fn case_c02_method_names_parse() -> ConformanceCase {
    // The A2A v1.0 §9.4 wire alphabet is CamelCase. This case pins every
    // documented method name to its `A2aMethod` enum variant — additions
    // to the spec must surface as a Fail here, not silently as a Skip.
    let cases: [(&str, A2aMethod); 11] = [
        ("SendMessage", A2aMethod::SendMessage),
        ("SendStreamingMessage", A2aMethod::SendStreamingMessage),
        ("GetTask", A2aMethod::GetTask),
        ("ListTasks", A2aMethod::ListTasks),
        ("CancelTask", A2aMethod::CancelTask),
        ("SubscribeToTask", A2aMethod::SubscribeToTask),
        (
            "CreateTaskPushNotificationConfig",
            A2aMethod::CreateTaskPushNotificationConfig,
        ),
        (
            "GetTaskPushNotificationConfig",
            A2aMethod::GetTaskPushNotificationConfig,
        ),
        (
            "ListTaskPushNotificationConfigs",
            A2aMethod::ListTaskPushNotificationConfigs,
        ),
        (
            "DeleteTaskPushNotificationConfig",
            A2aMethod::DeleteTaskPushNotificationConfig,
        ),
        ("GetExtendedAgentCard", A2aMethod::GetExtendedAgentCard),
    ];
    for (wire, expected) in cases {
        match A2aMethod::from_str(wire) {
            Ok(got) if got == expected => continue,
            Ok(other) => {
                return fail(
                    "C-02",
                    "§9.4",
                    format!("`{wire}` parsed as `{other:?}`, expected `{expected:?}`"),
                );
            }
            Err(e) => {
                return fail("C-02", "§9.4", format!("`{wire}` failed to parse: {e}"));
            }
        }
    }
    pass("C-02", "§9.4")
}

// ---- C-03 : §9.4 unknown method names reject cleanly ----
async fn case_c03_unknown_method_rejects() -> ConformanceCase {
    match A2aMethod::from_str("Frobnicate") {
        Err(A2aError::MethodNotFound(name)) if name == "Frobnicate" => pass("C-03", "§9.4"),
        Err(other) => fail(
            "C-03",
            "§9.4",
            format!("expected MethodNotFound, got {other:?}"),
        ),
        Ok(m) => fail("C-03", "§9.4", format!("unknown name parsed as {m:?}")),
    }
}

// ---- C-04 : §10 id: null is valid (notification) ----
async fn case_c04_id_null_accepted() -> ConformanceCase {
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: serde_json::Value::Null,
        method: "SendMessage".into(),
        params: json!({}),
    };
    let v = match serde_json::to_value(&req) {
        Ok(v) => v,
        Err(e) => return fail("C-04", "§10", format!("encode: {e}")),
    };
    let back: Result<JsonRpcRequest, _> = serde_json::from_value(v);
    match back {
        Ok(r) if r.id.is_null() => pass("C-04", "§10"),
        Ok(r) => fail("C-04", "§10", format!("id was {:?}, expected null", r.id)),
        Err(e) => fail("C-04", "§10", format!("decode: {e}")),
    }
}

// ---- C-05 : §10 id round-trips int + string variants ----
async fn case_c05_id_int_and_string_round_trip() -> ConformanceCase {
    for id in [json!(42_i64), json!("req-abc")] {
        let req = JsonRpcRequest {
            jsonrpc: "2.0".into(),
            id: id.clone(),
            method: "SendMessage".into(),
            params: json!({}),
        };
        let v = match serde_json::to_value(&req) {
            Ok(v) => v,
            Err(e) => return fail("C-05", "§10", format!("encode {id:?}: {e}")),
        };
        let back: JsonRpcRequest = match serde_json::from_value(v) {
            Ok(b) => b,
            Err(e) => return fail("C-05", "§10", format!("decode {id:?}: {e}")),
        };
        if back.id != id {
            return fail(
                "C-05",
                "§10",
                format!("id round-trip mismatch: {:?} != {:?}", back.id, id),
            );
        }
    }
    pass("C-05", "§10")
}

// ---- C-06 : §11 error-code table ----
async fn case_c06_error_code_mapping() -> ConformanceCase {
    let id = json!(1);
    let cases: [(A2aError, i32, &str); 4] = [
        (
            A2aError::MethodNotFound("X".into()),
            -32601,
            "MethodNotFound",
        ),
        (A2aError::InvalidParams("X".into()), -32602, "InvalidParams"),
        (A2aError::Server("X".into()), -32000, "Server"),
        (A2aError::Unauthorized("X".into()), -32001, "Unauthorized"),
    ];
    for (err, want, name) in cases {
        let resp = err.to_json_rpc_error(id.clone());
        let Some(payload) = resp.error else {
            return fail("C-06", "§11", format!("{name}: missing error payload"));
        };
        if payload.code != want {
            return fail(
                "C-06",
                "§11",
                format!("{name}: code {} != expected {want}", payload.code),
            );
        }
    }
    // ParseError synthesised via serde_json failure.
    let parse_err = match serde_json::from_str::<u32>("not json") {
        Err(e) => A2aError::from(e),
        Ok(_) => return fail("C-06", "§11", "cannot synthesise serde_json parse error"),
    };
    let resp = parse_err.to_json_rpc_error(id);
    let Some(payload) = resp.error else {
        return fail("C-06", "§11", "ParseError: missing error payload");
    };
    if payload.code != -32700 {
        return fail(
            "C-06",
            "§11",
            format!("ParseError: code {} != -32700", payload.code),
        );
    }
    // Also assert server errors fall in the documented -32000..=-32099 band.
    let server_resp = A2aError::Server("x".into()).to_json_rpc_error(json!(2));
    let code = server_resp.error.unwrap().code;
    if !(-32099..=-32000).contains(&code) {
        return fail(
            "C-06",
            "§11",
            format!("Server: code {code} outside -32000..-32099 band"),
        );
    }
    pass("C-06", "§11")
}

// ---- C-07 : §12 idempotency-key header round-trips through Headers ----
async fn case_c07_idempotency_key_header_round_trip() -> ConformanceCase {
    let mut h = Headers::new();
    h.insert("Idempotency-Key".into(), "key-123".into());
    h.insert("A2A-Version".into(), "1.0".into());
    let decoded = A2aHeaders::decode_from(&h);
    if decoded.a2a_version != "1.0" {
        return fail(
            "C-07",
            "§12",
            format!("a2a_version `{}` != `1.0`", decoded.a2a_version),
        );
    }
    // Idempotency-Key isn't currently materialised onto A2aHeaders, but the
    // Headers map preserves it for the dispatcher and for handler-side logic.
    match h.get("Idempotency-Key") {
        Some(v) if v == "key-123" => pass("C-07", "§12"),
        Some(v) => fail(
            "C-07",
            "§12",
            format!("Idempotency-Key round-trip mismatch: `{v}` != `key-123`"),
        ),
        None => fail("C-07", "§12", "Idempotency-Key lost in Headers map"),
    }
}

// ---- C-08 : §12 idempotency replay returns same task_id ----
async fn case_c08_idempotency_replay() -> ConformanceCase {
    // The current `A2aServer` dispatcher does not yet maintain an
    // idempotency-key → task_id ledger. A handler MAY implement this
    // itself but the crate-level contract isn't in place yet.
    //
    // When the ledger lands, this case will need a handler reference
    // again — construct a private fixture rather than threading the
    // caller's `AppState` to avoid colliding with production keys.
    skip("C-08", "§12", "idempotency ledger not yet implemented")
}

// ---- C-09 : §13 streaming subscription returns chunks in order ----
async fn case_c09_streaming_subscription_in_order() -> ConformanceCase {
    // §13 describes SSE-style streaming chunks for `SendStreamingMessage`
    // and `SubscribeToTask`. The klieo-a2a v0.0.1 wire layer round-trips
    // single JSON-RPC responses; chunked streaming is gated behind
    // capability `streaming: false` on the default AgentCard. We assert
    // ordering at the layer we *do* have — a `SubscribeToTaskParams`
    // payload round-trips identically through serde — and mark the
    // chunked-delivery clause as Skip with a structural rationale.
    let p = SubscribeToTaskParams { id: "t-1".into() };
    let v = match serde_json::to_value(&p) {
        Ok(v) => v,
        Err(e) => return fail("C-09", "§13", format!("encode SubscribeToTaskParams: {e}")),
    };
    let back: SubscribeToTaskParams = match serde_json::from_value(v) {
        Ok(b) => b,
        Err(e) => return fail("C-09", "§13", format!("decode SubscribeToTaskParams: {e}")),
    };
    if back.id != "t-1" {
        return fail("C-09", "§13", "SubscribeToTaskParams.id lost in round-trip");
    }
    skip(
        "C-09",
        "§13",
        "chunked streaming delivery (SendStreamingMessage) not yet implemented",
    )
}

// ---- C-10 : §13 cancel_task returns Canceled status ----
//
// This is a STRUCTURAL contract check: send_message must yield a
// referenceable Task, and cancel_task on that id must return with
// status == Canceled. It is NOT a latency benchmark — the 2 s wrappers
// are hang detectors only. A handler whose `cancel_task` is no-op (e.g.
// `EchoHandler` flipping the in-memory enum) passes structurally; a
// handler that drives cancellation through a real work queue would need
// its own performance assertions in addition to this case.
async fn case_c10_cancel_completes_with_canceled_status<H: A2aHandler + Send + Sync>(
    handler: &H,
) -> ConformanceCase {
    let msg = Message {
        messageId: "m-cancel".into(),
        contextId: Some("ctx-cancel".into()),
        taskId: None,
        role: Role::User,
        parts: vec![Part::Text {
            content: "task body".into(),
            metadata: None,
            mediaType: None,
            filename: None,
        }],
        metadata: None,
        extensions: vec![],
        referenceTaskIds: vec![],
    };
    let ctx = RequestContext::new(
        A2aHeaders::decode_from(&klieo_core::Headers::new()),
        Some(Identity::anonymous()),
    );
    let sent = match tokio::time::timeout(
        Duration::from_secs(2),
        handler.send_message(
            &ctx,
            SendMessageParams {
                message: msg,
                configuration: None,
            },
        ),
    )
    .await
    {
        Ok(Ok(r)) => r,
        Ok(Err(A2aError::MethodNotFound(_))) => {
            return skip("C-10", "§13", "handler does not implement send_message")
        }
        Ok(Err(e)) => return fail("C-10", "§13", format!("send_message: {e}")),
        Err(_) => return fail("C-10", "§13", "send_message exceeded 2s budget"),
    };
    let task_id = match sent {
        SendMessageResult::Task(t) => t.id,
        SendMessageResult::Message(_) => {
            return skip(
                "C-10",
                "§13",
                "handler returned a Message; no task to cancel",
            )
        }
    };
    match tokio::time::timeout(
        Duration::from_secs(2),
        handler.cancel_task(&ctx, CancelTaskParams { id: task_id }),
    )
    .await
    {
        Ok(Ok(task)) if matches!(task.status, TaskStatus::Canceled) => pass("C-10", "§13"),
        Ok(Ok(task)) => fail(
            "C-10",
            "§13",
            format!(
                "cancel returned status {:?}, expected Canceled",
                task.status
            ),
        ),
        Ok(Err(A2aError::MethodNotFound(_))) => {
            skip("C-10", "§13", "handler does not implement cancel_task")
        }
        Ok(Err(e)) => fail("C-10", "§13", format!("cancel_task: {e}")),
        Err(_) => fail("C-10", "§13", "cancel_task exceeded 2s budget"),
    }
}

// ---- C-11 : §14 task_store put then get round-trips ----
async fn case_c11_task_store_put_get_round_trip() -> ConformanceCase {
    let kv = MemoryKv::new();
    let store = A2aTaskStore::new(std::sync::Arc::new(kv), "a2a.tasks".into());
    let task = Task {
        id: "conf-task-1".into(),
        contextId: "conf-ctx-1".into(),
        status: TaskStatus::Submitted,
        artifacts: vec![],
        history: vec![],
        metadata: None,
    };
    if let Err(e) = store.put(&task).await {
        return fail("C-11", "§14", format!("put: {e}"));
    }
    match store.get("conf-task-1").await {
        Ok(Some(back)) if back.id == "conf-task-1" && back.contextId == "conf-ctx-1" => {
            pass("C-11", "§14")
        }
        Ok(Some(back)) => fail(
            "C-11",
            "§14",
            format!("got id={} ctx={}", back.id, back.contextId),
        ),
        Ok(None) => fail("C-11", "§14", "task missing after put"),
        Err(e) => fail("C-11", "§14", format!("get: {e}")),
    }
}

// ---- C-12 : §15 malformed JSON → ParseError (-32700) ----
async fn case_c12_malformed_json_is_parse_error() -> ConformanceCase {
    // Mirror the dispatcher's first-step decode: any malformed body must
    // surface as a JSON-RPC ParseError envelope, never bubble up as Ok.
    let parse_result: Result<JsonRpcRequest, _> = serde_json::from_slice(b"not json {");
    let err = match parse_result {
        Err(e) => A2aError::from(e),
        Ok(_) => {
            return fail(
                "C-12",
                "§15",
                "malformed body decoded as JsonRpcRequest unexpectedly",
            )
        }
    };
    let envelope: JsonRpcResponse = err.to_json_rpc_error(serde_json::Value::Null);
    match envelope.error {
        Some(p) if p.code == -32700 => pass("C-12", "§15"),
        Some(p) => fail("C-12", "§15", format!("code {} != -32700", p.code)),
        None => fail("C-12", "§15", "missing error payload"),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn case_status_labels() {
        assert_eq!(CaseStatus::Pass.label(), "PASS");
        assert_eq!(CaseStatus::Fail.label(), "FAIL");
        assert_eq!(CaseStatus::Skip(Cow::Borrowed("x")).label(), "SKIP");
    }
}