openlatch-client 0.6.3

//! Economics event assembly + emission (D-13).
//!
//! Assembles one `ai.openlatch.economics.*` CloudEvent from the staged
//! request-side observation + the response-side usage, runs it through the
//! **existing** `PrivacyFilter`, and hands it to `cloud_tx` (fire-and-forget) —
//! mirroring the `ai.openlatch.config.*` path (`build_modified_event` +
//! `log_and_forward`). It reuses `crate::cloud::CloudEvent`; it does **not**
//! touch the outbox (that is worker-only).
//!
//! > **The client emits tokens, never money (F-26).** No `ai.openlatch.cost.*`
//! > USD fields, no `pricebook.version` — the platform computes cost at ingest.
//! > The event carries **no prompt content** (C-10); the two-sided privacy test
//! > proves the sentinel is present in the forwarded body and absent from every
//! > emitted byte.

use tokio::sync::mpsc::Sender;

use crate::cloud::CloudEvent;
use crate::privacy::PrivacyFilter;

use super::billing::BillingMode;
use super::capture::{CaptureGap, CostBasis, PricingInputs, Usage};
use super::churn::ChurnFinding;
use super::session::Resolved;
use super::transforms::{TransformDecision, TRANSFORM_WIRE_KEYS};
use super::wire_format::WireFormat;

/// CloudEvents type for a per-call economics/usage row.
pub const ECON_TYPE: &str = "ai.openlatch.economics.usage";

/// Request-side facts staged at `observe_request` time, completed on the
/// response side and handed here for assembly.
#[derive(Clone, Debug)]
pub struct Observation {
    /// `false` = plan-01-style no-op (opaque path / measurement disabled / a
    /// caught observe panic) → nothing is emitted.
    pub measured: bool,
    /// `ai.openlatch.event.id` — UUIDv7 idempotency key, minted at request start.
    pub event_id: String,
    /// `ai.openlatch.event.occurred_at` — client clock at request start (RFC3339).
    pub occurred_at: String,
    /// `gen_ai.request.model`.
    pub model: Option<String>,
    /// Whether `model` is in the known (D-21) set — drives `unknown_model`.
    pub model_known: bool,
    /// `ai.openlatch.billing.mode`.
    pub billing: BillingMode,
    /// `ai.openlatch.session.install_id`.
    pub install_id: String,
    /// Resolved attribution triple + assurance.
    pub session: Resolved,
    /// Pricing-input modifiers.
    pub pricing: PricingInputs,
    /// Prefix-churn finding, when the prefix diverged.
    pub churn: Option<ChurnFinding>,
    /// Request body byte length — the estimate proxy for the interrupted path.
    pub request_body_len: usize,
    /// Whether the request carried a `cache_control` breakpoint.
    pub has_breakpoint: bool,
    /// The single highest-net would-have transform decision (I-3-01), when a
    /// baseline L-1/L-2 rule matched the parsed request clone. `None` when no rule
    /// matched (including every L-0-only request — L-0 is not a removal lever).
    /// Observe-only: it never changed the forwarded bytes.
    pub transform: Option<TransformDecision>,
    /// Privacy-safe schema-2 Optimize selection and delivery result.
    pub optimize: Option<OptimizeEvidence>,
    /// The wire format this request spoke, resolved from its route.
    ///
    /// `gen_ai.provider.name` is read from it rather than from a constant: the
    /// model relay forwards for more than one protocol now, and a hardcoded
    /// `"anthropic"` would label every Codex row with the wrong provider. It is
    /// resolved from the FORMAT and not from the upstream host, so a customer
    /// pointing us at a corporate gateway still reports the provider whose
    /// protocol they are speaking.
    pub wire_format: WireFormat,
    /// The agent an unattributed request of this format belongs to, when this
    /// install wired exactly one agent to speak it
    /// ([`WiringState::sole_wired_agent_for`](super::preflight::WiringState::sole_wired_agent_for)).
    ///
    /// Consulted only when the session cascade produced no source. It is not a
    /// second guess at the session — it says nothing about WHICH session — it
    /// only recovers the agent's NAME from the one thing this host knows for
    /// certain: what it configured. `None` when two agents share the format, or
    /// none does.
    pub attributable_agent: Option<&'static str>,
}

impl Observation {
    /// The plan-01 no-op observation — nothing is emitted for it.
    pub fn none() -> Self {
        Observation {
            measured: false,
            event_id: String::new(),
            occurred_at: String::new(),
            model: None,
            model_known: false,
            billing: BillingMode::Unknown,
            install_id: String::new(),
            session: Resolved::unknown(),
            pricing: PricingInputs::default(),
            churn: None,
            request_body_len: 0,
            has_breakpoint: false,
            transform: None,
            optimize: None,
            // Neutral, like every other value here. The opaque-path observation
            // overwrites it with the route's real format.
            wire_format: WireFormat::Unknown,
            attributable_agent: None,
        }
    }

    /// Attribution `agent_id` (NOT NULL): the resolved session agent id, falling
    /// back to the install id when unresolved (the same PII-free value in Phase 1).
    fn resolved_agent_id(&self) -> String {
        self.session
            .agent_id
            .clone()
            .unwrap_or_else(|| self.install_id.clone())
    }

    /// Attribution `source` (NOT NULL). Three rungs, most-certain first.
    ///
    /// 1. **The session's own source** — a hook observed this session and said
    ///    what it was. Nothing beats being told.
    /// 2. **The sole agent on this install wired to speak this request's
    ///    format** — see [`Observation::attributable_agent`]. Not a guess about
    ///    the session, a fact about our own configuration: if only one agent
    ///    here was ever pointed at this protocol, a request in it had one
    ///    possible author. **Inside an intercepted tunnel (I-4 D-41)** this
    ///    rung is instead the CONNECT host's sole detected `ProxyEnv` declarer
    ///    (`WiringState::sole_interceptor_for`, set by `proxy_any`'s
    ///    `tunnel_rung`) — the format rung never runs there, because a Cline
    ///    turn to `api.anthropic.com` would otherwise read as Claude Code.
    /// 3. **`"unknown"`** — and it stays. Naming an agent we did not identify
    ///    is wrong even when it happens to be right: it leaves no value meaning
    ///    "we could not tell", so the platform's
    ///    `(organization_id, source, agent_id)` join becomes unfalsifiable.
    ///    `source` is an open string on the wire (`AgentType` carries
    ///    `Unknown(String)`), so `"unknown"` round-trips with no schema change.
    ///
    /// Rung 2 exists because rung 3 was throwing away something this host
    /// already held. An unattributed Claude turn still arrived on
    /// `/v1/messages`, still reported `gen_ai.provider.name = "anthropic"` and
    /// `gen_ai.request.model = "claude-opus-5"`, and still ran on a host where
    /// Claude Code was the only agent wired to that protocol — and we reported
    /// `unknown` beside all three. That is not caution, it is an unread fact.
    fn resolved_source(&self) -> String {
        self.session
            .source
            .clone()
            .or_else(|| self.attributable_agent.map(str::to_string))
            .unwrap_or_else(|| "unknown".to_string())
    }
}

#[derive(Clone, Debug)]
pub struct OptimizeEvidence {
    pub context: crate::zone_eval::OptimizeContext,
    pub enforced: bool,
    pub result: &'static str,
    pub second_optimize_degraded: bool,
}

/// The vendor prefixes a gateway puts in front of a model id, and the provider
/// each one names as LiteLLM spells it.
///
/// OpenRouter and its imitators address a model as `vendor/model`
/// (`anthropic/claude-sonnet-4.5`) over the OpenAI-compatible route. Priced as
/// sent, that id matches nothing in a pricebook keyed on the vendor's own id.
/// Only vendors whose ids the pricebook carries are listed: an open-weights
/// `meta-llama/…` or a self-hosted `qwen2.5-coder:14b` is left exactly as it
/// came, and stays unpriced — which is the honest answer for a model nobody
/// bills per token.
const GATEWAY_VENDOR_PREFIXES: &[(&str, &str)] = &[
    ("anthropic", "anthropic"),
    ("openai", "openai"),
    ("google", "gemini"),
    ("gemini", "gemini"),
    ("x-ai", "xai"),
    ("xai", "xai"),
    ("mistralai", "mistral"),
    ("mistral", "mistral"),
    ("deepseek", "deepseek"),
    ("cohere", "cohere"),
];

/// `(gen_ai.provider.name, gen_ai.request.model)` in the vendor's own terms, as
/// LiteLLM — the pricebook's source — spells them.
///
/// Three rewrites, each of a spelling and never of a fact:
///
/// - **Google's format is `gemini`.** `google-generate-content` is Gemini's own
///   method shape (AI Studio and Vertex alike), and LiteLLM files those models
///   under `gemini`. A `models/` prefix Google's own clients sometimes send is
///   dropped.
/// - **A gateway's `vendor/` prefix is lifted into the provider**
///   ([`GATEWAY_VENDOR_PREFIXES`]) — only for a listed vendor.
/// - **Anthropic's versions are hyphenated.** OpenRouter writes
///   `claude-sonnet-4.5`; Anthropic and LiteLLM write `claude-sonnet-4-5`.
///
/// Anything else passes through: the format's own provider and the model as
/// the request named it.
pub fn priced_identity(format_provider: &str, model: &str) -> (String, String) {
    let model = model.trim();
    let (provider, model) = match model.split_once('/') {
        Some((vendor, rest)) if !rest.is_empty() => {
            match GATEWAY_VENDOR_PREFIXES
                .iter()
                .find(|(prefix, _)| prefix.eq_ignore_ascii_case(vendor))
            {
                Some((_, provider)) => ((*provider).to_string(), rest.to_string()),
                None if format_provider == "google" && vendor == "models" => {
                    ("gemini".to_string(), rest.to_string())
                }
                None => (format_provider.to_string(), model.to_string()),
            }
        }
        _ => (format_provider.to_string(), model.to_string()),
    };
    // Every spelling of Google's Gemini surface — our format's `google`, the
    // OTel GenAI `gcp.gemini` / `gcp.vertex_ai`, LiteLLM's `vertex_ai` — is
    // priced under the one key the pricebook carries.
    let provider = match provider.as_str() {
        "google" | "gcp.gemini" | "gcp.vertex_ai" | "vertex_ai" => "gemini".to_string(),
        _ => provider,
    };
    let model = if provider == "anthropic" {
        model.replace('.', "-")
    } else {
        model
    };
    (provider, model)
}

/// Assemble the economics `data` object — **exactly** the canonical-contract wire
/// fields, tokens only. Returns the `data` JSON (pre-privacy-filter).
pub fn assemble_data(
    obs: &Observation,
    usage: &Usage,
    basis: CostBasis,
    gap: Option<CaptureGap>,
    cache_preserved: bool,
) -> serde_json::Value {
    // Attribution: agent_id/source are NOT NULL; when unresolved they fall back
    // to the install_id / default source (in Phase 1 the same PII-free values).
    let agent_id = obs.resolved_agent_id();
    let source = obs.resolved_source();
    let (provider, model) = priced_identity(
        obs.wire_format.provider(),
        obs.model.as_deref().unwrap_or_default(),
    );

    let mut data = serde_json::json!({
        // --- token facts (the five raw counts + the 5m/1h split) ---
        "gen_ai.usage.input_tokens": usage.input_tokens,
        "gen_ai.usage.cache_creation.input_tokens": usage.cache_write,
        "gen_ai.usage.cache_read.input_tokens": usage.cache_read,
        "gen_ai.usage.output_tokens": usage.output_tokens,
        "ai.openlatch.cache.ephemeral_5m_input_tokens": usage.eph_5m,
        "ai.openlatch.cache.ephemeral_1h_input_tokens": usage.eph_1h,

        // --- model / provider ---
        "gen_ai.request.model": model,
        "gen_ai.provider.name": provider,

        // --- capture / billing bases ---
        "ai.openlatch.cost.basis": basis.as_str(),
        "ai.openlatch.billing.mode": obs.billing.as_str(),

        // --- pricing-input modifiers (batch/fast_mode NOT-NULL; geo nullable) ---
        "ai.openlatch.request.batch": obs.pricing.batch,
        "ai.openlatch.request.fast_mode": obs.pricing.fast_mode,
        "ai.openlatch.request.inference_geo": obs.pricing.inference_geo,

        // --- attribution (agent_id/source/install_id NOT-NULL) ---
        "ai.openlatch.session.agent_id": agent_id,
        "ai.openlatch.session.source": source,
        "ai.openlatch.session.install_id": obs.install_id,
        "ai.openlatch.session.agent_session_id": obs.session.session_id,
        crate::model_relay::session::ASSURANCE_KEY: obs.session.assurance.as_str(),

        // --- idempotency + dating ---
        "ai.openlatch.event.id": obs.event_id,
        "ai.openlatch.event.occurred_at": obs.occurred_at,

        // --- the (weak) cache-preserved signal ---
        "ai.openlatch.cache.preserved": cache_preserved,
    });

    // capture.gap — nullable; only set when capture was incomplete.
    data["ai.openlatch.capture.gap"] = match gap {
        Some(g) => serde_json::Value::String(g.as_str().to_string()),
        None => serde_json::Value::Null,
    };

    // prefix.* — all nullable; present only when the prefix diverged. Never
    // content: only the classification + offsets + the host-local finding id.
    let (offset, layer, class, block_index, byte_len, finding_id) = match &obs.churn {
        Some(f) => (
            serde_json::json!(f.divergence_offset),
            serde_json::json!(f.churn_layer.as_str()),
            serde_json::json!(f.churn_class.as_str()),
            serde_json::json!(f.churn_block_index),
            serde_json::json!(f.churn_byte_len),
            serde_json::json!(f.finding_id),
        ),
        None => (
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::Value::Null,
        ),
    };
    data["ai.openlatch.prefix.divergence_offset"] = offset;
    data["ai.openlatch.prefix.churn_layer"] = layer;
    data["ai.openlatch.prefix.churn_class"] = class;
    data["ai.openlatch.prefix.churn_block_index"] = block_index;
    data["ai.openlatch.prefix.churn_byte_len"] = byte_len;
    data["ai.openlatch.prefix.finding_id"] = finding_id;

    // transform.* — all nullable; present only when a baseline L-1/L-2 rule matched
    // the parsed request clone (the block is absent-today when no transform ran).
    // The tuple is observe-only (`ladder_stage` always `observe`, never `applied`).
    // `finding_id` is deliberately NOT emitted here — it is not in the contract.
    match &obs.transform {
        Some(t) => {
            if let serde_json::Value::Object(fields) = t.to_wire_object() {
                for (key, value) in fields {
                    data[key] = value;
                }
            }
        }
        None => {
            for key in TRANSFORM_WIRE_KEYS {
                data[key] = serde_json::Value::Null;
            }
        }
    }

    data
}

/// Assemble the full CloudEvent (envelope + agent_id) from a `data` object.
pub fn assemble_event(
    obs: &Observation,
    mut data: serde_json::Value,
    privacy: &PrivacyFilter,
) -> CloudEvent {
    // Run the event `data` through the existing privacy filter BEFORE it leaves
    // the process — belt-and-suspenders, since we only ever put derived facts in
    // it. This is the second half of the two-sided C-10 guarantee.
    crate::privacy::filter_event_with(&mut data, privacy);

    let source = obs.resolved_source();
    let agent_id = obs.resolved_agent_id();

    let mut envelope = serde_json::json!({
        "specversion": "1.0",
        "id": crate::envelope::new_event_id(),
        "source": source,
        "type": ECON_TYPE,
        "time": crate::envelope::current_timestamp(),
        "datacontenttype": "application/json",
        "data": data,
    });

    // The format this request actually spoke, from the route that carried it.
    //
    // The egress worker stamps `wireformat` for every producer off a per-source
    // map, and that map answers only for an agent wired to a single request
    // plane. This row came from the listener that served the request, so it
    // knows — and for an agent whose providers differ per slot it is the only
    // thing that does. `Unknown` is left unstamped rather than written: the
    // worker's map may still be able to name the format from the source, and an
    // explicit "unknown" here would talk it out of trying.
    if obs.wire_format != WireFormat::Unknown {
        envelope["wireformat"] = serde_json::json!(obs.wire_format.as_str());
        // The provider is a property of the format, so it rides the same gate:
        // an unresolved route names no provider rather than an `unknown` one.
        envelope["olmodelprovider"] = serde_json::json!(
            priced_identity(
                obs.wire_format.provider(),
                obs.model.as_deref().unwrap_or_default()
            )
            .0
        );
    }

    // The three normalized facts this row shares with the hook plane, under the
    // names the hook plane already uses.
    //
    // `data` carries all three as well, spelled the economics contract's way
    // (`gen_ai.request.model`, `ai.openlatch.session.assurance`). That is the
    // cost-row vocabulary and it stays; these are the envelope-level names a
    // consumer can group both planes by without knowing which one it is looking
    // at. Same value, twice, deliberately — the alternative is a consumer that
    // reads a different key depending on the event type it happens to have.
    envelope["olsessionassurance"] = serde_json::json!(obs.session.assurance.as_str());
    if let Some(model) = obs.model.as_deref().filter(|m| !m.trim().is_empty()) {
        let (_, model) = priced_identity(obs.wire_format.provider(), model);
        envelope["olmodelslug"] = serde_json::json!(model);
    }
    if let Some(optimize) = &obs.optimize {
        envelope["oloptimize"] = serde_json::to_value(&optimize.context).unwrap_or_default();
        envelope["olenforced"] = serde_json::json!(i64::from(optimize.enforced));
        envelope["olresult"] = serde_json::json!(optimize.result);
        if optimize.second_optimize_degraded {
            envelope["oloptimizedegraded"] = serde_json::json!("second_optimize");
        }
    }

    // The session id goes on the ENVELOPE, not only inside `data`.
    //
    // `subject` is where the schema says an agent session id lives — "consumers
    // can group events by subject for per-session analytics" — and it is what
    // the hook path has always stamped. The platform's session grouping is an
    // index on it (`organization_id, subject, time`), so an economics event
    // without one is not merely missing a field: it belongs to no session, and
    // the model relay's whole contribution drops out of the per-session view while
    // the hook events beside it show up. `data` carried the same id all along,
    // in the one place nothing joins on.
    //
    // Absent when no session was resolved. That is the honest shape — a request
    // the cascade could not place has no session to name, and stamping the
    // install id or an empty string here would invent a session that the
    // platform would then group other events into.
    if let Some(sid) = obs.session.session_id.as_deref() {
        envelope["subject"] = serde_json::Value::String(sid.to_string());
    }

    CloudEvent { envelope, agent_id }
}

/// Build and fire-and-forget the economics event. Non-blocking `try_send`
/// (CLOUD-01/CLOUD-09) — a full channel drops the event with a warn, never
/// blocking the response path. A no-op when the observation was not measured or
/// no sink is wired.
#[allow(clippy::too_many_arguments)]
pub fn build_and_emit(
    obs: &Observation,
    usage: &Usage,
    basis: CostBasis,
    gap: Option<CaptureGap>,
    cache_preserved: bool,
    privacy: &PrivacyFilter,
    cloud_tx: Option<&Sender<CloudEvent>>,
) {
    if !obs.measured {
        return;
    }
    let Some(tx) = cloud_tx else {
        return;
    };
    let data = assemble_data(obs, usage, basis, gap, cache_preserved);
    let event = assemble_event(obs, data, privacy);
    if tx.try_send(event).is_err() {
        tracing::warn!(
            code = crate::error::ERR_CLOUD_UNREACHABLE,
            "model relay: cloud channel full — economics event dropped"
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::model_relay::session::Assurance;

    fn obs() -> Observation {
        Observation {
            measured: true,
            event_id: "0190aaaa-bbbb-cccc-dddd-eeeeeeeeeeee".to_string(),
            occurred_at: "2026-07-23T12:00:00Z".to_string(),
            model: Some("claude-opus-4-8".to_string()),
            model_known: true,
            billing: BillingMode::ApiKey,
            install_id: "agt_install".to_string(),
            session: Resolved {
                agent_id: Some("agt_install".to_string()),
                source: Some("claude-code".to_string()),
                session_id: Some("sess_a".to_string()),
                assurance: Assurance::Attested,
            },
            pricing: PricingInputs::default(),
            churn: None,
            request_body_len: 100,
            has_breakpoint: true,
            transform: None,
            optimize: None,
            // The fixture is a Claude Code turn, so it keeps asserting
            // `"anthropic"` below on its own merits rather than by default.
            wire_format: WireFormat::AnthropicMessages,
            // Unset by default: the fixture pins `source: Some("claude-code")`,
            // so rung 1 answers and rung 2 must not be what makes it pass.
            attributable_agent: None,
        }
    }

    /// A relay row answers the same three questions a hook event does, off the
    /// same three keys.
    ///
    /// `data` carries all of them too, spelled the economics contract's way.
    /// These are the envelope-level names, so a consumer can group both planes
    /// without branching on the event type first.
    #[test]
    fn the_row_carries_the_shared_facts_under_the_shared_names() {
        let mut observation = obs();
        observation.wire_format = WireFormat::GoogleGenerateContent;
        observation.model = Some("gemini-3.1-flash-lite".to_string());
        observation.session.assurance = Assurance::Inferred;

        let ev = assemble_event(
            &observation,
            serde_json::json!({}),
            &PrivacyFilter::new(&[]),
        );
        assert_eq!(ev.envelope["olsessionassurance"], "inferred");
        assert_eq!(ev.envelope["olmodelslug"], "gemini-3.1-flash-lite");
        // Google's format is priced as LiteLLM files it: `gemini`.
        assert_eq!(ev.envelope["olmodelprovider"], "gemini");

        // No prompt, on this plane, ever. The relay emits tokens and never
        // content — the two-sided privacy guarantee depends on it.
        assert!(ev.envelope.get("olprompt").is_none());

        // An unresolved route names no provider rather than an `unknown` one.
        observation.wire_format = WireFormat::Unknown;
        let ev = assemble_event(
            &observation,
            serde_json::json!({}),
            &PrivacyFilter::new(&[]),
        );
        assert!(ev.envelope.get("olmodelprovider").is_none());
        assert_eq!(
            ev.envelope["olsessionassurance"], "inferred",
            "assurance does not depend on the route"
        );
    }

    /// The row carries the format its own request spoke.
    ///
    /// The egress worker stamps `wireformat` from a per-source map, which has no
    /// entry for an agent wired per provider slot — so a Cline row went out
    /// `unknown` while its `data` named `google`. This listener served the
    /// request, so it is the one producer that knows.
    #[test]
    fn the_row_carries_the_format_its_request_spoke() {
        let mut observation = obs();
        observation.wire_format = WireFormat::GoogleGenerateContent;
        let ev = assemble_event(
            &observation,
            serde_json::json!({}),
            &PrivacyFilter::new(&[]),
        );
        assert_eq!(ev.envelope["wireformat"], "google-generate-content");

        // A format the route could not name is left for the worker's map to
        // answer, rather than written as an explicit "unknown" that would talk
        // it out of trying.
        observation.wire_format = WireFormat::Unknown;
        let ev = assemble_event(
            &observation,
            serde_json::json!({}),
            &PrivacyFilter::new(&[]),
        );
        assert!(
            ev.envelope.get("wireformat").is_none(),
            "an unobserved format is absent, never asserted"
        );
    }

    /// The session id must reach the ENVELOPE, not just `data`.
    ///
    /// Found in the field on 2026-09-06: a Codex turn produced a correct
    /// economics event — attested, right session id in
    /// `data["ai.openlatch.session.agent_session_id"]`, exact provider-reported
    /// counts — that appeared under no session in the platform, while the hook
    /// events beside it did. The platform groups sessions on an index over
    /// `subject`, and this envelope left it null.
    #[test]
    fn the_envelope_carries_the_session_id_as_subject() {
        let ev = assemble_event(&obs(), serde_json::json!({}), &PrivacyFilter::new(&[]));
        assert_eq!(
            ev.envelope["subject"], "sess_a",
            "an economics event with no subject belongs to no session"
        );
    }

    #[test]
    fn an_unattributed_request_names_no_subject() {
        // The honest shape: the cascade could not place it, so there is no
        // session to name. Stamping the install id or an empty string here would
        // invent a session the platform would then group other events into.
        let mut o = obs();
        o.session.session_id = None;
        let ev = assemble_event(&o, serde_json::json!({}), &PrivacyFilter::new(&[]));
        assert!(
            ev.envelope.get("subject").is_none(),
            "no session resolved must mean no subject, not a fabricated one"
        );
    }

    /// D-03. `gen_ai.provider.name` follows the FORMAT, so a Codex row says
    /// `openai` — where a hardcoded const said `anthropic` for every provider on
    /// the host. Resolved from the format and not from the upstream host: a
    /// corporate gateway is still the provider's protocol, and reporting the
    /// gateway's hostname would be junk.
    #[test]
    fn provider_name_follows_the_format() {
        let usage = Usage::default();

        let mut anthropic = obs();
        anthropic.wire_format = WireFormat::AnthropicMessages;
        let data = assemble_data(&anthropic, &usage, CostBasis::ProviderReported, None, true);
        assert_eq!(data["gen_ai.provider.name"], "anthropic");

        let mut responses = obs();
        responses.wire_format = WireFormat::OpenAiResponses;
        let data = assemble_data(&responses, &usage, CostBasis::ProviderReported, None, true);
        assert_eq!(data["gen_ai.provider.name"], "openai");
    }

    /// `gen_ai.provider.name` is ALWAYS present and non-empty. The platform
    /// prices a row with no provider as `anthropic`, so a Gemini row that
    /// dropped it would be priced at Anthropic's rates.
    #[test]
    fn every_usage_row_names_its_provider() {
        let formats = [
            (
                WireFormat::AnthropicMessages,
                "claude-opus-4-8",
                "anthropic",
            ),
            (WireFormat::OpenAiResponses, "gpt-5", "openai"),
            (
                WireFormat::OpenAiChatCompletions,
                "qwen2.5-coder:14b",
                "openai-compatible",
            ),
            (
                WireFormat::GoogleGenerateContent,
                "gemini-2.5-pro",
                "gemini",
            ),
            (WireFormat::OllamaNative, "llama3:8b", "ollama"),
            (WireFormat::Unknown, "", "unknown"),
        ];
        for (format, model, provider) in formats {
            for model in [Some(model.to_string()), None] {
                let mut o = obs();
                o.wire_format = format;
                o.model = model.clone();
                let data = assemble_data(
                    &o,
                    &Usage::default(),
                    CostBasis::ProviderReported,
                    None,
                    true,
                );
                assert_eq!(
                    data["gen_ai.provider.name"], provider,
                    "{format:?} model={model:?}"
                );
            }
        }
        // A Gemini row: provider `gemini`, bare model id.
        let mut gemini = obs();
        gemini.wire_format = WireFormat::GoogleGenerateContent;
        gemini.model = Some("models/gemini-2.5-pro".to_string());
        let data = assemble_data(
            &gemini,
            &Usage::default(),
            CostBasis::ProviderReported,
            None,
            true,
        );
        assert_eq!(data["gen_ai.provider.name"], "gemini");
        assert_eq!(data["gen_ai.request.model"], "gemini-2.5-pro");

        // Every other spelling of the Gemini surface, and a gateway prefix.
        for (provider, model, want_model) in [
            ("gcp.gemini", "gemini-2.5-pro", "gemini-2.5-pro"),
            ("gcp.vertex_ai", "gemini-2.5-pro", "gemini-2.5-pro"),
            ("vertex_ai", "gemini-2.5-flash", "gemini-2.5-flash"),
            (
                "openai-compatible",
                "gemini/gemini-2.5-pro",
                "gemini-2.5-pro",
            ),
        ] {
            assert_eq!(
                priced_identity(provider, model),
                ("gemini".to_string(), want_model.to_string()),
                "{provider} {model}"
            );
        }
        // An unknown vendor prefix is not guessed at: left as-is, unpriced.
        assert_eq!(
            priced_identity("openai-compatible", "mystery/model-x"),
            (
                "openai-compatible".to_string(),
                "mystery/model-x".to_string()
            )
        );
    }

    /// Model ids reach the pricebook in the vendor's own spelling: a gateway's
    /// `vendor/` prefix lifted into the provider, Google's format as `gemini`,
    /// Anthropic versions hyphenated — and a self-hosted model left alone.
    #[test]
    fn model_ids_are_normalized_to_the_vendors_own() {
        let cases = [
            (
                "openai-compatible",
                "anthropic/claude-sonnet-4.5",
                "anthropic",
                "claude-sonnet-4-5",
            ),
            ("openai-compatible", "openai/gpt-4.1", "openai", "gpt-4.1"),
            (
                "openai-compatible",
                "google/gemini-2.5-pro",
                "gemini",
                "gemini-2.5-pro",
            ),
            ("openai-compatible", "x-ai/grok-4", "xai", "grok-4"),
            ("google", "gemini-2.5-pro", "gemini", "gemini-2.5-pro"),
            (
                "google",
                "models/gemini-2.5-flash",
                "gemini",
                "gemini-2.5-flash",
            ),
            (
                "anthropic",
                "claude-opus-4-8",
                "anthropic",
                "claude-opus-4-8",
            ),
            // Self-hosted and open-weights ids are not a vendor's: unpriced, as sent.
            (
                "openai-compatible",
                "qwen2.5-coder:14b",
                "openai-compatible",
                "qwen2.5-coder:14b",
            ),
            (
                "openai-compatible",
                "meta-llama/llama-3.3-70b",
                "openai-compatible",
                "meta-llama/llama-3.3-70b",
            ),
            ("ollama", "llama3:8b", "ollama", "llama3:8b"),
        ];
        for (format, model, provider, normalized) in cases {
            assert_eq!(
                priced_identity(format, model),
                (provider.to_string(), normalized.to_string()),
                "{format} {model}"
            );
        }

        let mut routed = obs();
        routed.wire_format = WireFormat::OpenAiChatCompletions;
        routed.model = Some("anthropic/claude-sonnet-4.5".to_string());
        let data = assemble_data(
            &routed,
            &Usage::default(),
            CostBasis::ProviderReported,
            None,
            true,
        );
        assert_eq!(data["gen_ai.provider.name"], "anthropic");
        assert_eq!(data["gen_ai.request.model"], "claude-sonnet-4-5");
    }

    #[test]
    fn assembled_data_has_exact_contract_fields_and_no_usd() {
        let usage = Usage {
            input_tokens: 50,
            cache_read: 100_000,
            cache_write: 0,
            eph_5m: 0,
            eph_1h: 0,
            output_tokens: 12,
        };
        let data = assemble_data(&obs(), &usage, CostBasis::ProviderReported, None, true);

        // Tokens present, EXACT wire names.
        assert_eq!(data["gen_ai.usage.input_tokens"], 50);
        assert_eq!(data["gen_ai.usage.cache_read.input_tokens"], 100_000);
        assert_eq!(data["gen_ai.provider.name"], "anthropic");
        assert_eq!(data["ai.openlatch.cost.basis"], "provider_reported");
        assert_eq!(data["ai.openlatch.billing.mode"], "api_key");
        assert_eq!(data["ai.openlatch.session.assurance"], "attested");
        assert_eq!(data["ai.openlatch.request.batch"], false);
        assert!(data["ai.openlatch.request.inference_geo"].is_null());
        assert!(data["ai.openlatch.capture.gap"].is_null());
        assert!(data["ai.openlatch.prefix.finding_id"].is_null());

        // No transform ran → the whole transform.* block is nullable-absent, and
        // the contract-absent finding_id is never emitted.
        assert!(data["ai.openlatch.transform.rule_id"].is_null());
        assert!(data["ai.openlatch.transform.outcome"].is_null());
        assert!(data["ai.openlatch.transform.tokens_net"].is_null());
        assert!(data.get("ai.openlatch.transform.finding_id").is_none());

        // C-3: total input is input + cache_creation + cache_read = 100050.
        let total = data["gen_ai.usage.input_tokens"].as_u64().unwrap()
            + data["gen_ai.usage.cache_creation.input_tokens"]
                .as_u64()
                .unwrap()
            + data["gen_ai.usage.cache_read.input_tokens"]
                .as_u64()
                .unwrap();
        assert_eq!(total, 100_050);

        // F-26: the client NEVER emits money or a pricebook version.
        let s = data.to_string();
        assert!(!s.contains("cost_input"));
        assert!(!s.contains("cost_total"));
        assert!(!s.contains("pricebook"));
        assert!(!s.contains("ai.openlatch.cost.input"));
    }

    #[test]
    fn a_matching_transform_populates_the_nullable_tuple() {
        use crate::model_relay::transforms::evaluate_would_have;

        // A trimmable conversation: two large removed messages over a tiny tail.
        let mut messages = vec![
            serde_json::json!({ "role": "user", "content": "a".repeat(400) }),
            serde_json::json!({ "role": "assistant", "content": "a".repeat(400) }),
        ];
        for _ in 0..6 {
            messages.push(serde_json::json!({ "role": "user", "content": "hi" }));
        }
        let body = serde_json::json!({ "model": "claude-opus-4-8", "messages": messages });
        let decision = evaluate_would_have(&body).expect("a matching L-1 rule");

        let mut obs = obs();
        obs.transform = Some(decision);
        let usage = Usage {
            input_tokens: 10,
            ..Usage::default()
        };
        let data = assemble_data(&obs, &usage, CostBasis::ProviderReported, None, true);

        assert_eq!(data["ai.openlatch.transform.rule_id"], "OL-ECO-001");
        assert_eq!(data["ai.openlatch.transform.lever"], "history_trim");
        assert_eq!(data["ai.openlatch.transform.outcome"], "skipped_stage");
        assert_eq!(data["ai.openlatch.transform.ladder_stage"], "observe");
        assert_eq!(data["ai.openlatch.transform.rule_version"], 1);
        assert_eq!(data["ai.openlatch.transform.bundle_revision"], 0);
        assert_eq!(data["ai.openlatch.transform.write_multiplier"], 1.25);
        assert!(
            data["ai.openlatch.transform.tokens_gross"]
                .as_u64()
                .unwrap()
                > 0
        );
        assert!(data["ai.openlatch.transform.tokens_net"].as_f64().unwrap() > 0.0);
        // Never applied, never a non-observe stage (D-26).
        assert_ne!(data["ai.openlatch.transform.outcome"], "applied");
        // finding_id is not in the transform contract.
        assert!(data.get("ai.openlatch.transform.finding_id").is_none());
    }

    #[test]
    fn no_op_observation_emits_nothing() {
        let filter = PrivacyFilter::new(&[]);
        // measured=false → build_and_emit is a no-op even with a live sink.
        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
        build_and_emit(
            &Observation::none(),
            &Usage::default(),
            CostBasis::TokenizerEstimated,
            None,
            false,
            &filter,
            Some(&tx),
        );
        assert!(rx.try_recv().is_err(), "no event for an unmeasured request");
    }

    /// The `None` branch of `resolved_source()` has never run in a test:
    /// [`obs`] pins `source: Some("claude-code")`, which is the only fixture
    /// this module has.
    ///
    /// A fabricated `claude-code` row is wrong even when it happens to be
    /// right — it leaves no value meaning "we could not tell", so the
    /// platform's `(org, source, agent_id)` join can never be falsified.
    #[test]
    fn unresolved_source_is_unknown() {
        let mut obs = obs();
        obs.session.source = None;
        // Rung 2 explicitly unavailable — this test is about rung 3, and it
        // would otherwise pass or fail on whatever the fixture happened to set.
        obs.attributable_agent = None;

        let data = assemble_data(
            &obs,
            &Usage::default(),
            CostBasis::ProviderReported,
            None,
            true,
        );
        assert_eq!(
            data["ai.openlatch.session.source"], "unknown",
            "an unattributed request must say so, not name an agent"
        );

        let event = assemble_event(&obs, data, &PrivacyFilter::new(&[]));
        assert_eq!(
            event.envelope["source"], "unknown",
            "the CloudEvents `source` carries the same claim as the data field"
        );
    }

    /// Rung 2. The fact this host holds and used to discard.
    ///
    /// The turn that motivated the fix: no live hook session (its `SessionStart`
    /// had not landed yet), but it arrived on `/v1/messages`, reported
    /// `anthropic` / `claude-opus-5`, and ran where Claude Code was the only
    /// agent wired to speak that protocol. Reporting `unknown` beside those
    /// three was not caution.
    #[test]
    fn the_sole_wired_speaker_names_an_otherwise_unattributed_request() {
        let mut obs = obs();
        obs.session.source = None;
        obs.wire_format = WireFormat::AnthropicMessages;
        obs.attributable_agent = Some("claude-code");

        let data = assemble_data(
            &obs,
            &Usage::default(),
            CostBasis::ProviderReported,
            None,
            true,
        );
        assert_eq!(
            data["ai.openlatch.session.source"], "claude-code",
            "the only agent wired to this protocol on this install is the author"
        );

        let event = assemble_event(&obs, data, &PrivacyFilter::new(&[]));
        assert_eq!(
            event.envelope["source"], "claude-code",
            "the CloudEvents `source` carries the same claim as the data field"
        );
    }

    /// Rung 1 outranks rung 2, and the ordering is not cosmetic.
    ///
    /// A hook that OBSERVED the session beats an inference from our own config.
    /// Were this reversed, a second agent added to the host would start
    /// relabelling the first agent's fully-attributed turns.
    #[test]
    fn an_observed_session_outranks_the_wiring_deduction() {
        let mut obs = obs();
        obs.session.source = Some("codex-cli".to_string());
        obs.attributable_agent = Some("claude-code");

        let data = assemble_data(
            &obs,
            &Usage::default(),
            CostBasis::ProviderReported,
            None,
            true,
        );
        assert_eq!(
            data["ai.openlatch.session.source"], "codex-cli",
            "a hook that saw the session outranks an inference from our own wiring"
        );
    }
}