contextgraph-conformance 2.0.0

//! `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.
//! - **attestation** — every detached [provenance
//!   attestation](contextgraph_types::ProvenanceAttestation) a provider serves
//!   is recomputed from the frame in hand and verified against a key the
//!   handshake published (`SPEC.md` §6.5, F6–F9). Where
//!   `provenance-fixture-consistency` proves a digest matches its bytes, this
//!   proves somebody *signed* for those bytes — the digest and the frame come
//!   from the same unauthenticated party, so §6.2 is satisfied in full by a
//!   provider that fabricated both. A provider that publishes no attester key
//!   and serves no attestation passes: §6.5 makes the construction mandatory
//!   and the signing optional.
//!
//! 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::{
    AttesterKey, ConsentRecord, ContextProvider, DigestVerification, DropReason, Host, HostError,
    RawStdioConnection, frame_kind_name, verify_file_provenance,
};
use contextgraph_types::capability::fingerprint_dimensions;
use contextgraph_types::{
    AttestationVerdict, Capabilities, ConsentReceipt, ContextQuery, ContextQueryResult, ErrorCode,
    FrameId, FrameKind, Grantor, ProviderInfo, verify_frame_attestation, verify_frame_inclusion,
};

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";
pub const CHECK_ATTESTATION: &str = "attestation";

/// 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);
            checks.push(attestation_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",
            ));
            checks.push(CheckResult::skip(
                CHECK_ATTESTATION,
                "wire-level §6.5 attestation 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"),
        ),
    }
}

/// **§6.5 (F6–F9)** — a provenance attestation is checked, never taken on
/// trust.
///
/// A digest proves the bytes have not moved since somebody wrote that number
/// down. It says nothing about *who* wrote it, and the digest and the frame come
/// from the same unauthenticated party — so §6.2 is satisfied in full by a
/// provider that fabricated both. A signature is the only thing that closes
/// that, and until this check existed the four guarantees over it were verified
/// by `contextgraph_types::attest`'s own unit tests and by nothing on the wire.
/// Every other guarantee in this protocol earns its credibility from a suite
/// with an adversarial mode behind it; a guarantee whose only witness is the
/// implementation asserting about itself is the self-attestation §11.1 rules
/// out.
///
/// The probe:
///
/// 1. reads the [attester keys](contextgraph_host::AttesterKey) the handshake
///    published — a *construction* anchor, not a trust one (see that type);
/// 2. queries, and takes the detached attestations off the *result* — their
///    one wire home (`SPEC.md` §6.5.5, #161);
/// 3. recomputes each named frame's commitment from the frame in hand and
///    verifies the signature over it, exactly as §6.5.4 orders the two steps;
/// 4. when anything fails to verify, re-asks the same provider **through the
///    reference [`Host`]** and asserts the affected frame still arrives —
///    F9's rule that an unverifiable attestation degrades a frame to
///    *unattested* rather than removing it.
///
/// **`provider_id` in the commitment is the handshake-declared `provider.name`.**
/// §6.5.2 binds a commitment to a provider id without saying which one, and the
/// host-chosen local id (`provider-under-test` here) is not a string the
/// provider ever sees — so it is not one a provider could sign against. The
/// declared name is the only identifier both ends observe. Pinning that in
/// SPEC.md is tracked separately; until it is, an implementation reading only
/// the spec could pick differently and fail this check for the wrong reason.
///
/// Raw-stdio like the §R1, §E1 and §H4 probes, but for a different reason: the
/// attester keys are read off the handshake, which [`Host`] performs and does
/// not surface.
///
/// Passing outcomes, in order of how a provider reaches them:
///
/// - publishes no key and serves no attestation ⇒ **pass**. §6.5 makes the
///   construction mandatory and the signing optional, so a provider that signs
///   nothing is conformant and this check has nothing to say about it.
/// - every attestation it serves verifies ⇒ **pass**.
/// - every attestation names a scheme this build cannot check ⇒ **skip**, per
///   F8: "I cannot check this" is never "this is good", and it is equally never
///   "this is forged".
async fn attestation_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_ATTESTATION,
                format!("could not spawn provider: {error}"),
            );
        }
    };
    let info = match conn.handshake().await {
        Ok((info, _)) => info,
        Err(error) => {
            return CheckResult::skip(
                CHECK_ATTESTATION,
                format!("handshake failed before the §6.5 probe could run: {error}"),
            );
        }
    };
    let keys: Vec<AttesterKey> = conn.attester_keys().to_vec();

    if let Err(error) = conn
        .send(&contextgraph_host::Envelope::Query {
            id: None,
            query: sample_query(),
        })
        .await
    {
        return CheckResult::fail(
            CHECK_ATTESTATION,
            format!("provider closed its input before the §6.5 probe query: {error}"),
        );
    }
    let (frames, attestations, result_attestation) = match conn.recv().await {
        // §6.5.5: the evidence rides the *result*, and nowhere else (#161).
        Ok(contextgraph_host::Envelope::Frames { result, .. }) => (
            result.frames,
            result.frame_attestations,
            result.result_attestation,
        ),
        Ok(other) => {
            return CheckResult::fail(
                CHECK_ATTESTATION,
                format!(
                    "provider answered the §6.5 probe with an unexpected `{}` envelope",
                    contextgraph_host::envelope_kind(&other)
                ),
            );
        }
        Err(error) => {
            return CheckResult::fail(
                CHECK_ATTESTATION,
                format!("provider mishandled the §6.5 probe: {error}"),
            );
        }
    };

    if keys.is_empty() && attestations.is_empty() {
        return CheckResult::pass(
            CHECK_ATTESTATION,
            "provider publishes no attester key and serves no attestation; §6.5 makes the construction mandatory and the signing optional, so there is nothing here to forge",
        );
    }
    if keys.is_empty() {
        return CheckResult::fail(
            CHECK_ATTESTATION,
            format!(
                "provider served {} attestation(s) but published no attester key at the handshake — nothing reading this wire can check them, and an unverifiable signature is decoration (§6.5.4)",
                attestations.len()
            ),
        );
    }
    if frames.is_empty() {
        return CheckResult::pass(
            CHECK_ATTESTATION,
            "provider returned 0 frames, so there is nothing to attest (permitted — nothing relevant to the probe)",
        );
    }
    if attestations.is_empty() {
        return CheckResult::fail(
            CHECK_ATTESTATION,
            format!(
                "provider published {} attester key(s) at the handshake but attested none of the {} frame(s) it served — a signing capability nothing can exercise (§6.5)",
                keys.len(),
                frames.len()
            ),
        );
    }

    let mut verified: Vec<String> = Vec::new();
    let mut uncheckable: Vec<String> = Vec::new();
    let mut problems: Vec<String> = Vec::new();
    // Frames whose attestation did not verify: F9 says these must still be
    // served, degraded to unattested rather than dropped.
    let mut degraded: Vec<String> = Vec::new();

    for entry in &attestations {
        // §6.5.5 names the frame by its whole `(provider_id, frame_id,
        // content_digest)` identity, never by a bare id. Matching on the triple
        // is what stops a provider lending one frame's signature to another
        // that happens to reuse its id.
        let named = &entry.frame.frame_id;
        let Some(frame) = frames
            .iter()
            .find(|frame| frame.identity(&info.name) == entry.frame)
        else {
            problems.push(format!(
                "attestation names frame `{named}`, whose identity is not in the answer it rides with (§6.5.2 binds a signature to one frame of one answer, by provider id, frame id and content digest)"
            ));
            continue;
        };

        // An entry carries a per-frame signature, or membership of the signed
        // result-set root, or both (§6.5.5, F13). The second is the cheapest
        // honest shape — one signature for n frames — and a probe that only
        // understood the first would report every provider that chose it as
        // having signed nothing.
        let (signature, proof) = match (&entry.attestation, &entry.inclusion_proof) {
            (Some(attestation), _) => (attestation, None),
            (None, Some(proof)) => match result_attestation.as_ref() {
                Some(root) => (root, Some(proof)),
                None => {
                    problems.push(format!(
                        "frame `{named}` is attested only by an inclusion proof, but the answer carries no `result_attestation` for that proof to establish membership of (§6.5.3)"
                    ));
                    degraded.push(named.clone());
                    continue;
                }
            },
            (None, None) => {
                problems.push(format!(
                    "the attestation entry for frame `{named}` carries neither a signature nor an inclusion proof, so it names a frame and asserts nothing about it (§6.5.5)"
                ));
                continue;
            }
        };

        let Some(key) = keys.iter().find(|key| key.key_id == signature.key_id) else {
            problems.push(format!(
                "frame `{named}` is signed under key_id `{}`, which the handshake never published",
                signature.key_id
            ));
            continue;
        };
        let Some(key_bytes) = decode_hex(&key.public_key) else {
            problems.push(format!(
                "published key `{}` is not lowercase hex, so no verifier can load it",
                key.key_id
            ));
            continue;
        };

        let verdict = match proof {
            Some(proof) => verify_frame_inclusion(&info.name, frame, proof, signature, &key_bytes),
            None => verify_frame_attestation(&info.name, frame, signature, &key_bytes),
        };

        match verdict {
            AttestationVerdict::Valid => verified.push(named.clone()),
            // The signature checks out over a preimage that binds identity and
            // provenance but says nothing about the frame's bytes, because the
            // frame declared no `content_digest` (#128). `SPEC.md` §6.5.2
            // requires a provider that signs a frame to populate one, so this
            // is a conformance failure of the *attester*, not an unverifiable
            // attestation.
            //
            // It is a problem rather than a degradation for that reason. The
            // frame is servable and the signature is real; what is wrong is the
            // claim a reader would take from it — a provider offering this is
            // publishing a signature that outlives the content it appears to
            // cover, and can re-serve different bytes under the same id
            // tomorrow without disturbing it. Saying that here is the only
            // place a provider author finds out before a consumer does.
            AttestationVerdict::ValidIdentityOnly => {
                problems.push(format!(
                    "frame `{named}` is signed but declares no `content_digest`, so the \
                     signature binds its identity and provenance and nothing about its \
                     content — the same id can be re-served with different bytes and \
                     this signature still verifies. SPEC.md §6.5.2 requires an attester \
                     to populate `content_digest` on any frame it signs"
                ));
                degraded.push(named.clone());
            }
            AttestationVerdict::UnknownAlgorithm(algorithm) => {
                uncheckable.push(format!("{named} (algorithm `{algorithm}`)"));
                degraded.push(named.clone());
            }
            verdict => {
                problems.push(describe_attestation_failure(named, &verdict));
                degraded.push(named.clone());
            }
        }
    }

    // F9. An unverifiable attestation degrades its frame to *unattested*; it
    // never removes the frame from the answer, because a host that dropped such
    // frames would hand any peer a denial-of-service primitive — attach garbage,
    // watch the evidence disappear. Asked of the reference host rather than of
    // this probe's own bookkeeping, so it is a claim about a host's behaviour
    // and not about the suite's.
    if !degraded.is_empty()
        && let Some(dropped) = frames_the_host_dropped(program, args, &degraded).await
    {
        problems.push(format!(
            "the host stopped serving {} after their attestation failed to verify — F9 requires an unverifiable attestation to degrade a frame to unattested, never to remove it",
            dropped.join(", ")
        ));
    }

    if !problems.is_empty() {
        return CheckResult::fail(
            CHECK_ATTESTATION,
            format!(
                "{} of {} attestation(s) did not verify (§6.5.4): {}",
                problems.len(),
                attestations.len(),
                problems.join("; ")
            ),
        );
    }
    if verified.is_empty() {
        return CheckResult::skip(
            CHECK_ATTESTATION,
            format!(
                "every attestation names a scheme this build cannot check, so F8 declines rather than guessing: {}",
                uncheckable.join(", ")
            ),
        );
    }

    let note = if uncheckable.is_empty() {
        String::new()
    } else {
        format!(
            "; {} left unattested by F8 as uncheckable here: {}",
            uncheckable.len(),
            uncheckable.join(", ")
        )
    };
    CheckResult::pass(
        CHECK_ATTESTATION,
        format!(
            "recomputed and verified {} detached attestation(s) over {} served frame(s) against the handshake-published key(s) (§6.5){note}",
            verified.len(),
            frames.len()
        ),
    )
}

/// Name an [`AttestationVerdict`] failure in the terms §6.5.4 separates them
/// into, because the two loudest ones call for opposite responses: a mismatch
/// says the frame moved after signing (tampering), a bad signature says the key
/// is wrong or the signature forged (key management).
fn describe_attestation_failure(frame_id: &str, verdict: &AttestationVerdict) -> String {
    match verdict {
        AttestationVerdict::CommitmentMismatch { expected, signed } => format!(
            "frame `{frame_id}` recomputes to {expected} but its attestation signs {signed} — the frame, its `content_digest`, or its provenance chain changed after signing (CommitmentMismatch)"
        ),
        AttestationVerdict::BadSignature => format!(
            "frame `{frame_id}` commits correctly but its signature does not verify under the published key — forged, or signed by a key the provider did not declare (BadSignature)"
        ),
        AttestationVerdict::MalformedKey => {
            format!("frame `{frame_id}`: the published key is not a well-formed key (MalformedKey)")
        }
        AttestationVerdict::MalformedSignature => format!(
            "frame `{frame_id}`: the signature field is not well-formed for its algorithm (MalformedSignature)"
        ),
        AttestationVerdict::MalformedCommitment => format!(
            "frame `{frame_id}`: `signed_commitment` is not the `sha256:<64 lowercase hex>` §F7 requires (MalformedCommitment)"
        ),
        // Handled by the caller, which reports it as uncheckable rather than
        // as a failure — F8's whole point.
        other => format!("frame `{frame_id}`: {other:?}"),
    }
}

/// Of `expected` frame ids, those the reference [`Host`] does **not** deliver —
/// F9's question, asked of a host rather than of this probe.
///
/// `None` when the question could not be put (the host could not be stood up,
/// or the query failed): an unanswerable question is not evidence of a
/// violation, and reporting it as one would fail a provider for the suite's own
/// trouble.
async fn frames_the_host_dropped(
    program: &str,
    args: &[String],
    expected: &[String],
) -> Option<Vec<String>> {
    let mut host = Host::new();
    let id = "f9-probe".to_string();
    host.add_stdio(id.clone(), program, args).await.ok()?;
    let result = host.query_provider(&id, &sample_query()).await.ok()?;
    let served: Vec<&str> = result
        .frames
        .iter()
        .map(|frame| frame.id.as_str())
        .collect();
    let _ = host.shutdown().await;
    let missing: Vec<String> = expected
        .iter()
        .filter(|id| !served.contains(&id.as_str()))
        .cloned()
        .collect();
    (!missing.is_empty()).then_some(missing)
}

/// Decode lowercase hex into bytes. `None` on an odd length or a non-hex digit
/// — a published key that cannot be decoded is a provider defect, not a
/// verification failure, and the two are reported differently.
fn decode_hex(hex: &str) -> Option<Vec<u8>> {
    if !hex.len().is_multiple_of(2) {
        return None;
    }
    // The even-length check above leaves no remainder, so `.0` drops nothing.
    // `as_chunks` yields `&[u8; 2]`, which indexes without a bounds check.
    hex.as_bytes()
        .as_chunks::<2>()
        .0
        .iter()
        .map(|pair| {
            let hi = (pair[0] as char).to_digit(16)?;
            let lo = (pair[1] as char).to_digit(16)?;
            Some((hi * 16 + lo) as u8)
        })
        .collect()
}

/// 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 base FrameKind vocabulary, so §Q1 cannot be probed"
            ),
        );
    };

    let query = ContextQuery {
        kinds: vec![kind.clone()],
        ..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 **base** [`FrameKind`]
/// vocabulary. `None` for anything outside it — a provider may declare an
/// extension kind, and §Q1 simply has nothing to say about it.
///
/// [`FrameKind::from_wire`] never fails: since the kind vocabulary was opened
/// for forward compatibility, an unrecognized string parses to
/// [`FrameKind::Unknown`] rather than erroring. This check wants the narrower
/// question — "is this one of the seven kinds §Q1 is written about?" — so it
/// parses and then filters on [`FrameKind::is_known`]. Restating the seven
/// names here would put the vocabulary in a second place and let the two drift.
fn frame_kind_from_wire(kind: &str) -> Option<FrameKind> {
    let parsed = FrameKind::from_wire(kind);
    parsed.is_known().then_some(parsed)
}

/// **§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,
    )
}