openlatch-client 0.1.18

//! 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};

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

/// The model provider in this PRD.
pub const PROVIDER: &str = "anthropic";

/// Phase-1 single-source default (the agent platform). Emitted as
/// `ai.openlatch.session.source`.
///
/// ⚠️ This is the **hook wire value** (`claude-code`, hyphen) — the same string
/// the Mode 1 hook stamps on `source` — so the platform attribution join on
/// `(organization_id, source, agent_id)` matches the hook stream. The PRD prose
/// writes `claude_code` (underscore); the hyphen form is what the generated
/// `AgentType` actually serialises and therefore what the join requires.
pub const DEFAULT_SOURCE: &str = "claude-code";

/// 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>,
}

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,
        }
    }

    /// 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): the resolved session source, falling back
    /// to the phase-1 default agent platform ([`DEFAULT_SOURCE`]).
    fn resolved_source(&self) -> String {
        self.session
            .source
            .clone()
            .unwrap_or_else(|| DEFAULT_SOURCE.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": 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,
        "ai.openlatch.session.assurance": 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 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,
    });

    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,
        }
    }

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