contextgraph-conformance 0.1.2

//! `contextgraph-conformance` — the public Context Graph Protocol conformance suite
//! (`SPEC.md` §11).
//!
//! "Context Graph Protocol conformant" means *green on this suite for your declared capability
//! set* — a checkable claim, which is what makes third-party adoption safe.
//! [`run_conformance`] drives a provider through the protocol and returns a
//! typed [`ConformanceReport`] with a pass/fail verdict per check and an
//! evidence string for each, so a failure says exactly what was wrong.
//!
//! The checks (all against the frozen `contextgraph-types` contracts):
//!
//! - **handshake** — the provider completes the handshake and reports a
//!   non-empty identity + capabilities (SPEC.md §3).
//! - **consent-scope** — the provider's declared egress scopes are well-formed
//!   and consistent with its `data_flow.egress` (`docs/context-reuse.md` §3):
//!   no off-machine scope alongside `egress: false`, and custom scopes
//!   namespaced.
//! - **frame-validity** — queried frames pass `contextgraph-types` validation: score
//!   in `[0, 1]`, a non-empty title, a non-empty `citation_label` (SPEC.md §6 —
//!   "NEVER a bare uuid").
//! - **verify-honesty** — a provider advertising `verify` answers `valid` for
//!   frames it just served and `stale` when their digests are mutated
//!   (`docs/context-reuse.md` §4). Skipped when `verify` is not advertised —
//!   that is the declared fallback, not a failure.
//! - **budget-honesty** — returned frames' summed `token_cost` never exceeds
//!   the query budget, every declared cost is the canonical count, and the
//!   frame count respects `max_frames` (SPEC.md §7 — "never lies about cost").
//! - **as-of-temporal** — a query pinned with `as_of` gets back no frame whose
//!   `valid_from` is after the pin, i.e. no content that was not yet true at
//!   the pinned instant (SPEC.md §6.1). SHOULD-strength and one-sided: a
//!   provider that returns fewer frames, or none, never fails it.
//! - **shutdown-clean** — the provider tears down without error (SPEC.md §3).
//! - **malformed-input-tolerance** — a garbage line is ignored, or errored with
//!   code `bad_request`, never crashing the host (SPEC.md §R1). Staying alive is
//!   the MUST; the structured `bad_request` code is the SHOULD this check now
//!   inspects (#9), so an arbitrary error no longer passes. Wire-level, so it
//!   applies to stdio providers.
//! - **embedding-fingerprint** — a provider declaring an
//!   `embeddings_fingerprint` rejects a query embedding whose length
//!   contradicts its declared dimension with `bad_request` (SPEC.md §E1). A
//!   SHOULD, gated on the provider declaring a fingerprint; wire-level, so like
//!   the malformed probe it applies to stdio providers.
//! - **provenance-fixture-consistency** — every `file` provenance digest the
//!   provider serves matches the bytes on disk it names, re-read and re-hashed
//!   by the host ([`contextgraph_host::verify_file_provenance`], §6.2/§F5). A
//!   grammatically valid digest that hashes *wrong* — a stale or forged claim —
//!   is caught here, where §F5's grammar check cannot see it. Host-local: a link
//!   to files this host cannot read is skipped, not failed.
//!
//! The suite is deliberately adversarial: pointed at a provider that lies
//! about costs, emits an out-of-range score, omits a citation label, or dies
//! mid-query, the matching check fails loudly. The bundled `contextgraph-example-docs`
//! fixture has `--misbehave` flags that trip each one, proving the suite
//! catches a broken provider (task deliverable).
//!
//! The **host** side of the protocol has binding rules too, which the
//! provider-facing checks above cannot exercise. Those live in
//! [`host_conformance`], the dual suite: [`run_host_conformance`] drives the
//! reference [`Host`] against adversarial in-process providers and asserts it
//! upholds them (`SPEC.md` §11.1; issue #14).
//!
//! Both of those suites certify code in *this* repository. A third,
//! [`composition_conformance`], is for code that is not: it takes a
//! [`ComposingHost`] and certifies **someone else's** composition layer — the step
//! above [`Host::query_all`] that turns a fan-out across several providers into
//! the one frame set that reaches a prompt. That step is where a downstream host
//! makes its own calls about a shared budget, and neither suite above can see it:
//! three providers each returning one honest 400-token frame against a
//! 1000-token query are individually conformant and jointly 200 over. Run
//! [`run_composition_conformance`] against your own host;
//! [`ReferenceComposingHost`] is the worked example that passes it.

use contextgraph_host::{
    ConsentRecord, ContextProvider, DigestVerification, DropReason, Host, HostError,
    RawStdioConnection, frame_kind_name, verify_file_provenance,
};
use contextgraph_types::capability::fingerprint_dimensions;
use contextgraph_types::{
    Capabilities, ConsentReceipt, ContextQuery, ContextQueryResult, ErrorCode, FrameId, FrameKind,
    Grantor, ProviderInfo,
};

pub mod composition_conformance;
pub mod host_conformance;
mod report;

pub use composition_conformance::{
    CCHECK_BUDGET_BOUND, CCHECK_DETERMINISM, CCHECK_QUARANTINE, CCHECK_TOTAL_PARTITION,
    ComposingHost, Composition, ExcludedFrame, ReferenceComposingHost, run_composition_conformance,
};
pub use host_conformance::{
    HCHECK_BUDGET_DROP, HCHECK_COMPOSITION_AUDIT, HCHECK_CONSENT_GATE, HCHECK_CONTENT_QUOTING,
    HCHECK_CRASH_ISOLATION, HCHECK_FRAME_LIMIT, HCHECK_PROVENANCE_BYTES, HCHECK_SCOPE_RECEIPT,
    HCHECK_VERSION_REJECT, run_host_conformance,
};
pub use report::{CheckResult, CheckStatus, ConformanceReport};

/// The stable check names, so reports and callers agree on identifiers.
pub const CHECK_HANDSHAKE: &str = "handshake";
pub const CHECK_CONSENT_SCOPE: &str = "consent-scope";
pub const CHECK_FRAME_VALIDITY: &str = "frame-validity";
pub const CHECK_VERIFY_HONESTY: &str = "verify-honesty";
pub const CHECK_BUDGET_HONESTY: &str = "budget-honesty";
pub const CHECK_AS_OF: &str = "as-of-temporal";
pub const CHECK_SHUTDOWN: &str = "shutdown-clean";
pub const CHECK_MALFORMED: &str = "malformed-input-tolerance";
pub const CHECK_EMBEDDING_FINGERPRINT: &str = "embedding-fingerprint";
pub const CHECK_CORRELATION: &str = "correlation";
pub const CHECK_KINDS_FILTER: &str = "kinds-filter";
pub const CHECK_ANCHOR_RELEVANCE: &str = "anchor-relevance";
pub const CHECK_PROVENANCE_FIXTURE_CONSISTENCY: &str = "provenance-fixture-consistency";

/// How to reach the provider under test. `contextgraph-inspect` builds one of these
/// from its CLI arguments; tests build them directly.
pub enum ProviderTarget {
    /// A child-process provider: `program` plus `args`.
    Stdio { program: String, args: Vec<String> },
    /// A remote provider at `url`.
    Http { url: String },
    /// An already-constructed in-process provider (e.g. a built-in).
    InProcess(Box<dyn ContextProvider>),
}

impl ProviderTarget {
    /// A one-line human description of the target, for the report header.
    pub fn describe(&self) -> String {
        match self {
            ProviderTarget::Stdio { program, args } => {
                if args.is_empty() {
                    format!("stdio: {program}")
                } else {
                    format!("stdio: {program} {}", args.join(" "))
                }
            }
            ProviderTarget::Http { url } => format!("http: {url}"),
            ProviderTarget::InProcess(provider) => format!("in-process: {}", provider.id()),
        }
    }
}

/// Run the full conformance suite against a provider, returning a typed
/// report. Never panics: every failure mode becomes a failing check with
/// evidence.
pub async fn run_conformance(target: ProviderTarget) -> ConformanceReport {
    let description = target.describe();

    // Capture stdio spawn info before `target` is consumed — the malformed
    // probe needs a second, independent connection to the same program.
    let stdio_probe = match &target {
        ProviderTarget::Stdio { program, args } => Some((program.clone(), args.clone())),
        _ => None,
    };

    let mut checks = Vec::new();

    match build_host(target).await {
        Ok((host, id, info, caps)) => {
            if info.name.trim().is_empty() || info.version.trim().is_empty() {
                checks.push(CheckResult::fail(
                    CHECK_HANDSHAKE,
                    format!(
                        "provider identity incomplete: name='{}' version='{}'",
                        info.name, info.version
                    ),
                ));
            } else {
                checks.push(CheckResult::pass(
                    CHECK_HANDSHAKE,
                    describe_handshake(&info, &caps),
                ));
            }
            checks.push(check_consent_scopes(&info));
            run_query_and_shutdown_checks(host, &id, &caps, &mut checks).await;
        }
        Err(error) => {
            checks.push(CheckResult::fail(
                CHECK_HANDSHAKE,
                format!("could not establish provider: {error}"),
            ));
            for name in [
                CHECK_FRAME_VALIDITY,
                CHECK_VERIFY_HONESTY,
                CHECK_CONSENT_SCOPE,
                CHECK_BUDGET_HONESTY,
                CHECK_AS_OF,
                CHECK_KINDS_FILTER,
                CHECK_ANCHOR_RELEVANCE,
                CHECK_PROVENANCE_FIXTURE_CONSISTENCY,
                CHECK_SHUTDOWN,
            ] {
                checks.push(CheckResult::skip(name, "handshake failed"));
            }
        }
    }

    match stdio_probe {
        Some((program, args)) => {
            checks.push(malformed_stdio_probe(&program, &args).await);
            checks.push(embedding_fingerprint_stdio_probe(&program, &args).await);
            checks.push(correlation_stdio_probe(&program, &args).await);
        }
        None => {
            checks.push(CheckResult::skip(
                CHECK_MALFORMED,
                "wire-level malformed-input probe applies to stdio providers only",
            ));
            checks.push(CheckResult::skip(
                CHECK_EMBEDDING_FINGERPRINT,
                "wire-level §E1 bad_request probe applies to stdio providers only",
            ));
            checks.push(CheckResult::skip(
                CHECK_CORRELATION,
                "wire-level §H4 id-echo probe applies to stdio providers only",
            ));
        }
    }

    ConformanceReport {
        target: description,
        checks,
    }
}

/// Stand up a one-provider host for the target and read back the provider's
/// negotiated identity + capabilities. Records consent for an egress
/// provider under test — running the suite *is* the consent to its declared
/// flow, so it isn't spuriously gated.
async fn build_host(
    target: ProviderTarget,
) -> Result<(Host, String, ProviderInfo, Capabilities), HostError> {
    let mut host = Host::new();
    let (id, info, caps) = match target {
        ProviderTarget::Stdio { program, args } => {
            let id = "provider-under-test".to_string();
            host.add_stdio(id.clone(), &program, &args).await?;
            capture_identity(&host, &id)?
        }
        ProviderTarget::Http { url } => {
            let id = "provider-under-test".to_string();
            host.add_http(id.clone(), url, None).await?;
            capture_identity(&host, &id)?
        }
        ProviderTarget::InProcess(provider) => {
            let id = provider.id().to_string();
            let info = provider.info().clone();
            let caps = provider.capabilities().clone();
            host.register(provider);
            (id, info, caps)
        }
    };

    // Running the suite *is* consent to the provider's declared flow, so it
    // isn't spuriously gated. Record the legacy boolean consent, and a receipt
    // for every off-machine egress scope the provider declares (the scope gate).
    if info.data_flow.egress {
        host.record_consent(ConsentRecord::new(
            id.clone(),
            info.data_flow.clone(),
            "conformance run under test",
        ));
    }
    for scope in info.data_flow.off_machine_scopes() {
        host.record_receipt(ConsentReceipt::new(
            id.clone(),
            &info,
            scope.clone(),
            Grantor::Policy("conformance-suite".into()),
            "2026-07-21T00:00:00Z",
        ));
    }

    Ok((host, id, info, caps))
}

fn capture_identity(
    host: &Host,
    id: &str,
) -> Result<(String, ProviderInfo, Capabilities), HostError> {
    let provider = host
        .provider(id)
        .ok_or_else(|| HostError::UnknownProvider(id.to_string()))?;
    Ok((
        id.to_string(),
        provider.info().clone(),
        provider.capabilities().clone(),
    ))
}

async fn run_query_and_shutdown_checks(
    host: Host,
    id: &str,
    caps: &Capabilities,
    checks: &mut Vec<CheckResult>,
) {
    let query = sample_query();
    match host.query_provider(id, &query).await {
        Ok(result) => {
            let (ok, evidence) = check_frames(&result);
            checks.push(CheckResult::from_bool(CHECK_FRAME_VALIDITY, ok, evidence));
            checks.push(check_verify_honesty(&host, id, caps, &result).await);

            let (budget_ok, budget_evidence) = check_budget(&result, &query);
            checks.push(CheckResult::from_bool(
                CHECK_BUDGET_HONESTY,
                budget_ok,
                budget_evidence,
            ));
        }
        Err(error) => {
            let evidence = format!("query failed: {error}");
            checks.push(CheckResult::fail(CHECK_FRAME_VALIDITY, evidence.clone()));
            checks.push(CheckResult::fail(CHECK_VERIFY_HONESTY, evidence.clone()));
            checks.push(CheckResult::fail(CHECK_BUDGET_HONESTY, evidence));
        }
    }

    // The temporal probe fires its own `as_of`-pinned query, so it stands on
    // its own regardless of how the unpinned query above fared. The §Q1 probe
    // is independent for the same reason — it narrows `kinds`, which the
    // unfiltered query above deliberately never does.
    checks.push(check_as_of(&host, id).await);
    checks.push(check_kinds_filter(&host, id, caps).await);
    checks.push(check_anchor_relevance(&host, id, caps).await);
    checks.push(check_provenance_fixture_consistency(&host, id).await);

    let results = host.shutdown().await;
    match results.iter().find(|(pid, _)| pid == id) {
        Some((_, Ok(()))) => checks.push(CheckResult::pass(
            CHECK_SHUTDOWN,
            "provider acknowledged shutdown and tore down cleanly",
        )),
        Some((_, Err(error))) => checks.push(CheckResult::fail(
            CHECK_SHUTDOWN,
            format!("shutdown error: {error}"),
        )),
        None => checks.push(CheckResult::fail(
            CHECK_SHUTDOWN,
            "provider vanished before shutdown could be attempted",
        )),
    }
}

/// Suffix appended to a real digest to simulate a mutated source. Derived from
/// the provider's own digest, so it is guaranteed to differ from it while
/// staying vanishingly unlikely to collide with any digest the provider
/// actually serves.
const MUTATED_SUFFIX: &str = "-contextgraph-conformance-mutated";

/// Probe `context/verify` honesty (`docs/context-reuse.md` §4, requirement V1).
///
/// A provider's digest is opaque and provider-declared, so only the provider
/// can say whether an identity still names its current bytes — which means the
/// suite cannot check the *answer*, only that the provider **distinguishes**.
/// So it asks twice about frames the provider just served:
///
/// 1. with the **real** digests it returned — an honest provider says `valid`;
/// 2. with those digests **mutated** — from the provider's side this is
///    indistinguishable from a source that changed underneath the host, and an
///    honest provider says `stale`.
///
/// A provider that rubber-stamps everything `valid` fails the second ask; one
/// that advertises `verify` but can never vouch for anything fails the first.
/// Both are caught without the suite needing to mutate a real source.
///
/// Skipped — not failed — when the provider does not advertise `verify`: that
/// is the declared capability-gated fallback (V3), and the host re-queries
/// instead.
async fn check_verify_honesty(
    host: &Host,
    id: &str,
    caps: &Capabilities,
    result: &ContextQueryResult,
) -> CheckResult {
    if !caps.verify {
        return CheckResult::skip(
            CHECK_VERIFY_HONESTY,
            "provider does not advertise `verify`; a host falls back to re-querying its frames (§4)",
        );
    }

    let held: Vec<FrameId> = result
        .frames
        .iter()
        .filter(|frame| frame.content_digest.is_some())
        .map(|frame| FrameId::new(id, frame.id.clone(), frame.content_digest.clone()))
        .collect();
    if held.is_empty() {
        return CheckResult::skip(
            CHECK_VERIFY_HONESTY,
            "provider served no frame carrying a `content_digest`, so nothing is verifiable (§1 D4)",
        );
    }

    // Ask 1: the real digests. Every frame the provider just served must
    // verify valid — otherwise it cannot vouch for its own output.
    let unchanged = host.verify_frames(&held).await;
    if !unchanged.dropped.is_empty() {
        let detail: Vec<String> = unchanged
            .dropped
            .iter()
            .map(|dropped| format!("{} => {:?}", dropped.frame.frame_id, dropped.reason))
            .collect();
        return CheckResult::fail(
            CHECK_VERIFY_HONESTY,
            format!(
                "provider advertises `verify` but did not answer `valid` for {} of {} frame(s) it had just served with unchanged digests: {}",
                unchanged.dropped.len(),
                held.len(),
                detail.join(", ")
            ),
        );
    }

    // Ask 2: the same frames with mutated digests — what a changed source
    // looks like from the provider's side.
    let mutated: Vec<FrameId> = held
        .iter()
        .map(|frame| {
            FrameId::new(
                id,
                frame.frame_id.clone(),
                frame
                    .content_digest
                    .as_ref()
                    .map(|digest| format!("{digest}{MUTATED_SUFFIX}")),
            )
        })
        .collect();
    let changed = host.verify_frames(&mutated).await;

    if !changed.retained.is_empty() {
        return CheckResult::fail(
            CHECK_VERIFY_HONESTY,
            format!(
                "provider answered `valid` for {} frame(s) whose content digest it never served — a rubber stamp that lets a host cite stale evidence",
                changed.retained.len()
            ),
        );
    }
    let not_stale: Vec<String> = changed
        .dropped
        .iter()
        .filter(|dropped| !matches!(dropped.reason, DropReason::Stale { .. }))
        .map(|dropped| format!("{} => {:?}", dropped.frame.frame_id, dropped.reason))
        .collect();
    if !not_stale.is_empty() {
        return CheckResult::fail(
            CHECK_VERIFY_HONESTY,
            format!(
                "a digest mismatch on a frame the provider still serves MUST verify `stale` (§4 V1); got: {}",
                not_stale.join(", ")
            ),
        );
    }

    CheckResult::pass(
        CHECK_VERIFY_HONESTY,
        format!(
            "provider verified {n} unchanged frame(s) `valid` and all {n} mutated digest(s) `stale`, carrying no frame bodies",
            n = held.len()
        ),
    )
}

/// Wire-level probe: complete the handshake on a fresh connection, inject a
/// malformed line, then send a valid query. A conforming provider either
/// ignores the garbage and answers the query, or errors on it with code
/// `bad_request` — and stays alive either way (SPEC.md §R1). A provider that
/// dies on one bad line fails; so, now, does one that stays alive but reports an
/// error *other* than `bad_request` — the code is read, not merely the fact of
/// an error (#9), so the check can tell a well-formed rejection from an
/// arbitrary failure.
async fn malformed_stdio_probe(program: &str, args: &[String]) -> CheckResult {
    let mut conn = match RawStdioConnection::spawn(program, args).await {
        Ok(conn) => conn,
        Err(error) => {
            return CheckResult::fail(
                CHECK_MALFORMED,
                format!("could not spawn provider: {error}"),
            );
        }
    };
    if let Err(error) = conn.handshake().await {
        return CheckResult::fail(
            CHECK_MALFORMED,
            format!("handshake failed before the probe could run: {error}"),
        );
    }
    if let Err(error) = conn.send_raw_line("this is not valid json {{{\n").await {
        return CheckResult::fail(
            CHECK_MALFORMED,
            format!("provider closed its input on a malformed line: {error}"),
        );
    }
    if let Err(error) = conn
        .send(&contextgraph_host::Envelope::Query {
            id: None,
            query: sample_query(),
        })
        .await
    {
        return CheckResult::fail(
            CHECK_MALFORMED,
            format!("provider died after a malformed line (before a valid query): {error}"),
        );
    }
    match conn.recv().await {
        Ok(contextgraph_host::Envelope::Frames { .. }) => CheckResult::pass(
            CHECK_MALFORMED,
            "provider ignored a malformed line and still answered a valid query",
        ),
        // §R1's SHOULD: staying alive is the MUST, but a *structured*
        // `bad_request` is what lets a host tell "your line was malformed" from
        // an arbitrary failure. Inspecting the code (as the §E1 probe does) is
        // the whole point of #9 — passing on any error would leave the code
        // unread and the distinction unmade.
        Ok(contextgraph_host::Envelope::Error {
            code: Some(ErrorCode::BadRequest),
            message,
            ..
        }) => CheckResult::pass(
            CHECK_MALFORMED,
            format!(
                "provider errored cleanly on malformed input with `bad_request` and stayed alive: {message}"
            ),
        ),
        // Alive, but the error is not the `bad_request` §R1 recommends (a
        // different code, or none at all). The MUST is met; the SHOULD is not,
        // and an unstructured failure is exactly what structured codes exist to
        // replace — so this is flagged.
        Ok(contextgraph_host::Envelope::Error { code, message, .. }) => CheckResult::fail(
            CHECK_MALFORMED,
            format!(
                "provider stayed alive but answered malformed input with `{}` rather than the `bad_request` §R1 recommends: {message}",
                code.map(|c| c.to_string())
                    .unwrap_or_else(|| "no code".to_string())
            ),
        ),
        Ok(other) => CheckResult::fail(
            CHECK_MALFORMED,
            format!(
                "provider replied to a valid query with an unexpected `{}` envelope",
                contextgraph_host::envelope_kind(&other)
            ),
        ),
        Err(HostError::ProviderCrashed { .. }) => CheckResult::fail(
            CHECK_MALFORMED,
            "provider crashed on a malformed line — it must error-or-ignore, not die",
        ),
        Err(error) => CheckResult::fail(
            CHECK_MALFORMED,
            format!("provider mishandled malformed input: {error}"),
        ),
    }
}

/// Wire-level probe for §E1: a provider that declares an
/// `embeddings_fingerprint` **SHOULD** reject a query embedding whose length
/// contradicts that fingerprint's dimension with `bad_request`, rather than
/// scoring a vector from a different space into plausible-looking, meaningless
/// similarity.
///
/// Driven on the raw wire like [`malformed_stdio_probe`], because the honest
/// reply is an `error` envelope carrying a *code* — and the host's query path
/// collapses that to a bare message, losing the `bad_request` §E1 names. Reading
/// the code directly is what makes this checkable, and is why the probe is
/// stdio-only (skipped for in-process/HTTP targets — a documented limitation).
///
/// Gated on the provider declaring a fingerprint: one that declares none has no
/// dimension to contradict and is skipped, exactly as a provider that does not
/// advertise `verify` skips `verify-honesty`.
async fn embedding_fingerprint_stdio_probe(program: &str, args: &[String]) -> CheckResult {
    let mut conn = match RawStdioConnection::spawn(program, args).await {
        Ok(conn) => conn,
        Err(error) => {
            return CheckResult::fail(
                CHECK_EMBEDDING_FINGERPRINT,
                format!("could not spawn provider: {error}"),
            );
        }
    };
    let caps = match conn.handshake().await {
        Ok((_, caps)) => caps,
        Err(error) => {
            return CheckResult::skip(
                CHECK_EMBEDDING_FINGERPRINT,
                format!("handshake failed before the §E1 probe could run: {error}"),
            );
        }
    };
    let Some(fingerprint) = caps.embeddings_fingerprint.clone() else {
        return CheckResult::skip(
            CHECK_EMBEDDING_FINGERPRINT,
            "provider declares no embeddings_fingerprint, so §E1 has no dimension to contradict",
        );
    };
    let Some(dimension) = fingerprint_dimensions(&fingerprint) else {
        return CheckResult::skip(
            CHECK_EMBEDDING_FINGERPRINT,
            format!(
                "fingerprint `{fingerprint}` declares no parseable dimension, so §E1 cannot be probed"
            ),
        );
    };

    // A length guaranteed to differ from the declared dimension — the
    // wrong-space vector §E1 says to reject.
    let wrong_len = if dimension == 1 { 2 } else { 1 };
    let mut query = sample_query();
    query.embedding = Some(vec![0.0; wrong_len]);
    if let Err(error) = conn
        .send(&contextgraph_host::Envelope::Query { id: None, query })
        .await
    {
        return CheckResult::fail(
            CHECK_EMBEDDING_FINGERPRINT,
            format!("provider closed its input before the §E1 probe query: {error}"),
        );
    }
    match conn.recv().await {
        // The recommended reply: it named the request wrong with the code §E1
        // specifies.
        Ok(contextgraph_host::Envelope::Error {
            code: Some(ErrorCode::BadRequest),
            ..
        }) => CheckResult::pass(
            CHECK_EMBEDDING_FINGERPRINT,
            format!(
                "provider declares {fingerprint} ({dimension}-dim) and rejected a {wrong_len}-dim embedding with `bad_request` (§E1)"
            ),
        ),
        // Refused, but not with the code §E1 recommends. Refusing at all is the
        // load-bearing half of a SHOULD, so this passes with a note.
        Ok(contextgraph_host::Envelope::Error { code, message, .. }) => CheckResult::pass(
            CHECK_EMBEDDING_FINGERPRINT,
            format!(
                "provider rejected a {wrong_len}-dim embedding against {fingerprint} with `{}` rather than the `bad_request` §E1 recommends: {message}",
                code.unwrap_or(ErrorCode::Internal)
            ),
        ),
        // The violation: it *scored* a vector from a different space.
        Ok(contextgraph_host::Envelope::Frames { .. }) => CheckResult::fail(
            CHECK_EMBEDDING_FINGERPRINT,
            format!(
                "provider declares {fingerprint} ({dimension}-dim) but scored a {wrong_len}-dim embedding into frames instead of rejecting it — meaningless similarity from a different vector space (§E1)"
            ),
        ),
        Ok(other) => CheckResult::fail(
            CHECK_EMBEDDING_FINGERPRINT,
            format!(
                "provider answered the §E1 probe with an unexpected `{}` envelope",
                contextgraph_host::envelope_kind(&other)
            ),
        ),
        Err(HostError::ProviderCrashed { .. }) => CheckResult::fail(
            CHECK_EMBEDDING_FINGERPRINT,
            "provider crashed on a dimension-mismatched embedding — §E1 asks it to reply `bad_request`, not die",
        ),
        Err(error) => CheckResult::fail(
            CHECK_EMBEDDING_FINGERPRINT,
            format!("provider mishandled the §E1 probe: {error}"),
        ),
    }
}

/// The correlation id the §H4 probe sends. Deliberately distinctive so a
/// provider that echoes *something* — a counter, its own id — fails rather
/// than coincidentally matching.
const CORRELATION_PROBE_ID: &str = "cgp-conformance-h4-7f3a";

/// **§H4** — a provider declaring `capabilities.correlation` **MUST** echo a
/// request's `id` verbatim on the corresponding `frames` or `error`.
///
/// This check exists because the guarantee was previously unenforceable from
/// outside. H4's only witness was the reference provider's
/// `drop-correlation-id` misbehave mode, and that mode "went red" merely
/// because dropping the id desynchronizes the host's demultiplexer and breaks
/// every *other* check downstream. Nothing actually asserted the echo — so an
/// external implementation (each of the three SDKs) could declare
/// `correlation: true`, never echo an id, and pass the suite. Requiring the
/// matching check in `conformance-red.sh` is what surfaced the hole.
///
/// The probe is raw-stdio rather than host-driven for the same reason the §E1
/// probe is: the host layer *interprets* correlation (it demultiplexes on the
/// id and raises `CorrelationMismatch`), so driving through it would test the
/// host's reaction rather than the provider's wire behavior.
async fn correlation_stdio_probe(program: &str, args: &[String]) -> CheckResult {
    let mut conn = match RawStdioConnection::spawn(program, args).await {
        Ok(conn) => conn,
        Err(error) => {
            return CheckResult::fail(
                CHECK_CORRELATION,
                format!("could not spawn provider: {error}"),
            );
        }
    };
    let caps = match conn.handshake().await {
        Ok((_, caps)) => caps,
        Err(error) => {
            return CheckResult::skip(
                CHECK_CORRELATION,
                format!("handshake failed before the §H4 probe could run: {error}"),
            );
        }
    };
    if !caps.correlation {
        // Not a failure: correlation is negotiated, and a lock-step provider
        // that never claims it is conformant. H4 binds only those who declare.
        return CheckResult::skip(
            CHECK_CORRELATION,
            "provider does not declare capabilities.correlation, so §H4 does not bind it",
        );
    }

    let query = sample_query();
    if let Err(error) = conn
        .send(&contextgraph_host::Envelope::Query {
            id: Some(CORRELATION_PROBE_ID.to_string()),
            query,
        })
        .await
    {
        return CheckResult::fail(
            CHECK_CORRELATION,
            format!("provider closed its input before the §H4 probe query: {error}"),
        );
    }

    let reply = match conn.recv().await {
        Ok(reply) => reply,
        Err(error) => {
            return CheckResult::fail(
                CHECK_CORRELATION,
                format!("provider mishandled the §H4 probe: {error}"),
            );
        }
    };

    let kind = contextgraph_host::envelope_kind(&reply);
    // `frames` and `error` both answer a query, and H4 binds both.
    match reply.correlation_id() {
        Some(echoed) if echoed == CORRELATION_PROBE_ID => CheckResult::pass(
            CHECK_CORRELATION,
            format!(
                "provider declares correlation and echoed the request id verbatim on its `{kind}` reply (§H4)"
            ),
        ),
        Some(echoed) => CheckResult::fail(
            CHECK_CORRELATION,
            format!(
                "provider declares correlation but echoed `{echoed}` on its `{kind}` reply instead of the request's `{CORRELATION_PROBE_ID}` — a host demultiplexing on the id would match this reply to the wrong request (§H4)"
            ),
        ),
        None if matches!(reply, contextgraph_host::Envelope::Frames { .. })
            || matches!(reply, contextgraph_host::Envelope::Error { .. }) =>
        {
            CheckResult::fail(
                CHECK_CORRELATION,
                format!(
                    "provider declares correlation but its `{kind}` reply carried no id — the host cannot match it to the request it answers, so the connection is forced back to lock-step (§H4)"
                ),
            )
        }
        None => CheckResult::fail(
            CHECK_CORRELATION,
            format!("provider answered the §H4 probe with an unexpected `{kind}` envelope"),
        ),
    }
}

/// The instant the `as_of` probe pins retrieval to (`SPEC.md` §6.1). Chosen to
/// fall *between* the reference fixture's two frame validity windows, so an
/// honest provider's pinned answer is observably narrower than its unpinned one.
const AS_OF_PIN: &str = "2026-07-01T00:00:00Z";

/// Probe `as_of` temporal pinning (`SPEC.md` §6.1, §F4). `as_of` pins retrieval
/// to an instant; a frame whose `valid_from` is strictly after the pin is
/// content that was not yet true then — exactly what the pin exists to keep out
/// of the answer.
///
/// SHOULD-strength and deliberately one-sided: it never penalizes a provider for
/// returning *fewer* frames (or none) under a pin, because implementing
/// time-travel retrieval is optional. It fails only on a frame the provider
/// *did* return whose `valid_from` provably postdates the pin — a temporal lie
/// no matter how sophisticated the provider's time handling. Comparison is
/// lexicographic on the UTC strings, which is chronological because the
/// timestamp profile admits one spelling per instant (§6.1). A provider serving
/// no timestamped content trivially passes.
async fn check_as_of(host: &Host, id: &str) -> CheckResult {
    match host.query_provider(id, &as_of_query()).await {
        Ok(result) => {
            let not_yet_valid: Vec<String> = result
                .frames
                .iter()
                .filter_map(|frame| {
                    frame
                        .valid_from
                        .as_deref()
                        .filter(|valid_from| *valid_from > AS_OF_PIN)
                        .map(|valid_from| format!("{} (valid_from={valid_from})", frame.id))
                })
                .collect();
            if not_yet_valid.is_empty() {
                CheckResult::pass(
                    CHECK_AS_OF,
                    format!(
                        "as_of={AS_OF_PIN}: none of the {} returned frame(s) is dated after the pin",
                        result.frames.len()
                    ),
                )
            } else {
                CheckResult::fail(
                    CHECK_AS_OF,
                    format!(
                        "provider returned {} frame(s) whose valid_from is after as_of={AS_OF_PIN} — content that was not yet true at the pinned instant (§6.1): {}",
                        not_yet_valid.len(),
                        not_yet_valid.join(", ")
                    ),
                )
            }
        }
        Err(error) => CheckResult::fail(CHECK_AS_OF, format!("as_of query failed: {error}")),
    }
}

/// **§Q1** — a non-empty `kinds` is a filter a provider must honor.
///
/// The probe narrows to a single kind drawn from the provider's *own* declared
/// `capabilities.query.kinds`, so it can never be an unfair request: the
/// provider said it serves this kind. Every returned frame must then be of that
/// kind.
///
/// Worth stating why this check did not exist until now: [`sample_query`] sends
/// `kinds: []`, so every provider was only ever asked the unfiltered question,
/// and a provider that ignored the filter entirely passed the whole suite. All
/// four reference implementations did exactly that.
async fn check_kinds_filter(host: &Host, id: &str, caps: &Capabilities) -> CheckResult {
    let Some(declared) = caps.query.kinds.first() else {
        return CheckResult::skip(
            CHECK_KINDS_FILTER,
            "provider declares no query kinds, so §Q1 has no kind to narrow to",
        );
    };
    let Some(kind) = frame_kind_from_wire(declared) else {
        return CheckResult::skip(
            CHECK_KINDS_FILTER,
            format!(
                "provider declares kind `{declared}`, which is outside the closed FrameKind vocabulary, so §Q1 cannot be probed"
            ),
        );
    };

    let query = ContextQuery {
        kinds: vec![kind],
        ..sample_query()
    };
    match host.query_provider(id, &query).await {
        Ok(result) => {
            let off_kind: Vec<String> = result
                .frames
                .iter()
                .filter(|frame| frame.kind != kind)
                .map(|frame| format!("{} (kind={})", frame.id, frame_kind_name(frame.kind)))
                .collect();
            if off_kind.is_empty() {
                CheckResult::pass(
                    CHECK_KINDS_FILTER,
                    format!(
                        "kinds=[{declared}]: all {} returned frame(s) are of the requested kind (§Q1)",
                        result.frames.len()
                    ),
                )
            } else {
                CheckResult::fail(
                    CHECK_KINDS_FILTER,
                    format!(
                        "provider returned {} frame(s) outside the requested kinds=[{declared}] — content the host explicitly excluded, charged against its budget (§Q1): {}",
                        off_kind.len(),
                        off_kind.join(", ")
                    ),
                )
            }
        }
        Err(error) => CheckResult::fail(
            CHECK_KINDS_FILTER,
            format!("kinds-filtered query failed: {error}"),
        ),
    }
}

/// Parse a declared capability kind string back into the closed [`FrameKind`]
/// vocabulary. `None` for anything outside it — a provider may declare an
/// extension kind, and §Q1 simply has nothing to say about it.
fn frame_kind_from_wire(kind: &str) -> Option<FrameKind> {
    match kind {
        "snippet" => Some(FrameKind::Snippet),
        "symbol" => Some(FrameKind::Symbol),
        "fact" => Some(FrameKind::Fact),
        "doc" => Some(FrameKind::Doc),
        "memory" => Some(FrameKind::Memory),
        "episode" => Some(FrameKind::Episode),
        "graph" => Some(FrameKind::Graph),
        _ => None,
    }
}

/// **§G3/§G4** — a graph-declaring provider must actually do something with
/// `anchors`.
///
/// The graph is what the protocol is *named* for, and it was the least
/// exercised surface in the repo: the reference fixture declared
/// `graph: false` and served frames with `relations: vec![]`, so G1 and G2
/// passed vacuously (no edges to validate) and G3's boost was never witnessed
/// at all.
///
/// The probe first asks an unanchored question to discover a URI the provider
/// actually serves, then re-asks anchored on it. Discovering the anchor from
/// the provider's own output is what keeps this fair: the suite never invents a
/// URI and demands the provider know it.
async fn check_anchor_relevance(host: &Host, id: &str, caps: &Capabilities) -> CheckResult {
    if !caps.graph {
        return CheckResult::skip(
            CHECK_ANCHOR_RELEVANCE,
            "provider does not declare capabilities.graph, so §G3/§G4 do not bind it",
        );
    }

    let baseline = match host.query_provider(id, &sample_query()).await {
        Ok(result) => result,
        Err(error) => {
            return CheckResult::fail(
                CHECK_ANCHOR_RELEVANCE,
                format!("baseline query failed: {error}"),
            );
        }
    };

    // Prefer a one-hop anchor (a relation target): it proves the provider
    // traverses edges, not merely compares its own `uri`.
    let anchor = baseline
        .frames
        .iter()
        .find_map(|frame| frame.relations.first().map(|r| r.target_uri.clone()))
        .or_else(|| baseline.frames.iter().find_map(|frame| frame.uri.clone()));
    let Some(anchor) = anchor else {
        return CheckResult::skip(
            CHECK_ANCHOR_RELEVANCE,
            "provider declares graph but served no frame carrying a uri or a relation target to anchor on",
        );
    };

    let anchored_query = ContextQuery {
        anchors: vec![anchor.clone()],
        ..sample_query()
    };
    match host.query_provider(id, &anchored_query).await {
        Ok(result) => {
            let anchored: Vec<&contextgraph_types::ContextFrame> = result
                .frames
                .iter()
                .filter(|frame| frame_is_anchored(frame, &anchor))
                .collect();
            if anchored.is_empty() {
                return CheckResult::fail(
                    CHECK_ANCHOR_RELEVANCE,
                    format!(
                        "provider declares capabilities.graph but returned no frame anchored on `{anchor}` — a URI drawn from its own previous answer (§G4)"
                    ),
                );
            }
            // G3 is a SHOULD, so ranking is reported rather than enforced: a
            // provider that finds the anchored frame but orders it second is
            // still conformant, and saying so is more honest than inventing a
            // MUST the spec does not state.
            let first_is_anchored = result
                .frames
                .first()
                .is_some_and(|frame| frame_is_anchored(frame, &anchor));
            let ranking = if first_is_anchored {
                "and ranked it first"
            } else {
                "though it did not rank it first (§G3 is a SHOULD)"
            };
            CheckResult::pass(
                CHECK_ANCHOR_RELEVANCE,
                format!(
                    "anchored on `{anchor}`: provider returned {} anchored frame(s) {ranking}",
                    anchored.len()
                ),
            )
        }
        Err(error) => CheckResult::fail(
            CHECK_ANCHOR_RELEVANCE,
            format!("anchored query failed: {error}"),
        ),
    }
}

/// §G4's anchoring predicate: the frame's own `uri` (zero hops) or any labelled
/// edge's `target_uri` (one hop) equals the anchor.
fn frame_is_anchored(frame: &contextgraph_types::ContextFrame, anchor: &str) -> bool {
    frame.uri.as_deref() == Some(anchor) || frame.relations.iter().any(|r| r.target_uri == anchor)
}

/// **§6.2/§F5 (bytes)** — every `file` provenance digest a provider serves must
/// match the bytes on disk it names.
///
/// The `frame-validity` §F5 check proves a provenance digest is *shaped* like a
/// sha256; only re-reading the file it addresses proves it is the *right* one.
/// This check re-reads each `file` provenance the provider serves and re-hashes
/// it with [`contextgraph_host::verify_file_provenance`], the host's own
/// byte-level verifier.
///
/// A definitive failure is a **`Mismatch`**: the bytes are here and hash to
/// something else — provenance forgery, or a fixture that drifted out of sync
/// with its own files. An **`Unreadable`** link (a `file://` this host cannot
/// see — an out-of-tree or remote provider) is *not* a failure: byte
/// verification is a host-local capability, and a provider is not broken because
/// its files do not sit on this machine. A provider serving no locally-readable
/// file provenance is therefore skipped, not failed — mirroring how
/// `verify-honesty` skips a provider that does not advertise `verify`.
async fn check_provenance_fixture_consistency(host: &Host, id: &str) -> CheckResult {
    let result = match host.query_provider(id, &sample_query()).await {
        Ok(result) => result,
        Err(error) => {
            return CheckResult::fail(
                CHECK_PROVENANCE_FIXTURE_CONSISTENCY,
                format!("query failed: {error}"),
            );
        }
    };

    let mut verified = 0usize;
    let mut unreadable = 0usize;
    let mut mismatches = Vec::new();
    for frame in &result.frames {
        for (index, outcome) in verify_file_provenance(frame) {
            match outcome {
                DigestVerification::Verified => verified += 1,
                DigestVerification::Mismatch { expected, actual } => mismatches.push(format!(
                    "{} provenance[{index}] declared {expected} but its bytes hash to {actual}",
                    frame.id
                )),
                DigestVerification::Unreadable { .. } => unreadable += 1,
                DigestVerification::NotFileProvenance => {}
            }
        }
    }

    if !mismatches.is_empty() {
        return CheckResult::fail(
            CHECK_PROVENANCE_FIXTURE_CONSISTENCY,
            format!(
                "{} file-provenance digest(s) do not match the bytes they name — a stale or forged digest that passes §F5's grammar but not its bytes (§6.2): {}",
                mismatches.len(),
                mismatches.join("; ")
            ),
        );
    }
    if verified == 0 {
        return CheckResult::skip(
            CHECK_PROVENANCE_FIXTURE_CONSISTENCY,
            format!(
                "no locally re-readable file provenance to verify ({unreadable} link(s) name files this host cannot see); §6.2 byte-verification is host-local"
            ),
        );
    }
    CheckResult::pass(
        CHECK_PROVENANCE_FIXTURE_CONSISTENCY,
        format!(
            "re-read and re-hashed {verified} file-provenance digest(s) against the bytes on disk — all match (§6.2)"
        ),
    )
}

/// The [`sample_query`] pinned to [`AS_OF_PIN`] — the query the temporal probe
/// fires. Everything else is held equal so only the pin varies.
fn as_of_query() -> ContextQuery {
    ContextQuery {
        as_of: Some(AS_OF_PIN.into()),
        ..sample_query()
    }
}

/// The query the suite probes every provider with — no `kinds` filter, so any
/// provider is asked for its best frames (SPEC.md §5).
pub fn sample_query() -> ContextQuery {
    ContextQuery {
        goal: "conformance probe: return your most relevant frames".into(),
        query_text: Some("conformance probe".into()),
        embedding: None,
        kinds: vec![],
        anchors: vec![],
        max_frames: 8,
        max_tokens: 4096,
        as_of: None,
        representation_preferences: vec![],
    }
}

/// Validate a query result's frames against the `ContextFrame` contract
/// (SPEC.md §6). Returns `(passed, evidence)`. Zero frames is permitted — a
/// provider may simply have nothing relevant.
pub fn check_frames(result: &ContextQueryResult) -> (bool, String) {
    if result.frames.is_empty() {
        return (
            true,
            "provider returned 0 frames (permitted — nothing relevant to the probe)".into(),
        );
    }

    let mut problems = Vec::new();
    for (i, frame) in result.frames.iter().enumerate() {
        if !frame.has_valid_score() {
            problems.push(format!("frame[{i}] score {} is outside [0,1]", frame.score));
        }
        if frame.title.trim().is_empty() {
            problems.push(format!("frame[{i}] has an empty title"));
        }
        match &frame.citation_label {
            Some(label) if !label.trim().is_empty() => {}
            _ => problems.push(format!(
                "frame[{i}] is missing a citation_label (§F3 — never a bare id)"
            )),
        }
        // §P1–P3: a frame must not lie about how it carries its content — a
        // `reference` carrying inline content, a `compact` missing its
        // canonical hash. `representation_invariants` names the exact breach.
        // The predicate shipped in PR #42 with no caller; this is the caller.
        if let Err(violation) = frame.representation_invariants() {
            problems.push(format!("frame[{i}] {violation} (§P1–P3)"));
        }
        // §F4: temporal fields must be in the protocol's timestamp profile.
        // Naming the offending field is what makes this actionable — before
        // this check, `"valid_from": "last tuesday"` was fully conformant and
        // the bi-temporal guarantee was unfalsifiable.
        for field in frame.invalid_temporal_fields() {
            problems.push(format!(
                "frame[{i}] field `{field}` is not an RFC 3339 UTC timestamp (§F4)"
            ));
        }
        // §D1: the frame's own content_digest, when present, must be in the
        // protocol's digest form. Like §G2 this was listed as verified here and
        // read by nothing — so the digest that anchors deterministic
        // composition, usage reports and `context/verify` was held to a looser
        // standard than the §F5 provenance digests immediately below it.
        if !frame.has_usable_content_digest() {
            problems.push(format!(
                "frame[{i}] content_digest is present but not `sha256:<64 lowercase hex>` (§D1)"
            ));
        }
        // §F5: file provenance must carry a well-formed digest, since that is
        // the only provenance a host can independently re-read and verify.
        for index in frame.provenance_with_unusable_digests() {
            problems.push(format!(
                "frame[{i}] provenance[{index}] addresses a file but its digest is missing or not `sha256:<64 lowercase hex>` (§F5)"
            ));
        }
        // §G1/§G2: a graph edge must be citable by a human label, and must
        // actually point somewhere. G2 was listed as "Verified by
        // frame-validity" while no code read `target_uri` at all — the exact
        // self-attestation §11.1 rejects. It is verified here now.
        for (edge_index, edge) in frame.relations.iter().enumerate() {
            if !edge.has_display_name() {
                problems.push(format!(
                    "frame[{i}] relation[{edge_index}] `{}` has no display_name (§G1 — an edge is surfaced by label, never a raw id)",
                    edge.rel
                ));
            }
            if !edge.has_target_uri() {
                problems.push(format!(
                    "frame[{i}] relation[{edge_index}] `{}` has an empty target_uri (§G2 — an edge to nowhere is not an edge)",
                    edge.rel
                ));
            }
        }
    }

    if problems.is_empty() {
        (
            true,
            format!(
                "{} frame(s) — scores in [0,1], titles, citation labels, honest representations, RFC 3339 timestamps, well-formed digests, labelled and targeted relations",
                result.frames.len()
            ),
        )
    } else {
        (false, problems.join("; "))
    }
}

/// Validate a query result against the budget contract (`SPEC.md` §B1, §B3,
/// §B4). Returns `(passed, evidence)`.
///
/// Three distinct promises, deliberately checked separately so a failure says
/// which one broke:
///
/// - **§B1** the declared costs sum within `max_tokens`;
/// - **§B3** each declared cost equals the canonical count for its content —
///   this is what turned the check from arithmetic into truth;
/// - **§B4** the frame count respects `max_frames`.
pub fn check_budget(result: &ContextQueryResult, query: &ContextQuery) -> (bool, String) {
    let mut problems = Vec::new();

    let declared = result.total_token_cost();
    if declared > query.max_tokens as u64 {
        problems.push(format!(
            "declared cost {declared} exceeds the query budget of {} (§B1)",
            query.max_tokens
        ));
    }

    let dishonest = result.frames_with_dishonest_cost();
    if !dishonest.is_empty() {
        let canonical = result.canonical_token_cost();
        problems.push(format!(
            "{} frame(s) misdeclare token_cost — {} (§B3); declared total {declared}, canonical total {canonical}",
            dishonest.len(),
            dishonest.join(", ")
        ));
    }

    if !result.respects_frame_limit(query.max_frames) {
        problems.push(format!(
            "returned {} frames against max_frames={} (§B4)",
            result.frames.len(),
            query.max_frames
        ));
    }

    if problems.is_empty() {
        (
            true,
            format!(
                "{} frame(s), {declared} tokens within the {} budget; every declared cost matches its canonical count",
                result.frames.len(),
                query.max_tokens
            ),
        )
    } else {
        (false, problems.join("; "))
    }
}

/// Validate a provider's declared egress scopes against its `data_flow`
/// (`docs/context-reuse.md` §3, requirement C5): every scope must be well-formed
/// (custom scopes namespaced), and no off-machine scope may be declared with
/// `egress: false`. A scope-lying provider — one claiming local posture while
/// naming a destination that leaves — fails here.
fn check_consent_scopes(info: &ProviderInfo) -> CheckResult {
    if info.data_flow.scopes_consistent() {
        let scopes: Vec<&str> = info
            .data_flow
            .egress_scopes
            .iter()
            .map(|scope| scope.as_str())
            .collect();
        CheckResult::pass(
            CHECK_CONSENT_SCOPE,
            format!(
                "declared egress scopes {scopes:?} are well-formed and consistent with egress={}",
                info.data_flow.egress
            ),
        )
    } else {
        CheckResult::fail(
            CHECK_CONSENT_SCOPE,
            format!(
                "egress scopes {:?} are inconsistent with egress={}: an off-machine scope alongside egress=false, or a non-namespaced custom scope (§3, C5)",
                info.data_flow
                    .egress_scopes
                    .iter()
                    .map(|scope| scope.as_str())
                    .collect::<Vec<_>>(),
                info.data_flow.egress
            ),
        )
    }
}

fn describe_handshake(info: &ProviderInfo, caps: &Capabilities) -> String {
    format!(
        "provider '{}' v{} — data-flow reads={} writes={} egress={}; query kinds={:?}, graph={}",
        info.name,
        info.version,
        info.data_flow.reads,
        info.data_flow.writes,
        info.data_flow.egress,
        caps.query.kinds,
        caps.graph,
    )
}