//! Protocol violation evidence collected by the client pump.
//!
//! Why this exists (WOR-1311): mcptest is conformance tooling, so
//! evidence of peer misbehavior on the wire (string-id echoes, orphan
//! responses, malformed frames) must survive to the report instead of
//! being eaten by the pump. Before this module, a server echoing the
//! numeric id 1 as the string "1" produced an opaque RequestTimeout
//! with no trace of the actual bug.

use std::collections::{BTreeMap, HashSet, VecDeque};
use std::fmt;

use super::message::JsonRpcMessage;

/// Cap on retained violation examples per connection. The total counter
/// keeps counting past the cap, so a chatty broken server cannot grow
/// memory without bound while reports still see the real magnitude.
const EXAMPLE_CAP: usize = 32;

/// Bytes of an offending frame kept as evidence. Enough to identify the
/// frame, small enough that a hostile peer cannot bloat the recorder.
const SNIPPET_BYTES: usize = 256;

/// Classes of wire misbehavior the pump can diagnose without aborting
/// the connection. An enum rather than a free-form string so reporters
/// can group findings and the runner can match on specific kinds.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum ProtocolViolationKind {
    /// The server echoed a numeric request id back as a JSON string.
    StringIdEcho,
    /// A response arrived whose id matched no in-flight request.
    OrphanResponse,
    /// A frame failed JSON-RPC envelope classification.
    MalformedFrame,
}

impl fmt::Display for ProtocolViolationKind {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            ProtocolViolationKind::StringIdEcho => "string-id echo",
            ProtocolViolationKind::OrphanResponse => "orphan response",
            ProtocolViolationKind::MalformedFrame => "malformed frame",
        })
    }
}

/// One recorded violation: the kind plus a short human-readable detail
/// (offending id, frame snippet, parse error). Kept small on purpose;
/// full frames do not belong in a per-connection ring buffer.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProtocolViolation {
    /// Which class of misbehavior was observed.
    pub kind: ProtocolViolationKind,
    /// Short evidence string a reporter can print verbatim.
    pub detail: String,
}

/// Snapshot of every violation recorded on one connection. `total` can
/// exceed `examples.len()` because the example ring is capped.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProtocolViolationSummary {
    /// Count of every violation observed, including evicted examples.
    pub total: u64,
    /// Up to the most recent 32 violations, oldest first.
    pub examples: Vec<ProtocolViolation>,
    /// Exact per-kind totals, in enum order. Kept separately from the
    /// example ring because the ring is capped: a chatty broken server
    /// can evict every example of one kind while the report still owes
    /// the operator the true magnitude per kind (WOR-1311).
    pub kind_counts: Vec<(ProtocolViolationKind, u64)>,
}

impl ProtocolViolationSummary {
    /// Render the hint attached to timeout diagnostics, or `None` when
    /// the connection was clean. Why: a request timeout on a connection
    /// with recorded violations is usually the violation's fault, and
    /// the hint points the user at the protocol findings instead of
    /// leaving them to suspect their own timeout setting. Returned bare
    /// (no decoration) so the consumer decides the framing: the client
    /// wraps it in `(note: ...)` on [`super::ProtocolError`] Display.
    pub fn timeout_hint(&self) -> Option<String> {
        if self.total == 0 {
            return None;
        }
        let kinds: Vec<String> = self
            .kind_counts
            .iter()
            .map(|(kind, _)| kind.to_string())
            .collect();
        let plural = if self.total == 1 { "" } else { "s" };
        Some(format!(
            "{} protocol violation{} observed on this connection, including {}; \
             the run's protocol findings may explain this timeout",
            self.total,
            plural,
            kinds.join(", ")
        ))
    }
}

/// Mutable recorder the pump writes into, shared behind the same
/// `Arc<Mutex<...>>` pattern as the in-flight map.
#[derive(Debug, Default)]
pub(crate) struct ViolationRecorder {
    total: u64,
    examples: VecDeque<ProtocolViolation>,
    warned: HashSet<ProtocolViolationKind>,
    kind_totals: BTreeMap<ProtocolViolationKind, u64>,
}

impl ViolationRecorder {
    /// Record one violation: bump the total, keep the example (evicting
    /// the oldest past the cap), and warn once per kind so logs flag
    /// the problem without flooding on a repeatedly broken server.
    pub(crate) fn record(&mut self, kind: ProtocolViolationKind, detail: String) {
        self.total += 1;
        *self.kind_totals.entry(kind).or_insert(0) += 1;
        if self.warned.insert(kind) {
            tracing::warn!(
                target: "mcptest_core::protocol",
                kind = %kind,
                detail = %detail,
                "protocol violation observed (further violations of this kind are recorded silently)"
            );
        }
        if self.examples.len() == EXAMPLE_CAP {
            self.examples.pop_front();
        }
        self.examples.push_back(ProtocolViolation { kind, detail });
    }

    /// Snapshot the recorder for the accessor and the timeout hint.
    pub(crate) fn summary(&self) -> ProtocolViolationSummary {
        ProtocolViolationSummary {
            total: self.total,
            examples: self.examples.iter().cloned().collect(),
            kind_counts: self.kind_totals.iter().map(|(k, n)| (*k, *n)).collect(),
        }
    }
}

/// Render a bounded snippet of an offending frame for violation
/// evidence (WOR-1311). Bounded so a hostile or buggy peer cannot bloat
/// the recorder with megabyte frames.
pub(crate) fn frame_snippet(msg: &JsonRpcMessage) -> String {
    // Serializing our own envelope type cannot realistically fail; the
    // placeholder keeps the no-unwrap rule without losing the event.
    let raw = serde_json::to_string(msg).unwrap_or_else(|_| "<unserializable frame>".to_owned());
    if raw.len() <= SNIPPET_BYTES {
        return raw;
    }
    let mut end = SNIPPET_BYTES;
    while !raw.is_char_boundary(end) {
        end -= 1;
    }
    format!("{}... ({} bytes total)", &raw[..end], raw.len())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::protocol::JsonRpcId;
    use serde_json::json;

    #[test]
    fn example_ring_caps_at_32_while_total_keeps_counting() {
        let mut rec = ViolationRecorder::default();
        for i in 0..40 {
            rec.record(ProtocolViolationKind::OrphanResponse, format!("v{i}"));
        }
        let summary = rec.summary();
        assert_eq!(summary.total, 40);
        assert_eq!(summary.examples.len(), 32);
        // The eight oldest examples were evicted; the ring holds 8..40.
        assert_eq!(summary.examples[0].detail, "v8");
        assert_eq!(summary.examples[31].detail, "v39");
    }

    #[test]
    fn timeout_hint_names_each_observed_kind_once() {
        let mut rec = ViolationRecorder::default();
        rec.record(ProtocolViolationKind::StringIdEcho, "a".into());
        rec.record(ProtocolViolationKind::StringIdEcho, "b".into());
        rec.record(ProtocolViolationKind::MalformedFrame, "c".into());
        let hint = rec.summary().timeout_hint().expect("hint present");
        assert!(hint.contains("3 protocol violations"), "got: {hint}");
        assert_eq!(hint.matches("string-id echo").count(), 1);
        assert!(hint.contains("malformed frame"));
    }

    #[test]
    fn kind_counts_survive_example_eviction() {
        let mut rec = ViolationRecorder::default();
        // 40 orphans evict the early examples; the per-kind total must
        // still report all 40 plus the single echo.
        for i in 0..40 {
            rec.record(ProtocolViolationKind::OrphanResponse, format!("v{i}"));
        }
        rec.record(ProtocolViolationKind::StringIdEcho, "echo".into());
        let summary = rec.summary();
        assert_eq!(
            summary.kind_counts,
            vec![
                (ProtocolViolationKind::StringIdEcho, 1),
                (ProtocolViolationKind::OrphanResponse, 40),
            ]
        );
    }

    #[test]
    fn timeout_hint_is_absent_on_clean_connection() {
        let rec = ViolationRecorder::default();
        assert!(rec.summary().timeout_hint().is_none());
    }

    #[test]
    fn frame_snippet_truncates_oversized_frames() {
        let msg = JsonRpcMessage::request(
            JsonRpcId::Number(1),
            "tools/call",
            json!({"blob": "y".repeat(2000)}),
        );
        let snippet = frame_snippet(&msg);
        assert!(snippet.contains("bytes total"), "got: {snippet}");
        assert!(
            snippet.len() < SNIPPET_BYTES + 64,
            "snippet too long: {} bytes",
            snippet.len()
        );
    }

    #[test]
    fn frame_snippet_keeps_small_frames_intact() {
        let msg = JsonRpcMessage::notification("ping", json!({}));
        let snippet = frame_snippet(&msg);
        assert!(snippet.contains("\"method\":\"ping\""), "got: {snippet}");
        assert!(!snippet.contains("bytes total"));
    }
}
