lean_ctx/core/finops_export/
cbf.rs

1//! CloudZero Common Bill Format (CBF) serializer + AnyCost Stream uploader.
2//!
3//! Spec pinned: CBF as documented at
4//! <https://docs.cloudzero.com/docs/anycost-common-bill-format-cbf> —
5//! required columns `time/usage_start` + `cost/cost`; savings are emitted as
6//! `lineitem/type=Discount` rows with negative cost (CBF's documented
7//! mechanism, included in CloudZero "Real Cost").
8//!
9//! Idempotency: the Stream API request carries `"operation": "replace_drop"`,
10//! which replaces all previously dropped data for the same month — re-running
11//! an export overwrites instead of duplicating (CloudZero-side guarantee).
12
13use super::{csv_field, DailyCostRow};
14
15pub const HEADER: &[&str] = &[
16    "lineitem/type",
17    "time/usage_start",
18    "resource/service",
19    "resource/id",
20    "resource/account",
21    "usage/amount",
22    "usage/units",
23    "cost/cost",
24    "resource/tag:project",
25    "resource/tag:agent_role",
26    "resource/tag:model",
27    "resource/tag:tool",
28];
29
30const SERVICE: &str = "LeanCTX";
31
32fn record(row: &DailyCostRow, line_type: &str, amount: u64, cost: f64) -> Vec<String> {
33    vec![
34        line_type.to_string(),
35        format!("{}T00:00:00Z", row.date),
36        SERVICE.to_string(),
37        format!("leanctx/{}/{}/{}", row.project, row.agent_role, row.model),
38        row.project.clone(),
39        amount.to_string(),
40        "tokens".to_string(),
41        format!("{cost:.6}"),
42        row.project.clone(),
43        row.agent_role.clone(),
44        row.model.clone(),
45        row.tool.clone(),
46    ]
47}
48
49fn records(rows: &[DailyCostRow]) -> Vec<Vec<String>> {
50    let mut out = Vec::new();
51    for row in rows {
52        out.push(record(row, "Usage", row.tokens_actual, row.cost_usd));
53        if row.tokens_saved > 0 {
54            out.push(record(row, "Discount", row.tokens_saved, -row.savings_usd));
55        }
56    }
57    out
58}
59
60/// CBF CSV (for AnyCost bucket drops or manual import).
61pub fn to_csv(rows: &[DailyCostRow]) -> String {
62    let mut out = HEADER.join(",");
63    out.push('\n');
64    for rec in records(rows) {
65        let line = rec
66            .iter()
67            .map(|f| csv_field(f))
68            .collect::<Vec<_>>()
69            .join(",");
70        out.push_str(&line);
71        out.push('\n');
72    }
73    out
74}
75
76/// AnyCost Stream request body for one month (`YYYY-MM`): all rows must
77/// belong to that month; `replace_drop` makes the upload idempotent.
78pub fn to_stream_body(rows: &[DailyCostRow], month: &str) -> serde_json::Value {
79    let data: Vec<serde_json::Value> = records(rows)
80        .into_iter()
81        .map(|rec| {
82            let mut obj = serde_json::Map::new();
83            for (key, val) in HEADER.iter().zip(rec) {
84                obj.insert((*key).to_string(), serde_json::Value::String(val));
85            }
86            serde_json::Value::Object(obj)
87        })
88        .collect();
89    serde_json::json!({
90        "month": month,
91        "operation": "replace_drop",
92        "data": data,
93    })
94}
95
96/// Upload one month to the AnyCost Stream API.
97///
98/// Credentials: `CLOUDZERO_API_KEY` (Authorization header) and
99/// `CLOUDZERO_CONNECTION_ID` (the AnyCost Stream connection).
100pub fn upload(rows: &[DailyCostRow], month: &str) -> Result<String, String> {
101    let api_key =
102        std::env::var("CLOUDZERO_API_KEY").map_err(|_| "CLOUDZERO_API_KEY not set".to_string())?;
103    let connection = std::env::var("CLOUDZERO_CONNECTION_ID")
104        .map_err(|_| "CLOUDZERO_CONNECTION_ID not set".to_string())?;
105    let url = format!(
106        "https://api.cloudzero.com/v2/connections/billing/anycost/{connection}/billing_drops"
107    );
108
109    let body = serde_json::to_vec(&to_stream_body(rows, month)).map_err(|e| e.to_string())?;
110    let agent = ureq::Agent::new_with_config(
111        ureq::config::Config::builder()
112            .timeout_global(Some(std::time::Duration::from_secs(30)))
113            .http_status_as_error(false)
114            .build(),
115    );
116    let resp = agent
117        .post(&url)
118        .header("Content-Type", "application/json")
119        .header("Authorization", api_key.trim())
120        .send(body.as_slice())
121        .map_err(|e| format!("cloudzero unreachable: {e}"))?;
122
123    let status = resp.status().as_u16();
124    let text = resp.into_body().read_to_string().unwrap_or_default();
125    if (200..300).contains(&status) {
126        Ok(format!("CloudZero accepted month {month} ({status})"))
127    } else {
128        Err(format!("CloudZero rejected ({status}): {text}"))
129    }
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135
136    fn row() -> DailyCostRow {
137        DailyCostRow {
138            date: "2026-06-01".into(),
139            project: "proj_a".into(),
140            agent_role: "coder".into(),
141            model: "claude".into(),
142            tool: "ctx_read".into(),
143            tokens_actual: 400,
144            tokens_saved: 1600,
145            cost_usd: 0.001,
146            savings_usd: 0.004,
147        }
148    }
149
150    #[test]
151    fn usage_and_discount_rows() {
152        let csv = to_csv(&[row()]);
153        let lines: Vec<&str> = csv.lines().collect();
154        assert_eq!(lines.len(), 3);
155        assert!(lines[1].starts_with("Usage,2026-06-01T00:00:00Z"));
156        assert!(lines[2].starts_with("Discount,"));
157        assert!(
158            lines[2].contains("-0.004"),
159            "discount negative: {}",
160            lines[2]
161        );
162    }
163
164    #[test]
165    fn stream_body_is_idempotent_replace_drop() {
166        let body = to_stream_body(&[row()], "2026-06");
167        assert_eq!(body["month"], "2026-06");
168        assert_eq!(body["operation"], "replace_drop");
169        let data = body["data"].as_array().unwrap();
170        assert_eq!(data.len(), 2);
171        assert_eq!(data[0]["lineitem/type"], "Usage");
172        assert_eq!(data[0]["cost/cost"], "0.001000");
173        assert_eq!(data[1]["resource/tag:project"], "proj_a");
174    }
175
176    #[test]
177    fn required_cbf_columns_present() {
178        let body = to_stream_body(&[row()], "2026-06");
179        let first = &body["data"][0];
180        assert!(first.get("time/usage_start").is_some());
181        assert!(first.get("cost/cost").is_some());
182    }
183}
lean_ctx/core/finops_export/cbf.rs

lean_ctx/core/finops_export/
cbf.rs
