contextgraph_host/

host.rs

//! The [`Host`] — one uniform handle over every provider, and the fan-out
//! router.
//!
//! The host does the four jobs providers never do: routes a query to
//! capability-matching providers (`SPEC.md` §5), gates consent so nothing
//! reaches an unconsented egress provider (`SPEC.md` §4, C1–C2), enforces
//! per-provider timeouts, and audits budget honesty on two axes — a provider
//! whose frames sum above the query budget lied about `token_cost` (`SPEC.md`
//! §7, B2), and one that returns more frames than `max_frames` overspent a
//! budget the token count never captures (`SPEC.md` §7, B4). Either way its
//! frames are dropped with a loud named report rather than silently trusted.
//! Per-provider isolation is total: one provider erroring, timing out, being
//! dropped for a budget lie, or crashing mid-query never poisons the others
//! (task deliverable 5).

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

use contextgraph_types::{
    ConsentReceipt, ContextFrame, ContextQuery, ContextQueryResult, DataFlow, EgressScope, FrameId,
    ProviderUsage, ServedFrame, UsageReport, Verdict, VerifyRequest,
};

use crate::consent::{ConsentDecision, ConsentRecord, ConsentStore};
use crate::error::HostError;
use crate::provider::{ContextProvider, capability_matches};
use crate::stdio::StdioProvider;
use crate::trust::{AttestationLedger, FrameAttestationOutcome, TrustStore, TrustedKey};

/// Default per-provider query budget — a slow or hung provider is cut off at
/// this and reported as [`HostError::Timeout`], never allowed to stall the
/// fan-out.
const DEFAULT_PROVIDER_TIMEOUT: Duration = Duration::from_secs(30);

/// Registers in-process, stdio, and HTTP providers behind one handle and
/// fans queries out across them.
pub struct Host {
    providers: Vec<Box<dyn ContextProvider>>,
    consent: ConsentStore,
    trust: TrustStore,
    per_provider_timeout: Duration,
}

impl Default for Host {
    fn default() -> Self {
        Self::new()
    }
}

impl Host {
    /// A host with no providers and the default per-provider timeout.
    pub fn new() -> Self {
        Self {
            providers: Vec::new(),
            consent: ConsentStore::new(),
            trust: TrustStore::new(),
            per_provider_timeout: DEFAULT_PROVIDER_TIMEOUT,
        }
    }

    /// A host with a custom per-provider timeout.
    pub fn with_timeout(per_provider_timeout: Duration) -> Self {
        Self {
            per_provider_timeout,
            ..Self::new()
        }
    }

    /// Register an in-process provider (a built-in, e.g. the code graph).
    pub fn register(&mut self, provider: Box<dyn ContextProvider>) {
        self.providers.push(provider);
    }

    /// Spawn and register a child-process provider over stdio, completing the
    /// handshake (`SPEC.md` §3).
    pub async fn add_stdio(
        &mut self,
        id: impl Into<String>,
        program: &str,
        args: &[String],
    ) -> Result<(), HostError> {
        let provider = StdioProvider::spawn(id, program, args).await?;
        self.providers.push(Box::new(provider));
        Ok(())
    }

    /// Connect and register a remote HTTP provider, completing the handshake.
    ///
    /// `credential` is an optional bearer [`Credential`](crate::http::Credential)
    /// attached to every request; pass `None` for an unauthenticated provider.
    /// A plaintext (`http://`) transport to a non-loopback provider is refused
    /// before any bytes leave the host ([`HostError::InsecureTransport`], C7),
    /// and the credential is never logged (C8).
    pub async fn add_http(
        &mut self,
        id: impl Into<String>,
        url: impl Into<String>,
        credential: Option<crate::http::Credential>,
    ) -> Result<(), HostError> {
        let provider = crate::http::HttpProvider::connect_with_auth(id, url, credential).await?;
        self.providers.push(Box::new(provider));
        Ok(())
    }

    /// Record legacy boolean consent for a provider, unlocking an egress
    /// provider that declares no scopes for querying (§3.5).
    pub fn record_consent(&mut self, record: ConsentRecord) {
        self.consent.record(record);
    }

    /// Append a scope-level [`ConsentReceipt`] to the audit ledger, authorizing
    /// one egress scope for one provider (`docs/context-reuse.md` §3). A
    /// provider that declares off-machine egress scopes stays gated until every
    /// such scope has a receipt.
    pub fn record_receipt(&mut self, receipt: ConsentReceipt) {
        self.consent.record_receipt(receipt);
    }

    /// The consent store (read-only), e.g. to persist decisions.
    pub fn consent(&self) -> &ConsentStore {
        &self.consent
    }

    /// Trust `key` for `provider_id`'s provenance attestations
    /// ([ADR 0016](https://github.com/macanderson/context-graph-protocol/blob/main/docs/adr/0016-attestation-trust-roots.md)).
    ///
    /// The operator is the trust root: there is no discovery and no
    /// trust-on-first-use, so a key is here because a person put it here, from
    /// the same material as the provider's own configuration. A host with a UI
    /// shows [`TrustedKey::fingerprint`] beside the consent prompt, so "I
    /// consent to this provider" and "I trust this key" are one decision.
    ///
    /// A host that calls this for nobody verifies nothing and loses nothing:
    /// every signed frame reads as
    /// [`NoTrustedKey`](crate::AttestationState::NoTrustedKey) and is still
    /// served (`SPEC.md` F9).
    pub fn trust_key(&mut self, provider_id: impl Into<String>, key: TrustedKey) {
        self.trust.trust(provider_id, key);
    }

    /// The trust store (read-only), e.g. to persist it beside the consent
    /// ledger or to render what an operator has trusted.
    pub fn trust(&self) -> &TrustStore {
        &self.trust
    }

    /// Replace the whole trust store — for a host restoring one it persisted.
    pub fn set_trust_store(&mut self, trust: TrustStore) {
        self.trust = trust;
    }

    /// The ids of every registered provider, in registration order.
    pub fn provider_ids(&self) -> Vec<&str> {
        self.providers.iter().map(|p| p.id()).collect()
    }

    /// Borrow a registered provider by id, e.g. to read its cached
    /// capabilities.
    pub fn provider(&self, id: &str) -> Option<&dyn ContextProvider> {
        self.providers
            .iter()
            .find(|p| p.id() == id)
            .map(|p| p.as_ref())
    }

    /// How many providers are registered.
    pub fn len(&self) -> usize {
        self.providers.len()
    }

    pub fn is_empty(&self) -> bool {
        self.providers.is_empty()
    }

    /// Revalidate frames this host already holds, so unchanged context can be
    /// reused without re-querying it (`docs/context-reuse.md` §4).
    ///
    /// Identities are grouped by provider and each capable provider is asked
    /// once. The rule is **default-deny**: a frame is retained only on an
    /// explicit [`Verdict::Valid`], and every other outcome — a negative
    /// verdict, a missing digest, a provider that doesn't support verify, an
    /// unregistered provider, a failed request — drops the frame with a reason
    /// (requirement V2). Reasons that
    /// [warrant a re-query](DropReason::warrants_requery) tell the host which
    /// dropped frames are worth fetching again.
    ///
    /// This method holds **no state**: it neither caches frames nor tracks turn
    /// boundaries. When to re-verify is the host's policy (§4 gives informative
    /// guidance); the protocol's job is only to answer the question when asked.
    /// No frame body travels in either direction.
    pub async fn verify_frames(&self, held: &[FrameId]) -> VerifyOutcome {
        use futures_util::future::join_all;

        // Group by provider, preserving first-seen provider order so the
        // outcome is deterministic for a given input.
        let mut order: Vec<&str> = Vec::new();
        let mut grouped: HashMap<&str, Vec<FrameId>> = HashMap::new();
        for frame in held {
            let id = frame.provider_id.as_str();
            if !grouped.contains_key(id) {
                order.push(id);
            }
            grouped.entry(id).or_default().push(frame.clone());
        }

        let legs = order.into_iter().map(|provider_id| {
            let frames = grouped.remove(provider_id).unwrap_or_default();
            self.verify_one_provider(provider_id, frames)
        });

        let mut outcome = VerifyOutcome::default();
        for leg in join_all(legs).await {
            outcome.retained.extend(leg.retained);
            outcome.dropped.extend(leg.dropped);
        }
        outcome
    }

    /// Verify one provider's slice of the held set, converting every failure
    /// mode into dropped frames rather than a propagated error — one provider's
    /// verify failure never affects another's.
    async fn verify_one_provider(&self, provider_id: &str, frames: Vec<FrameId>) -> VerifyOutcome {
        let mut outcome = VerifyOutcome::default();

        let Some(provider) = self.provider(provider_id) else {
            outcome.drop_all(frames, DropReason::UnknownProvider);
            return outcome;
        };
        if !provider.capabilities().verify {
            // The declared fallback: a provider that can't verify gets its
            // frames re-queried rather than trusted (§4, requirement V3).
            outcome.drop_all(frames, DropReason::VerifyUnsupported);
            return outcome;
        }

        // A frame with no digest can't be revalidated — §1's D4 makes that a
        // re-query, not a reuse. Filter before asking, so the request only
        // carries answerable identities.
        let (verifiable, undigested): (Vec<FrameId>, Vec<FrameId>) =
            frames.into_iter().partition(FrameId::is_verifiable);
        outcome.drop_all(undigested, DropReason::NoDigest);
        if verifiable.is_empty() {
            return outcome;
        }

        let request = VerifyRequest::new(verifiable.clone());
        let response = match tokio::time::timeout(
            self.per_provider_timeout,
            provider.verify(&request),
        )
        .await
        {
            Ok(Ok(response)) => response,
            Ok(Err(error)) => {
                outcome.drop_all(verifiable, DropReason::VerifyFailed(error.to_string()));
                return outcome;
            }
            Err(_) => {
                let error = HostError::Timeout {
                    id: provider_id.to_string(),
                    timeout_ms: self.per_provider_timeout.as_millis() as u64,
                };
                outcome.drop_all(verifiable, DropReason::VerifyFailed(error.to_string()));
                return outcome;
            }
        };

        for frame in verifiable {
            // Correlate by full identity, never by position. A provider that
            // omits an answer gets `Unknown` — silence is not validity.
            match response.verdict_for(&frame) {
                Some(Verdict::Valid) => outcome.retained.push(frame),
                Some(Verdict::Stale { replacement_digest }) => outcome.drop_one(
                    frame,
                    DropReason::Stale {
                        replacement_digest: replacement_digest.clone(),
                    },
                ),
                Some(Verdict::Gone) => outcome.drop_one(frame, DropReason::Gone),
                Some(Verdict::Unknown) | None => outcome.drop_one(frame, DropReason::Unknown),
            }
        }
        outcome
    }

    /// Query a single provider by id, honoring the consent gate and the
    /// per-provider timeout. Querying an unconsented egress provider is
    /// [`HostError::ConsentRequired`] (legacy boolean) or
    /// [`HostError::ConsentScopeRequired`] (an off-machine scope with no
    /// receipt, §3), and the payload is never transmitted (§3.5).
    pub async fn query_provider(
        &self,
        id: &str,
        query: &ContextQuery,
    ) -> Result<ContextQueryResult, HostError> {
        Ok(self.query_provider_attested(id, query).await?.0)
    }

    /// [`query_provider`](Self::query_provider), plus what the host found when
    /// it checked the provider's attestations against its
    /// [`TrustStore`](crate::TrustStore) (`SPEC.md` §6.5,
    /// [ADR 0016](https://github.com/macanderson/context-graph-protocol/blob/main/docs/adr/0016-attestation-trust-roots.md)).
    ///
    /// Returns the result exactly as the provider served it and one
    /// [`FrameAttestationOutcome`] per frame in it — including the frames no
    /// attestation covered. **No frame is ever withheld for failing the check**
    /// (`SPEC.md` F9): the outcomes are a fact recorded beside the evidence, not
    /// a filter over it.
    pub async fn query_provider_attested(
        &self,
        id: &str,
        query: &ContextQuery,
    ) -> Result<(ContextQueryResult, Vec<FrameAttestationOutcome>), HostError> {
        let provider = self
            .providers
            .iter()
            .find(|p| p.id() == id)
            .ok_or_else(|| HostError::UnknownProvider(id.to_string()))?;

        match self.consent.evaluate(provider.id(), provider.info()) {
            ConsentDecision::Permitted => {}
            ConsentDecision::NeedsConsent => {
                return Err(HostError::ConsentRequired {
                    id: id.to_string(),
                    data_flow: provider.info().data_flow.clone(),
                });
            }
            ConsentDecision::NeedsReceipts(scopes) => {
                return Err(HostError::ConsentScopeRequired {
                    id: id.to_string(),
                    scopes,
                });
            }
        }

        let result =
            match tokio::time::timeout(self.per_provider_timeout, provider.query(query)).await {
                Ok(result) => result?,
                Err(_) => {
                    return Err(HostError::Timeout {
                        id: id.to_string(),
                        timeout_ms: self.per_provider_timeout.as_millis() as u64,
                    });
                }
            };

        // Two ids, deliberately. Trust is keyed on `id`, the host's own key for
        // this provider — the string the *operator* chose, in the same act as the
        // consent grant. The commitment is recomputed with the provider's
        // handshake-declared name, because that is the id §6.5.2 puts in the
        // signed preimage, and the only one the provider could have signed
        // against. Collapsing them made an honest signature read as
        // `CommitmentMismatch` — a tampering finding — for every operator whose
        // config id differed from the provider's declared name.
        let signing_id = provider.info().name.clone();
        let outcomes = self.trust.check_result_signed_as(id, &signing_id, &result);
        Ok((result, outcomes))
    }

    /// Fan a query out to every capability-matching provider concurrently,
    /// collecting a per-provider outcome. Each provider is consent-gated,
    /// timed out, and budget-audited independently — the crash-consistency
    /// contract means one provider's failure never affects another
    /// (task deliverables 3 + 5).
    pub async fn query_all(&self, query: &ContextQuery) -> FanOut {
        use futures_util::future::join_all;

        let futures: Vec<_> = self
            .providers
            .iter()
            .filter(|p| capability_matches(p.capabilities(), query))
            .map(|p| self.query_one_isolated(p.as_ref(), query))
            .collect();

        FanOut {
            outcomes: join_all(futures).await,
        }
    }

    /// Fan a query out under a **global** token budget, splitting it into a
    /// per-provider `max_tokens` share *before* building each provider's query
    /// (issue #15). Where [`query_all`](Self::query_all) hands the same
    /// `max_tokens` to every provider — so N honest providers can each spend the
    /// whole budget and the honest total is N× the intended prompt budget — this
    /// gives each capability-matching provider a slice of `global_budget`, so the
    /// honest legs sum to `<= global_budget`.
    ///
    /// `template` supplies every field of the query *except* `max_tokens`, which
    /// is overwritten per provider with its share from
    /// [`compose::budget_split`](crate::compose::budget_split) — an equal split
    /// by default, documented there as swappable for a weighted one. Only
    /// capability-matching providers (the same filter `query_all` applies) count
    /// toward the split and receive a query. Each leg is still consent-gated,
    /// timed out, and budget-audited exactly as in `query_all`, so a provider
    /// that overspends *its share* is dropped with a report by the existing B2
    /// audit — the split composes with per-leg honesty rather than replacing it.
    ///
    /// [`query_all`](Self::query_all) stays the un-budgeted legacy path.
    pub async fn query_all_budgeted(&self, template: &ContextQuery, global_budget: u32) -> FanOut {
        use futures_util::future::join_all;

        // The providers this query would reach — the same capability filter
        // `query_all` uses, so the split is over exactly the legs that run.
        let matching: Vec<&dyn ContextProvider> = self
            .providers
            .iter()
            .map(|provider| provider.as_ref())
            .filter(|provider| capability_matches(provider.capabilities(), template))
            .collect();

        // Shares are computed once, up front, from the count of matching
        // providers — before any provider's query is built.
        let shares = crate::compose::budget_split(global_budget, matching.len());

        // Materialize each provider's query so it outlives the borrowed fan-out
        // futures below; only `max_tokens` differs from the template.
        let queries: Vec<ContextQuery> = shares
            .iter()
            .map(|&share| ContextQuery {
                max_tokens: share,
                ..template.clone()
            })
            .collect();

        let futures: Vec<_> = matching
            .iter()
            .zip(queries.iter())
            .map(|(provider, query)| self.query_one_isolated(*provider, query))
            .collect();

        FanOut {
            outcomes: join_all(futures).await,
        }
    }

    /// Run one provider's leg of a fan-out, converting every failure mode into
    /// a value — never a propagated error that could abort sibling legs.
    async fn query_one_isolated(
        &self,
        provider: &dyn ContextProvider,
        query: &ContextQuery,
    ) -> ProviderOutcome {
        let id = provider.id().to_string();

        // Consent gate first: the query payload itself may carry workspace
        // content, so it must never reach an unconsented egress provider —
        // whether gated by the legacy boolean flag or by an unconsented
        // off-machine egress scope (§3).
        match self.consent.evaluate(provider.id(), provider.info()) {
            ConsentDecision::Permitted => {}
            ConsentDecision::NeedsConsent => {
                return ProviderOutcome::unattested(
                    id,
                    ProviderResult::ConsentRequired(provider.info().data_flow.clone()),
                );
            }
            ConsentDecision::NeedsReceipts(scopes) => {
                return ProviderOutcome::unattested(
                    id,
                    ProviderResult::ConsentScopeRequired {
                        data_flow: provider.info().data_flow.clone(),
                        missing: scopes,
                    },
                );
            }
        }

        let result =
            match tokio::time::timeout(self.per_provider_timeout, provider.query(query)).await {
                Ok(Ok(result)) => result,
                Ok(Err(error)) => {
                    return ProviderOutcome::unattested(id, ProviderResult::Failed(error));
                }
                Err(_) => {
                    let error = HostError::Timeout {
                        id: id.clone(),
                        timeout_ms: self.per_provider_timeout.as_millis() as u64,
                    };
                    return ProviderOutcome::unattested(id, ProviderResult::Failed(error));
                }
            };

        // Budget honesty, axis 1 (§7, B2): frames that sum above the query
        // budget are a lie about `token_cost`. Drop them, report loudly.
        if !result.respects_budget(query.max_tokens) {
            return ProviderOutcome::unattested(
                id,
                ProviderResult::BudgetLie {
                    claimed_tokens: result.total_token_cost(),
                    max_tokens: query.max_tokens,
                    dropped_frames: result.frames.len(),
                },
            );
        }

        // Budget honesty, axis 2 (§7, B4): more frames than `max_frames` is an
        // overspend the token budget never captures — each frame carries a
        // title, a citation label, and rendering chrome. Symmetric to B2: drop
        // the whole leg, report it loudly, never silently truncate.
        if !result.respects_frame_limit(query.max_frames) {
            return ProviderOutcome::unattested(
                id,
                ProviderResult::FrameFlood {
                    returned_frames: result.frames.len(),
                    max_frames: query.max_frames,
                },
            );
        }

        // Attestation last, and only over frames that already survived every
        // other gate. Signature verification is attacker-controlled work, so it
        // runs on a set the `max_frames` audit above has already bounded — and
        // it can only *annotate* that set. F9: whatever it finds, these frames
        // are served.
        //
        // Two ids again, as in `query_provider_attested`: `id` is the operator's
        // local routing key, `signing_id` is the name this provider declared at
        // handshake and signed under (§6.5.2). Matching evidence on `id` alone
        // would silently read every attested frame as unattested whenever the
        // two differ.
        let signing_id = provider.info().name.clone();
        let attestations = self.trust.check_result_signed_as(&id, &signing_id, &result);

        ProviderOutcome {
            provider_id: id,
            result: ProviderResult::Frames(result),
            attestations,
        }
    }

    /// Shut every provider down cleanly, consuming the host so its stdio
    /// children are reaped as they drop. Returns each provider's shutdown
    /// result so a caller can log stragglers.
    pub async fn shutdown(self) -> Vec<(String, Result<(), HostError>)> {
        let mut results = Vec::with_capacity(self.providers.len());
        for provider in &self.providers {
            results.push((provider.id().to_string(), provider.shutdown().await));
        }
        results
    }
}

/// The result of fanning one query out across all capability-matching
/// providers.
#[derive(Debug)]
pub struct FanOut {
    /// One entry per provider that matched the query's frame kinds, in
    /// registration order.
    pub outcomes: Vec<ProviderOutcome>,
}

impl FanOut {
    /// Every frame from providers that passed the consent gate, the timeout,
    /// and the budget-honesty audit — the frames a host may honestly compose
    /// into a prompt.
    pub fn accepted_frames(&self) -> impl Iterator<Item = &ContextFrame> {
        self.outcomes
            .iter()
            .filter_map(|outcome| match &outcome.result {
                ProviderResult::Frames(result) => Some(result.frames.iter()),
                _ => None,
            })
            .flatten()
    }

    /// The summed honest token cost of every accepted frame.
    pub fn total_accepted_tokens(&self) -> u64 {
        self.accepted_frames().map(|f| f.token_cost as u64).sum()
    }

    /// Every accepted frame paired with the id of the provider that served it
    /// — the input to deterministic composition (`docs/context-reuse.md` §1).
    pub fn accepted_with_provider(&self) -> impl Iterator<Item = (&str, &ContextFrame)> {
        self.outcomes
            .iter()
            .filter_map(|outcome| match &outcome.result {
                ProviderResult::Frames(result) => Some(
                    result
                        .frames
                        .iter()
                        .map(move |frame| (outcome.provider_id.as_str(), frame)),
                ),
                _ => None,
            })
            .flatten()
    }

    /// Compose every accepted frame into a byte-stable context block via the
    /// deterministic composition contract — canonical order, relevance-free
    /// rendering (`docs/context-reuse.md` §1). Two fan-outs over the same
    /// frame set compose to identical bytes, so an unchanged turn extends the
    /// provider's prompt cache instead of busting it.
    pub fn compose(&self) -> String {
        crate::compose::compose_context(self.accepted_with_provider())
    }

    /// Every accepted frame's attestation state, keyed by identity
    /// (`SPEC.md` §6.5,
    /// [ADR 0016](https://github.com/macanderson/context-graph-protocol/blob/main/docs/adr/0016-attestation-trust-roots.md)).
    ///
    /// A **total** account of the accepted frames: a frame the provider signed
    /// nothing for is [`Unattested`](crate::AttestationState::Unattested), so a
    /// missing entry never has to be guessed at. Frames from a leg that served
    /// none — failed, timed out, consent-gated, budget-dropped — are absent
    /// because there is nothing to say about frames the host does not hold.
    pub fn attestation_ledger(&self) -> AttestationLedger {
        self.outcomes
            .iter()
            .flat_map(|outcome| outcome.attestations.iter().cloned())
            .collect()
    }

    /// Whether any accepted frame is attested by a key this host trusts. A host
    /// rendering a "this evidence is signed" affordance asks this before
    /// drawing the chrome.
    pub fn any_attested(&self) -> bool {
        self.outcomes
            .iter()
            .flat_map(|outcome| outcome.attestations.iter())
            .any(|outcome| outcome.state.is_attested())
    }

    /// Compose every accepted frame into a prompt-ready block via the reference
    /// composer (issue #15): the R3 evidence preamble, cross-provider-deduped and
    /// value-ordered fenced frames packed under `global_budget`, a citation map,
    /// and a [`CompositionAudit`](crate::compose::CompositionAudit) explaining
    /// every included and excluded frame. Pair with
    /// [`Host::query_all_budgeted`](crate::Host::query_all_budgeted): the fan-out
    /// splits the budget across providers, and this packs the survivors under the
    /// same whole so `audit.tokens_used <= global_budget`.
    ///
    /// Each audit entry also carries this fan-out's
    /// [`attestation_ledger`](Self::attestation_ledger) state, so a reader can
    /// tell attested evidence from unattested. It changes nothing about which
    /// frames are chosen or where they land: acting on the state is a host's
    /// policy decision, and F9 forbids the composer from making it.
    ///
    /// Ranks by raw `score`
    /// ([`ScoreDescending`](crate::compose::ranking::ScoreDescending)); a host
    /// with its own cross-provider policy calls
    /// [`compose_for_prompt_with`](Self::compose_for_prompt_with).
    pub fn compose_for_prompt(&self, global_budget: u32) -> crate::compose::ComposedPrompt {
        self.compose_for_prompt_with(global_budget, &crate::compose::ranking::ScoreDescending)
    }

    /// [`compose_for_prompt`](Self::compose_for_prompt) under an explicit
    /// cross-provider ranking policy (`SPEC.md` §6.6, F10), still carrying this
    /// fan-out's attestation ledger into the audit.
    ///
    /// The two are orthogonal by construction: the strategy decides which
    /// frames are packed and in what order, and the ledger only describes what
    /// was found about each one. Nothing here lets an attestation state move a
    /// frame, which is what keeps F9 true no matter which policy a host picks.
    pub fn compose_for_prompt_with<S>(
        &self,
        global_budget: u32,
        strategy: &S,
    ) -> crate::compose::ComposedPrompt
    where
        S: crate::compose::ranking::RankingStrategy + ?Sized,
    {
        crate::compose::compose_for_prompt_attested(
            self.accepted_with_provider(),
            global_budget,
            strategy,
            &self.attestation_ledger(),
        )
    }

    /// Roll this fan-out up into a per-request [`UsageReport`] for metering
    /// (`docs/context-reuse.md` §2). One [`ProviderUsage`] per provider the
    /// query reached: accepted frames are itemized by stable identity and
    /// declared cost, a budget-lying provider's dropped frames count as
    /// rejected, and a failed or consent-gated provider served nothing.
    ///
    /// The report is a pure function of this fan-out plus the two host-supplied
    /// scalars: `budget_requested` is the query's `max_tokens`, and `as_of` is
    /// the accounting snapshot time (an RFC 3339 string the host stamps — the
    /// report's own as-of, *not* the query's bi-temporal `as_of` pin). The
    /// result always satisfies [`UsageReport::is_consistent`]: its totals are
    /// summed from the same served frames it itemizes.
    pub fn usage_report(&self, query: &ContextQuery, as_of: impl Into<String>) -> UsageReport {
        let providers: Vec<ProviderUsage> = self
            .outcomes
            .iter()
            .map(|outcome| {
                let provider_id = outcome.provider_id.clone();
                match &outcome.result {
                    ProviderResult::Frames(result) => {
                        let served_frames: Vec<ServedFrame> = result
                            .frames
                            .iter()
                            .map(|frame| ServedFrame {
                                frame: frame.identity(&provider_id),
                                token_cost: frame.token_cost,
                            })
                            .collect();
                        let token_cost = served_frames.iter().map(|s| s.token_cost as u64).sum();
                        ProviderUsage {
                            provider_id,
                            frames_served: served_frames.len() as u32,
                            frames_rejected: 0,
                            token_cost,
                            served_frames,
                        }
                    }
                    // A budget lie: the provider's frames were dropped whole,
                    // so nothing was served and every offered frame is rejected.
                    ProviderResult::BudgetLie { dropped_frames, .. } => ProviderUsage {
                        provider_id,
                        frames_served: 0,
                        frames_rejected: *dropped_frames as u32,
                        token_cost: 0,
                        served_frames: vec![],
                    },
                    // A frame flood (§B4): same shape as a budget lie — the
                    // whole leg was dropped, so every returned frame is rejected
                    // and nothing was served.
                    ProviderResult::FrameFlood {
                        returned_frames, ..
                    } => ProviderUsage {
                        provider_id,
                        frames_served: 0,
                        frames_rejected: *returned_frames as u32,
                        token_cost: 0,
                        served_frames: vec![],
                    },
                    // Consent-gated or failed: no frames offered, none served,
                    // none rejected — the leg simply contributed nothing.
                    ProviderResult::ConsentRequired(_)
                    | ProviderResult::ConsentScopeRequired { .. }
                    | ProviderResult::Failed(_) => ProviderUsage {
                        provider_id,
                        frames_served: 0,
                        frames_rejected: 0,
                        token_cost: 0,
                        served_frames: vec![],
                    },
                }
            })
            .collect();

        let budget_consumed = providers.iter().map(|p| p.token_cost).sum();
        UsageReport {
            budget_requested: query.max_tokens,
            budget_consumed,
            as_of: as_of.into(),
            providers,
        }
    }

    /// Providers that failed (error, timeout, or crash), with their errors.
    pub fn failures(&self) -> impl Iterator<Item = (&str, &HostError)> {
        self.outcomes
            .iter()
            .filter_map(|outcome| match &outcome.result {
                ProviderResult::Failed(error) => Some((outcome.provider_id.as_str(), error)),
                _ => None,
            })
    }

    /// Providers whose frames were dropped for exceeding the query budget —
    /// the loud report the host must surface, never swallow (SPEC.md §7, B2).
    pub fn budget_liars(&self) -> impl Iterator<Item = &ProviderOutcome> {
        self.outcomes
            .iter()
            .filter(|outcome| matches!(outcome.result, ProviderResult::BudgetLie { .. }))
    }

    /// Providers whose frames were dropped for exceeding `max_frames` — the
    /// frame-count twin of [`budget_liars`](Self::budget_liars), surfaced
    /// loudly rather than silently truncated (SPEC.md §7, B4).
    pub fn frame_floods(&self) -> impl Iterator<Item = &ProviderOutcome> {
        self.outcomes
            .iter()
            .filter(|outcome| matches!(outcome.result, ProviderResult::FrameFlood { .. }))
    }
}

/// One provider's outcome within a [`FanOut`].
#[derive(Debug)]
pub struct ProviderOutcome {
    pub provider_id: String,
    pub result: ProviderResult,
    /// What the host found when it checked this provider's attestations
    /// against its [`TrustStore`](crate::TrustStore) — one entry per accepted
    /// frame, including the frames no attestation covered (`SPEC.md` §6.5,
    /// [ADR 0016](https://github.com/macanderson/context-graph-protocol/blob/main/docs/adr/0016-attestation-trust-roots.md)).
    ///
    /// Empty for a leg that served no frames — consent-gated, failed, timed
    /// out, or dropped for a budget lie. Nothing here ever removes a frame
    /// from `result` (`SPEC.md` F9).
    pub attestations: Vec<FrameAttestationOutcome>,
}

impl ProviderOutcome {
    /// An outcome from a leg that produced no frames to attest — every failure
    /// mode, and every gate that ran before the frames arrived.
    fn unattested(provider_id: String, result: ProviderResult) -> Self {
        Self {
            provider_id,
            result,
            attestations: Vec::new(),
        }
    }
}

/// What became of one provider's leg of a fan-out — a total function over
/// every failure mode, so no leg can abort another.
#[derive(Debug)]
pub enum ProviderResult {
    /// Frames the host accepted: passed consent, timeout, and budget honesty.
    Frames(ContextQueryResult),
    /// The provider's frames summed above the query budget — a `token_cost`
    /// lie. Dropped and reported (SPEC.md §7, B2).
    BudgetLie {
        claimed_tokens: u64,
        max_tokens: u32,
        dropped_frames: usize,
    },
    /// The provider returned more frames than `max_frames` — a frame-count
    /// overspend. Dropped whole and reported, symmetric to a [`BudgetLie`]
    /// (SPEC.md §7, B4).
    FrameFlood {
        returned_frames: usize,
        max_frames: u32,
    },
    /// Skipped: an egress provider (declaring no scopes) without recorded
    /// boolean consent. The query payload was **not** transmitted (§3.5).
    ConsentRequired(DataFlow),
    /// Skipped: the provider declares off-machine egress scope(s) with no
    /// recorded consent receipt (`docs/context-reuse.md` §3). `missing` names
    /// the scopes lacking a receipt. The query payload was **not** transmitted.
    ConsentScopeRequired {
        data_flow: DataFlow,
        missing: Vec<EgressScope>,
    },
    /// The provider errored, timed out, or crashed mid-query.
    Failed(HostError),
}

/// Why a held frame was dropped by [`Host::verify_frames`]
/// (`docs/context-reuse.md` §4).
///
/// The first three mirror the provider's [`Verdict`]s; the rest are host-side
/// reasons a frame could not be revalidated at all. Either way the frame leaves
/// the composed context — the difference matters only for deciding whether to
/// re-query it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DropReason {
    /// The provider answered `stale`: the frame exists but its content changed.
    /// Carries the provider's current digest when it offered one.
    Stale { replacement_digest: Option<String> },
    /// The provider answered `gone` — the frame no longer exists.
    Gone,
    /// The provider answered `unknown`, or returned no verdict for the frame at
    /// all. Silence is not validity.
    Unknown,
    /// The frame carries no `content_digest`, so it cannot be revalidated
    /// (§1, requirement D4).
    NoDigest,
    /// The provider does not advertise the `verify` capability, so the host
    /// falls back to re-querying its frames (§4, requirement V3).
    VerifyUnsupported,
    /// No provider with this frame's `provider_id` is registered with the host.
    UnknownProvider,
    /// The verify request itself failed — a transport error or a timeout.
    VerifyFailed(String),
}

impl DropReason {
    /// Whether re-querying the provider could recover usable content for this
    /// frame. False only for [`Gone`](Self::Gone) — every other reason means
    /// the host simply doesn't have a trustworthy copy and should ask again.
    pub fn warrants_requery(&self) -> bool {
        !matches!(self, Self::Gone)
    }
}

/// One dropped frame and why (`docs/context-reuse.md` §4).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DroppedFrame {
    /// The identity that was dropped.
    pub frame: FrameId,
    /// Why it was dropped.
    pub reason: DropReason,
}

/// The result of revalidating a held frame set (`docs/context-reuse.md` §4).
///
/// Partitions the input into frames the host may keep reusing and frames it
/// must drop. The partition is **total and default-deny**: every input identity
/// appears in exactly one of the two lists, and it lands in `retained` only on
/// an explicit `valid`.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct VerifyOutcome {
    /// Frames that verified `valid` — safe to keep reusing, and the frames
    /// whose byte-stable reuse §1's canonical ordering was built to protect.
    pub retained: Vec<FrameId>,
    /// Frames that must leave the composed context, each with its reason.
    pub dropped: Vec<DroppedFrame>,
}

impl VerifyOutcome {
    /// The dropped frames worth re-querying — everything except `gone`, which
    /// is not there to re-fetch.
    pub fn requery(&self) -> impl Iterator<Item = &FrameId> {
        self.dropped
            .iter()
            .filter(|dropped| dropped.reason.warrants_requery())
            .map(|dropped| &dropped.frame)
    }

    /// Whether an identity was dropped.
    pub fn was_dropped(&self, frame: &FrameId) -> bool {
        self.dropped.iter().any(|dropped| &dropped.frame == frame)
    }

    /// The reason an identity was dropped, if it was.
    pub fn drop_reason(&self, frame: &FrameId) -> Option<&DropReason> {
        self.dropped
            .iter()
            .find(|dropped| &dropped.frame == frame)
            .map(|dropped| &dropped.reason)
    }

    fn drop_one(&mut self, frame: FrameId, reason: DropReason) {
        self.dropped.push(DroppedFrame { frame, reason });
    }

    fn drop_all(&mut self, frames: impl IntoIterator<Item = FrameId>, reason: DropReason) {
        for frame in frames {
            self.drop_one(frame, reason.clone());
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use contextgraph_types::Grantor;
    use contextgraph_types::capability::QueryCapability;
    use contextgraph_types::{Capabilities, ContextFrame, FrameKind, ProviderInfo};
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, Ordering};

    /// A configurable in-process provider for exercising the router.
    struct FakeProvider {
        id: String,
        info: ProviderInfo,
        capabilities: Capabilities,
        behavior: Behavior,
        queried: Arc<AtomicBool>,
    }

    enum Behavior {
        Frames(Vec<ContextFrame>),
        Fail(String),
        Slow(Duration),
    }

    impl FakeProvider {
        fn new(id: &str, egress: bool, behavior: Behavior) -> Self {
            Self::with_data_flow(
                id,
                DataFlow {
                    reads: true,
                    writes: false,
                    egress,
                    egress_scopes: vec![],
                },
                behavior,
            )
        }

        /// A provider declaring egress scopes, for the scope-consent gate.
        fn scoped(id: &str, scopes: Vec<EgressScope>, behavior: Behavior) -> Self {
            Self::with_data_flow(
                id,
                DataFlow {
                    reads: true,
                    writes: false,
                    egress: true,
                    egress_scopes: scopes,
                },
                behavior,
            )
        }

        fn with_data_flow(id: &str, data_flow: DataFlow, behavior: Behavior) -> Self {
            Self {
                id: id.into(),
                info: ProviderInfo {
                    name: id.into(),
                    version: "0.0.1".into(),
                    data_flow,
                },
                capabilities: Capabilities {
                    query: QueryCapability {
                        kinds: vec!["doc".into()],
                    },
                    ..Capabilities::default()
                },
                behavior,
                queried: Arc::new(AtomicBool::new(false)),
            }
        }
    }

    #[async_trait]
    impl ContextProvider for FakeProvider {
        fn id(&self) -> &str {
            &self.id
        }
        fn info(&self) -> &ProviderInfo {
            &self.info
        }
        fn capabilities(&self) -> &Capabilities {
            &self.capabilities
        }
        async fn query(&self, _query: &ContextQuery) -> Result<ContextQueryResult, HostError> {
            self.queried.store(true, Ordering::SeqCst);
            match &self.behavior {
                Behavior::Frames(frames) => Ok(ContextQueryResult {
                    frames: frames.clone(),
                    truncated: false,
                    dropped_estimate: None,
                    ..Default::default()
                }),
                Behavior::Fail(message) => Err(HostError::Provider {
                    id: self.id.clone(),
                    code: None,
                    message: message.clone(),
                }),
                Behavior::Slow(duration) => {
                    tokio::time::sleep(*duration).await;
                    Ok(ContextQueryResult {
                        frames: vec![],
                        truncated: false,
                        dropped_estimate: None,
                        ..Default::default()
                    })
                }
            }
        }
    }

    fn frame(id: &str, cost: u32) -> ContextFrame {
        ContextFrame {
            id: id.into(),
            kind: FrameKind::Doc,
            title: id.into(),
            content: Some("c".into()),
            content_digest: None,
            uri: None,
            representation: Default::default(),
            content_fidelity: None,
            canonical_content_hash: None,
            content_ref: None,
            transform: None,
            minimum_content_fidelity: None,
            inline_content_requirement: None,
            score: 0.5,
            token_cost: cost,
            canonical_token_cost: None,
            tokenizer_ref: None,
            valid_from: None,
            valid_to: None,
            recorded_at: None,
            provenance: vec![],
            citation_label: Some(id.into()),
            embedding: None,
            relations: vec![],
        }
    }

    fn query() -> ContextQuery {
        ContextQuery {
            goal: "g".into(),
            query_text: None,
            embedding: None,
            kinds: vec![],
            anchors: vec![],
            max_frames: 10,
            max_tokens: 1000,
            as_of: None,
            representation_preferences: vec![],
        }
    }

    #[tokio::test]
    async fn query_all_collects_frames_from_healthy_providers() {
        let mut host = Host::new();
        host.register(Box::new(FakeProvider::new(
            "a",
            false,
            Behavior::Frames(vec![frame("f1", 100), frame("f2", 100)]),
        )));
        host.register(Box::new(FakeProvider::new(
            "b",
            false,
            Behavior::Frames(vec![frame("f3", 50)]),
        )));

        let fanout = host.query_all(&query()).await;
        assert_eq!(fanout.outcomes.len(), 2);
        assert_eq!(fanout.accepted_frames().count(), 3);
        assert_eq!(fanout.total_accepted_tokens(), 250);
    }

    #[tokio::test]
    async fn a_budgeted_fan_out_keeps_honest_legs_under_the_global_budget() {
        // Four honest providers, each returning a frame that fits its equal share
        // of a 1000-token global budget (250 each). Under the budgeted fan-out
        // every leg is accepted and the honest total stays under the whole — the
        // overrun `query_all` allows (each provider spending the full budget) is
        // closed by allocating shares before fan-out.
        let mut host = Host::new();
        for id in ["a", "b", "c", "d"] {
            host.register(Box::new(FakeProvider::new(
                id,
                false,
                Behavior::Frames(vec![frame(&format!("{id}1"), 200)]),
            )));
        }
        let template = query(); // max_tokens on the template is ignored by the split
        let fanout = host.query_all_budgeted(&template, 1000).await;
        assert_eq!(
            fanout.accepted_frames().count(),
            4,
            "each share fits its leg"
        );
        assert!(
            fanout.total_accepted_tokens() <= 1000,
            "honest legs must sum to <= the global budget, got {}",
            fanout.total_accepted_tokens()
        );
        assert_eq!(fanout.budget_liars().count(), 0);
    }

    #[tokio::test]
    async fn the_budget_split_enforces_a_per_leg_ceiling_the_flat_fan_out_does_not() {
        // A provider returning a 300-token frame against a 1000-token whole split
        // four ways gets a 250-token share — so its frame is a budget lie against
        // *its share* and is dropped, even though 300 <= the 1000 global. The
        // same frame sails through the un-budgeted `query_all` (300 <= 1000),
        // which is exactly the global overrun the split exists to prevent.
        let mut host = Host::new();
        host.register(Box::new(FakeProvider::new(
            "greedy",
            false,
            Behavior::Frames(vec![frame("g", 300)]),
        )));
        for id in ["b", "c", "d"] {
            host.register(Box::new(FakeProvider::new(
                id,
                false,
                Behavior::Frames(vec![frame(&format!("{id}1"), 100)]),
            )));
        }

        let budgeted = host.query_all_budgeted(&query(), 1000).await;
        assert!(
            budgeted
                .budget_liars()
                .any(|outcome| outcome.provider_id == "greedy"),
            "a leg overspending its share is dropped by the existing B2 audit"
        );
        assert!(
            budgeted
                .accepted_with_provider()
                .all(|(id, _)| id != "greedy"),
            "the greedy leg contributes nothing under the split"
        );

        // Un-budgeted, the same 300-cost frame is within the flat 1000 budget and
        // is accepted — the overrun the split closes.
        let flat = host.query_all(&query()).await;
        assert!(flat.accepted_with_provider().any(|(id, _)| id == "greedy"));
    }

    #[tokio::test]
    async fn the_host_composes_the_same_frame_set_to_identical_bytes_across_turns() {
        // The reference host's deterministic-composition round trip
        // (`docs/context-reuse.md` §1): the same frame set, fanned out twice,
        // composes to byte-identical bytes — so an unchanged turn extends the
        // provider's prompt-cache prefix instead of forfeiting it.
        let mut host = Host::new();
        host.register(Box::new(FakeProvider::new(
            "prov-b",
            false,
            Behavior::Frames(vec![frame("f2", 100), frame("f1", 100)]),
        )));
        host.register(Box::new(FakeProvider::new(
            "prov-a",
            false,
            Behavior::Frames(vec![frame("f3", 50)]),
        )));

        let first = host.query_all(&query()).await.compose();
        let second = host.query_all(&query()).await.compose();
        assert_eq!(
            first, second,
            "an unchanged frame set must compose to identical bytes"
        );
        // All three frames are present, each fenced exactly once, and the
        // lower-sorting provider id renders first regardless of registration
        // order.
        assert_eq!(first.matches("<frame ").count(), 3);
        assert!(first.find("prov-a").unwrap() < first.find("prov-b").unwrap());
    }

    #[tokio::test]
    async fn a_fan_out_rolls_up_into_a_self_consistent_usage_report() {
        let mut host = Host::new();
        host.register(Box::new(FakeProvider::new(
            "a",
            false,
            Behavior::Frames(vec![frame("f1", 100), frame("f2", 100)]),
        )));
        // 1200 tokens against a 1000-token budget: dropped as a budget lie.
        host.register(Box::new(FakeProvider::new(
            "liar",
            false,
            Behavior::Frames(vec![frame("big", 1200)]),
        )));

        let query = query();
        let fanout = host.query_all(&query).await;
        let report = fanout.usage_report(&query, "2026-07-21T00:00:00Z");

        assert_eq!(report.budget_requested, 1000);
        assert_eq!(report.budget_consumed, 200);
        assert_eq!(report.as_of, "2026-07-21T00:00:00Z");
        // The report re-sums from its own itemized frames…
        assert!(report.is_consistent());
        assert!(report.within_budget());
        // …and its consumed total equals an INDEPENDENT re-sum of the accepted
        // frames — the arithmetic identity, not a build-then-assert tautology.
        let independent: u64 = fanout.accepted_frames().map(|f| f.token_cost as u64).sum();
        assert_eq!(report.budget_consumed, independent);
        assert_eq!(report.budget_consumed, fanout.total_accepted_tokens());

        let a = report
            .providers
            .iter()
            .find(|p| p.provider_id == "a")
            .expect("provider a is in the report");
        assert_eq!(a.frames_served, 2);
        assert_eq!(a.frames_rejected, 0);
        assert_eq!(a.token_cost, 200);
        // Served frames are itemized by stable identity for audit walk-back.
        let ids: Vec<&str> = a
            .served_frames
            .iter()
            .map(|s| s.frame.frame_id.as_str())
            .collect();
        assert!(ids.contains(&"f1") && ids.contains(&"f2"));
        assert!(a.served_frames.iter().all(|s| s.frame.provider_id == "a"));

        let liar = report
            .providers
            .iter()
            .find(|p| p.provider_id == "liar")
            .expect("the liar is still accounted for");
        assert_eq!(liar.frames_served, 0);
        assert_eq!(liar.frames_rejected, 1);
        assert_eq!(liar.token_cost, 0);
        assert!(liar.served_frames.is_empty());
    }

    #[tokio::test]
    async fn a_provider_lying_about_token_cost_has_its_frames_dropped_loudly() {
        let mut host = Host::new();
        // 1200 tokens claimed against a 1000-token budget: a lie.
        host.register(Box::new(FakeProvider::new(
            "liar",
            false,
            Behavior::Frames(vec![frame("big", 1200)]),
        )));
        host.register(Box::new(FakeProvider::new(
            "honest",
            false,
            Behavior::Frames(vec![frame("ok", 200)]),
        )));

        let fanout = host.query_all(&query()).await;
        // The liar's frames never reach the accepted set…
        assert_eq!(fanout.accepted_frames().count(), 1);
        assert_eq!(fanout.total_accepted_tokens(), 200);
        // …and the lie is reported loudly, not swallowed.
        let liars: Vec<_> = fanout.budget_liars().collect();
        assert_eq!(liars.len(), 1);
        assert_eq!(liars[0].provider_id, "liar");
        match liars[0].result {
            ProviderResult::BudgetLie {
                claimed_tokens,
                max_tokens,
                dropped_frames,
            } => {
                assert_eq!(claimed_tokens, 1200);
                assert_eq!(max_tokens, 1000);
                assert_eq!(dropped_frames, 1);
            }
            _ => unreachable!(),
        }
    }

    #[tokio::test]
    async fn a_provider_returning_more_than_max_frames_has_them_dropped_loudly() {
        // §B4, the frame-count twin of the budget lie: 12 individually-cheap
        // frames respect the token budget but blow `max_frames = 10`. Before
        // this audit they sailed straight through, because only the token sum
        // was checked.
        let query = query();
        let flood: Vec<ContextFrame> = (0..12).map(|i| frame(&format!("f{i}"), 1)).collect();
        // The token budget alone would have accepted every one of them — the
        // frame cap is the only thing that catches this.
        let as_result = ContextQueryResult {
            frames: flood.clone(),
            truncated: false,
            dropped_estimate: None,
            ..Default::default()
        };
        assert!(as_result.respects_budget(query.max_tokens));
        assert!(!as_result.respects_frame_limit(query.max_frames));

        let mut host = Host::new();
        host.register(Box::new(FakeProvider::new(
            "flood",
            false,
            Behavior::Frames(flood),
        )));
        host.register(Box::new(FakeProvider::new(
            "honest",
            false,
            Behavior::Frames(vec![frame("ok", 200)]),
        )));

        let fanout = host.query_all(&query).await;
        // The flooder's frames never reach the accepted set; the honest peer's do.
        assert_eq!(fanout.accepted_frames().count(), 1);
        assert_eq!(fanout.total_accepted_tokens(), 200);

        // The overspend is reported loudly, not silently truncated.
        let floods: Vec<_> = fanout.frame_floods().collect();
        assert_eq!(floods.len(), 1);
        assert_eq!(floods[0].provider_id, "flood");
        match floods[0].result {
            ProviderResult::FrameFlood {
                returned_frames,
                max_frames,
            } => {
                assert_eq!(returned_frames, 12);
                assert_eq!(max_frames, 10);
            }
            _ => unreachable!(),
        }
        // A flood is not a budget lie — the two audits are distinct.
        assert_eq!(fanout.budget_liars().count(), 0);

        // The usage report accounts every flooded frame as rejected, none served.
        let report = fanout.usage_report(&query, "2026-07-21T00:00:00Z");
        let flooder = report
            .providers
            .iter()
            .find(|p| p.provider_id == "flood")
            .expect("the flooder is still accounted for");
        assert_eq!(flooder.frames_served, 0);
        assert_eq!(flooder.frames_rejected, 12);
        assert_eq!(flooder.token_cost, 0);
        assert!(flooder.served_frames.is_empty());
        assert!(report.is_consistent());
    }

    #[tokio::test]
    async fn one_failing_provider_never_poisons_the_others() {
        let mut host = Host::new();
        host.register(Box::new(FakeProvider::new(
            "healthy",
            false,
            Behavior::Frames(vec![frame("f", 10)]),
        )));
        host.register(Box::new(FakeProvider::new(
            "broken",
            false,
            Behavior::Fail("kaboom".into()),
        )));

        let fanout = host.query_all(&query()).await;
        assert_eq!(fanout.accepted_frames().count(), 1);
        let failures: Vec<_> = fanout.failures().collect();
        assert_eq!(failures.len(), 1);
        assert_eq!(failures[0].0, "broken");
    }

    #[tokio::test]
    async fn a_slow_provider_is_timed_out_without_stalling_the_fan_out() {
        let mut host = Host::with_timeout(Duration::from_millis(50));
        host.register(Box::new(FakeProvider::new(
            "fast",
            false,
            Behavior::Frames(vec![frame("f", 10)]),
        )));
        host.register(Box::new(FakeProvider::new(
            "slow",
            false,
            Behavior::Slow(Duration::from_secs(30)),
        )));

        let fanout = host.query_all(&query()).await;
        assert_eq!(fanout.accepted_frames().count(), 1);
        let failures: Vec<_> = fanout.failures().collect();
        assert_eq!(failures.len(), 1);
        assert_eq!(failures[0].0, "slow");
        assert!(matches!(failures[0].1, HostError::Timeout { .. }));
    }

    #[tokio::test]
    async fn an_egress_provider_is_not_queried_until_consent_is_recorded() {
        let mut host = Host::new();
        let provider = FakeProvider::new("github", true, Behavior::Frames(vec![frame("f", 10)]));
        let queried = provider.queried.clone();
        host.register(Box::new(provider));

        // Without consent: skipped, and — critically — query() never ran, so
        // the payload never left.
        let fanout = host.query_all(&query()).await;
        assert_eq!(fanout.accepted_frames().count(), 0);
        assert!(!queried.load(Ordering::SeqCst), "payload must not be sent");
        assert!(matches!(
            fanout.outcomes[0].result,
            ProviderResult::ConsentRequired(_)
        ));
        // Direct query is the named error.
        assert!(matches!(
            host.query_provider("github", &query()).await,
            Err(HostError::ConsentRequired { .. })
        ));

        // After consent: queried and its frames accepted.
        host.record_consent(ConsentRecord::new(
            "github",
            DataFlow {
                reads: true,
                writes: false,
                egress: true,
                egress_scopes: vec![],
            },
            "issue titles leave to github.com",
        ));
        let fanout = host.query_all(&query()).await;
        assert!(queried.load(Ordering::SeqCst));
        assert_eq!(fanout.accepted_frames().count(), 1);
    }

    #[tokio::test]
    async fn a_scoped_egress_provider_is_not_queried_until_a_receipt_is_recorded() {
        let mut host = Host::new();
        let provider = FakeProvider::scoped(
            "cloud",
            vec![EgressScope::ThirdPartyModel],
            Behavior::Frames(vec![frame("f", 10)]),
        );
        let queried = provider.queried.clone();
        let info = provider.info().clone();
        host.register(Box::new(provider));

        // Without a receipt: skipped as a scope-consent gap, and the payload
        // never left.
        let fanout = host.query_all(&query()).await;
        assert_eq!(fanout.accepted_frames().count(), 0);
        assert!(!queried.load(Ordering::SeqCst), "payload must not be sent");
        match &fanout.outcomes[0].result {
            ProviderResult::ConsentScopeRequired { missing, .. } => {
                assert_eq!(missing, &vec![EgressScope::ThirdPartyModel]);
            }
            other => panic!("expected ConsentScopeRequired, got {other:?}"),
        }
        // Direct query is the scope-specific typed error naming what would leave.
        match host.query_provider("cloud", &query()).await {
            Err(HostError::ConsentScopeRequired { scopes, .. }) => {
                assert_eq!(scopes, vec![EgressScope::ThirdPartyModel]);
            }
            other => panic!("expected ConsentScopeRequired error, got {other:?}"),
        }

        // After a receipt for the declared scope: queried and accepted.
        host.record_receipt(ConsentReceipt::new(
            "cloud",
            &info,
            EgressScope::ThirdPartyModel,
            Grantor::Human("ops@oxagen.sh".into()),
            "2026-07-21T00:00:00Z",
        ));
        let fanout = host.query_all(&query()).await;
        assert!(queried.load(Ordering::SeqCst));
        assert_eq!(fanout.accepted_frames().count(), 1);
    }

    #[tokio::test]
    async fn query_provider_reports_unknown_ids() {
        let host = Host::new();
        assert!(matches!(
            host.query_provider("nope", &query()).await,
            Err(HostError::UnknownProvider(_))
        ));
    }

    // ---- context/verify (§4) ----

    use contextgraph_types::{FrameVerdict, VerifyResponse};
    use std::collections::HashMap as StdHashMap;

    /// A provider that answers `context/verify` from a scripted verdict table.
    struct VerifyingProvider {
        id: String,
        capabilities: Capabilities,
        /// frame id -> verdict. A frame absent from the table gets no verdict
        /// entry at all, exercising the "silence is not validity" path.
        verdicts: StdHashMap<String, Verdict>,
        /// When set, `verify` fails instead of answering.
        verify_error: Option<String>,
        /// Identities this provider was actually asked about.
        asked: Arc<std::sync::Mutex<Vec<FrameId>>>,
    }

    impl VerifyingProvider {
        fn new(id: &str, supports_verify: bool, verdicts: &[(&str, Verdict)]) -> Self {
            Self {
                id: id.into(),
                capabilities: Capabilities {
                    query: QueryCapability {
                        kinds: vec!["doc".into()],
                    },
                    verify: supports_verify,
                    ..Capabilities::default()
                },
                verdicts: verdicts
                    .iter()
                    .map(|(f, v)| ((*f).to_string(), v.clone()))
                    .collect(),
                verify_error: None,
                asked: Arc::new(std::sync::Mutex::new(Vec::new())),
            }
        }

        fn failing(id: &str) -> Self {
            let mut provider = Self::new(id, true, &[]);
            provider.verify_error = Some("index unavailable".into());
            provider
        }
    }

    #[async_trait]
    impl ContextProvider for VerifyingProvider {
        fn id(&self) -> &str {
            &self.id
        }
        fn info(&self) -> &ProviderInfo {
            // Local-only: nothing here is about consent.
            static_info()
        }
        fn capabilities(&self) -> &Capabilities {
            &self.capabilities
        }
        async fn query(&self, _query: &ContextQuery) -> Result<ContextQueryResult, HostError> {
            Ok(ContextQueryResult {
                frames: vec![],
                truncated: false,
                dropped_estimate: None,
                ..Default::default()
            })
        }
        async fn verify(&self, request: &VerifyRequest) -> Result<VerifyResponse, HostError> {
            self.asked
                .lock()
                .unwrap()
                .extend(request.frames.iter().cloned());
            if let Some(message) = &self.verify_error {
                return Err(HostError::Provider {
                    id: self.id.clone(),
                    code: None,
                    message: message.clone(),
                });
            }
            Ok(VerifyResponse::new(
                request
                    .frames
                    .iter()
                    .filter_map(|frame| {
                        self.verdicts
                            .get(&frame.frame_id)
                            .map(|verdict| FrameVerdict::new(frame.clone(), verdict.clone()))
                    })
                    .collect(),
            ))
        }
    }

    fn static_info() -> &'static ProviderInfo {
        use std::sync::OnceLock;
        static INFO: OnceLock<ProviderInfo> = OnceLock::new();
        INFO.get_or_init(|| ProviderInfo {
            name: "verifier".into(),
            version: "0.0.1".into(),
            data_flow: DataFlow {
                reads: true,
                writes: false,
                egress: false,
                egress_scopes: vec![],
            },
        })
    }

    fn held(provider: &str, frame: &str, digest: Option<&str>) -> FrameId {
        FrameId::new(provider, frame, digest.map(String::from))
    }

    #[tokio::test]
    async fn a_stale_frame_is_dropped_and_a_valid_one_is_retained() {
        // The core §4 guarantee: the host demonstrably evicts a frame the
        // provider says has changed, and keeps the one it vouches for.
        let mut host = Host::new();
        host.register(Box::new(VerifyingProvider::new(
            "docs",
            true,
            &[
                ("fresh", Verdict::Valid),
                (
                    "changed",
                    Verdict::Stale {
                        replacement_digest: Some("sha256:new".into()),
                    },
                ),
            ],
        )));

        let fresh = held("docs", "fresh", Some("sha256:a"));
        let changed = held("docs", "changed", Some("sha256:b"));
        let outcome = host.verify_frames(&[fresh.clone(), changed.clone()]).await;

        assert_eq!(outcome.retained, vec![fresh]);
        assert!(outcome.was_dropped(&changed));
        assert_eq!(
            outcome.drop_reason(&changed),
            Some(&DropReason::Stale {
                replacement_digest: Some("sha256:new".into())
            })
        );
        // A stale frame is worth re-fetching; the replacement digest tells the
        // host what it would be getting.
        assert_eq!(outcome.requery().collect::<Vec<_>>(), vec![&changed]);
    }

    #[tokio::test]
    async fn a_gone_frame_is_dropped_and_not_worth_re_querying() {
        let mut host = Host::new();
        host.register(Box::new(VerifyingProvider::new(
            "docs",
            true,
            &[("deleted", Verdict::Gone)],
        )));
        let deleted = held("docs", "deleted", Some("sha256:a"));
        let outcome = host.verify_frames(std::slice::from_ref(&deleted)).await;

        assert!(outcome.retained.is_empty());
        assert_eq!(outcome.drop_reason(&deleted), Some(&DropReason::Gone));
        // Nothing to re-fetch — `gone` is the one reason that doesn't warrant it.
        assert_eq!(outcome.requery().count(), 0);
    }

    #[tokio::test]
    async fn an_unknown_verdict_and_a_missing_verdict_both_drop_the_frame() {
        // Silence is not validity: a provider that omits an answer must not
        // have that read as "still good".
        let mut host = Host::new();
        host.register(Box::new(VerifyingProvider::new(
            "docs",
            true,
            &[("shrugged", Verdict::Unknown)],
        )));
        let shrugged = held("docs", "shrugged", Some("sha256:a"));
        let unanswered = held("docs", "never-mentioned", Some("sha256:b"));
        let outcome = host
            .verify_frames(&[shrugged.clone(), unanswered.clone()])
            .await;

        assert!(outcome.retained.is_empty());
        assert_eq!(outcome.drop_reason(&shrugged), Some(&DropReason::Unknown));
        assert_eq!(outcome.drop_reason(&unanswered), Some(&DropReason::Unknown));
        assert_eq!(outcome.requery().count(), 2);
    }

    #[tokio::test]
    async fn a_provider_without_verify_support_is_never_asked_and_falls_back_to_requery() {
        // The capability gate (V3): the host doesn't send a verify request at
        // all, it just re-queries.
        let mut host = Host::new();
        let provider = VerifyingProvider::new("docs", false, &[("anything", Verdict::Valid)]);
        let asked = provider.asked.clone();
        host.register(Box::new(provider));

        let frame = held("docs", "anything", Some("sha256:a"));
        let outcome = host.verify_frames(std::slice::from_ref(&frame)).await;

        assert!(asked.lock().unwrap().is_empty(), "must not be asked");
        assert!(outcome.retained.is_empty());
        assert_eq!(
            outcome.drop_reason(&frame),
            Some(&DropReason::VerifyUnsupported)
        );
        assert_eq!(outcome.requery().count(), 1);
    }

    #[tokio::test]
    async fn a_frame_without_a_digest_is_unverifiable_and_never_sent() {
        // §1 D4: no digest, no revalidation — and the request only carries
        // answerable identities.
        let mut host = Host::new();
        let provider = VerifyingProvider::new("docs", true, &[("bare", Verdict::Valid)]);
        let asked = provider.asked.clone();
        host.register(Box::new(provider));

        let bare = held("docs", "bare", None);
        let digested = held("docs", "digested", Some("sha256:a"));
        let outcome = host.verify_frames(&[bare.clone(), digested.clone()]).await;

        assert_eq!(outcome.drop_reason(&bare), Some(&DropReason::NoDigest));
        let asked = asked.lock().unwrap().clone();
        assert_eq!(
            asked,
            vec![digested],
            "only verifiable identities go on the wire"
        );
    }

    #[tokio::test]
    async fn a_failed_verify_drops_that_providers_frames_without_touching_another() {
        // Per-provider isolation, same contract as a query fan-out leg.
        let mut host = Host::new();
        host.register(Box::new(VerifyingProvider::failing("broken")));
        host.register(Box::new(VerifyingProvider::new(
            "healthy",
            true,
            &[("good", Verdict::Valid)],
        )));

        let broken = held("broken", "any", Some("sha256:a"));
        let good = held("healthy", "good", Some("sha256:b"));
        let outcome = host.verify_frames(&[broken.clone(), good.clone()]).await;

        assert_eq!(outcome.retained, vec![good], "one failure must not poison");
        assert!(matches!(
            outcome.drop_reason(&broken),
            Some(DropReason::VerifyFailed(_))
        ));
        assert!(outcome.requery().any(|frame| frame == &broken));
    }

    #[tokio::test]
    async fn frames_from_an_unregistered_provider_are_dropped_not_ignored() {
        let host = Host::new();
        let orphan = held("never-registered", "f", Some("sha256:a"));
        let outcome = host.verify_frames(std::slice::from_ref(&orphan)).await;
        assert!(outcome.retained.is_empty());
        assert_eq!(
            outcome.drop_reason(&orphan),
            Some(&DropReason::UnknownProvider)
        );
    }

    #[tokio::test]
    async fn held_frames_are_grouped_into_one_request_per_provider() {
        // Verification costs bytes, not tokens — so it must not cost a round
        // trip per frame either.
        let mut host = Host::new();
        let docs = VerifyingProvider::new(
            "docs",
            true,
            &[("a", Verdict::Valid), ("b", Verdict::Valid)],
        );
        let asked = docs.asked.clone();
        host.register(Box::new(docs));

        let outcome = host
            .verify_frames(&[
                held("docs", "a", Some("sha256:1")),
                held("docs", "b", Some("sha256:2")),
            ])
            .await;
        assert_eq!(outcome.retained.len(), 2);
        // Both identities arrived together in a single verify call.
        assert_eq!(asked.lock().unwrap().len(), 2);
    }

    #[tokio::test]
    async fn the_partition_is_total_so_every_held_frame_is_accounted_for() {
        let mut host = Host::new();
        host.register(Box::new(VerifyingProvider::new(
            "docs",
            true,
            &[("keep", Verdict::Valid), ("drop", Verdict::Gone)],
        )));
        let input = vec![
            held("docs", "keep", Some("sha256:1")),
            held("docs", "drop", Some("sha256:2")),
            held("docs", "nodigest", None),
            held("elsewhere", "orphan", Some("sha256:3")),
        ];
        let outcome = host.verify_frames(&input).await;
        assert_eq!(
            outcome.retained.len() + outcome.dropped.len(),
            input.len(),
            "every held identity must land in exactly one bucket"
        );
        for frame in &input {
            assert!(
                outcome.retained.contains(frame) || outcome.was_dropped(frame),
                "{frame:?} was silently lost"
            );
        }
    }

    #[tokio::test]
    async fn verifying_an_empty_held_set_is_a_no_op() {
        let host = Host::new();
        let outcome = host.verify_frames(&[]).await;
        assert_eq!(outcome, VerifyOutcome::default());
    }
}