openlatch-client 0.3.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>,
    /// The wire format this request spoke, resolved from its route.
    ///
    /// `gen_ai.provider.name` is read from it rather than from a constant: the
    /// boundary 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,
            // 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.
    /// 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())
    }
}

/// 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 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": obs.model.clone().unwrap_or_default(),
        "gen_ai.provider.name": obs.wire_format.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::boundary::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 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 boundary'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,
            "boundary: cloud channel full — economics event dropped"
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::boundary::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,
            // 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,
        }
    }

    /// 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");
    }

    #[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::boundary::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"
        );
    }
}