lean_ctx_protocol/hive/
telemetry.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Cockpit metrics and receipts.
4//!
5//! - `POST /v1/metrics` [`MetricsUploadV1`] (`leanctx.cloud-metrics/v1`,
6//!   unchanged from the cloud cockpit) → [`MetricsAckV1`]. Idempotent replace
7//!   per `(device.id, date, vault.id)`; `vault.id` is the resolved Hive
8//!   `vault_id`.
9//! - 1.9 (X4): a day may carry `hive` ([`MetricsHiveV1`]), how Hive
10//!   helped in that vault: counts and durations only, the same for sealed
11//!   vaults. Claim conflicts are not reported; Hive derives them from its
12//!   own refusals and bus lease conflicts.
13//! - `POST /v1/receipts` [`ReceiptsUploadV1`] → [`ReceiptsAckV1`]; a receipt
14//!   already stored (same `receipt_id` and `digest`) counts as duplicate.
15
16use serde::{Deserialize, Serialize};
17use serde_json::Value;
18
19use super::{MAX_BODY_BYTES, require_schema, validate_id};
20use crate::{Sha256Digest, UtcTimestamp, ValidationError};
21
22pub const METRICS_SCHEMA: &str = "leanctx.cloud-metrics/v1";
23pub const METRICS_ACK_SCHEMA: &str = "leanctx.hive.metrics-ack/v1";
24pub const RECEIPTS_SCHEMA: &str = "leanctx.hive.receipts/v1";
25pub const RECEIPTS_ACK_SCHEMA: &str = "leanctx.hive.receipts-ack/v1";
26
27pub const MAX_METRIC_DAYS: usize = 62;
28pub const MAX_METRIC_SPEND_ROWS: usize = 32;
29pub const MAX_METRIC_AGENT_ROWS: usize = 16;
30pub const MAX_RECEIPTS: usize = 256;
31
32#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
33#[serde(deny_unknown_fields)]
34pub struct MetricsUploadV1 {
35    pub schema: String,
36    pub device: MetricsDeviceV1,
37    pub days: Vec<MetricsDayV1>,
38}
39
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
41#[serde(deny_unknown_fields)]
42pub struct MetricsDeviceV1 {
43    pub id: String,
44    pub label: Option<String>,
45    pub engine_version: String,
46    pub os: String,
47}
48
49#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
50#[serde(deny_unknown_fields)]
51pub struct MetricsDayV1 {
52    /// `YYYY-MM-DD` (UTC).
53    pub date: String,
54    pub vault: MetricsVaultV1,
55    pub savings: MetricsSavingsV1,
56    pub spend: Vec<MetricsSpendV1>,
57    pub ledger: MetricsLedgerV1,
58    pub guards: MetricsGuardsV1,
59    pub agents: Vec<MetricsAgentV1>,
60    pub memory: MetricsMemoryV1,
61    pub receipts: u64,
62    /// 1.9 (X4).
63    #[serde(default, skip_serializing_if = "Option::is_none")]
64    pub hive: Option<MetricsHiveV1>,
65}
66
67pub const MAX_METRIC_TASK_ROWS: usize = 64;
68/// 1.9 R3: ceilings of [`MetricsHiveV1`] (counts per day, the median in
69/// milliseconds up to one day, tokens and USD micros per task row); the
70/// names match the Hive server's.
71pub const MAX_METRIC_DAY_COUNT: u64 = 1_000_000;
72pub const MAX_METRIC_DAY_MS: u64 = 86_400_000;
73pub const MAX_METRIC_TASK_TOKENS: u64 = 10_000_000_000;
74pub const MAX_METRIC_TASK_USD_MICROS: u64 = 10_000_000_000;
75
76/// 1.9 (X4): one vault day of Hive use.
77#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
78#[serde(deny_unknown_fields)]
79pub struct MetricsHiveV1 {
80    /// Hive facts delivered in a session and then used in it (M1).
81    pub knowledge_hits_referenced: u64,
82    /// Ledger usage while a session held a task claim, per task (M2);
83    /// sorted by `number`, at most one row per task.
84    #[serde(default)]
85    pub task_usage: Vec<MetricsTaskUsageV1>,
86    /// Median time from session start to the first Hive context delivered,
87    /// over `first_context_sessions` sessions (M5); absent with none.
88    #[serde(default, skip_serializing_if = "Option::is_none")]
89    pub first_context_ms_p50: Option<u64>,
90    pub first_context_sessions: u64,
91}
92
93#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
94#[serde(deny_unknown_fields)]
95pub struct MetricsTaskUsageV1 {
96    pub number: u64,
97    pub input_tokens: u64,
98    pub output_tokens: u64,
99    /// List price, not an invoice amount.
100    pub list_usd_micros: u64,
101}
102
103impl MetricsHiveV1 {
104    pub fn validate(&self) -> Result<(), ValidationError> {
105        if self.task_usage.len() > MAX_METRIC_TASK_ROWS
106            || self.task_usage.iter().any(|row| row.number == 0)
107            || self
108                .task_usage
109                .windows(2)
110                .any(|pair| pair[0].number >= pair[1].number)
111        {
112            return Err(ValidationError::new(
113                "task_usage holds at most 64 rows sorted by task number without repeats",
114            ));
115        }
116        if self.first_context_ms_p50.is_some() != (self.first_context_sessions > 0) {
117            return Err(ValidationError::new(
118                "first_context_ms_p50 comes exactly with first_context_sessions",
119            ));
120        }
121        // R3: ceilings no real day reaches, so a day cannot skew the rollup.
122        if self.knowledge_hits_referenced > MAX_METRIC_DAY_COUNT
123            || self.first_context_sessions > MAX_METRIC_DAY_COUNT
124            || self
125                .first_context_ms_p50
126                .is_some_and(|ms| ms > MAX_METRIC_DAY_MS)
127            || self.task_usage.iter().any(|row| {
128                row.input_tokens > MAX_METRIC_TASK_TOKENS
129                    || row.output_tokens > MAX_METRIC_TASK_TOKENS
130                    || row.list_usd_micros > MAX_METRIC_TASK_USD_MICROS
131            })
132        {
133            return Err(ValidationError::new(
134                "hive day values exceed their ceilings",
135            ));
136        }
137        Ok(())
138    }
139}
140
141#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
142#[serde(deny_unknown_fields)]
143pub struct MetricsVaultV1 {
144    pub id: String,
145    pub label: Option<String>,
146}
147
148#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
149#[serde(deny_unknown_fields)]
150pub struct MetricsSavingsV1 {
151    pub tokens_requested: u64,
152    pub tokens_delivered: u64,
153    pub tokens_saved: u64,
154    pub tool_calls: u64,
155    pub cache_hits: u64,
156}
157
158#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
159#[serde(deny_unknown_fields)]
160pub struct MetricsSpendV1 {
161    pub model: String,
162    pub provider: String,
163    pub requests: u64,
164    pub input_tokens: u64,
165    pub output_tokens: u64,
166    pub cached_tokens: u64,
167    /// List-price USD, not an invoice amount.
168    pub list_usd: f64,
169}
170
171#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
172#[serde(deny_unknown_fields)]
173pub struct MetricsLedgerV1 {
174    pub reduction: u64,
175    pub edit: u64,
176    pub correction: u64,
177    pub cache_discount: u64,
178    pub other: u64,
179}
180
181#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
182#[serde(deny_unknown_fields)]
183pub struct MetricsGuardsV1 {
184    pub secrets_redacted: u64,
185    pub pii_redacted: u64,
186    pub injection_flags: u64,
187    pub content_withheld: u64,
188    pub paths_blocked: u64,
189    pub shell_blocked: u64,
190    pub partial_inspections: u64,
191}
192
193#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
194#[serde(deny_unknown_fields)]
195pub struct MetricsAgentV1 {
196    pub host: String,
197    pub sessions: u64,
198    pub tool_calls: u64,
199}
200
201#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
202#[serde(deny_unknown_fields)]
203pub struct MetricsMemoryV1 {
204    pub facts_current: u64,
205    pub facts_added: u64,
206    pub lessons_linked: u64,
207}
208
209impl MetricsUploadV1 {
210    /// Shape and size bounds. The per-field rules of the cockpit metrics
211    /// contract (§3: id/label/version syntax, date window, count ceilings)
212    /// stay with the engine's metrics builder, which never sends a body the
213    /// service would refuse.
214    pub fn validate(&self) -> Result<(), ValidationError> {
215        require_schema(&self.schema, METRICS_SCHEMA)?;
216        if self.days.is_empty() || self.days.len() > MAX_METRIC_DAYS {
217            return Err(ValidationError::new("days must hold 1-62 entries"));
218        }
219        let mut seen = std::collections::HashSet::new();
220        for day in &self.days {
221            if day.spend.len() > MAX_METRIC_SPEND_ROWS || day.agents.len() > MAX_METRIC_AGENT_ROWS {
222                return Err(ValidationError::new("too many spend or agent rows"));
223            }
224            if !seen.insert((day.date.as_str(), day.vault.id.as_str())) {
225                return Err(ValidationError::new("a (date, vault) pair repeats"));
226            }
227            if let Some(hive) = &day.hive {
228                hive.validate()?;
229            }
230        }
231        let size = serde_json::to_vec(self)
232            .map_err(|error| ValidationError::new(format!("metrics: {error}")))?
233            .len();
234        if size > MAX_BODY_BYTES {
235            return Err(ValidationError::new("metrics body exceeds 1 MiB"));
236        }
237        Ok(())
238    }
239}
240
241/// Response of `POST /v1/metrics`.
242#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
243pub struct MetricsAckV1 {
244    pub schema: String,
245    /// Day entries stored (inserted or replaced).
246    pub accepted: u32,
247}
248
249/// One receipt. `document` is the receipt as persisted locally.
250#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
251#[serde(deny_unknown_fields)]
252pub struct ReceiptItemV1 {
253    pub receipt_id: String,
254    /// `engine` or `agent_connector`.
255    pub kind: String,
256    pub vault_id: String,
257    pub created_at: UtcTimestamp,
258    /// [`receipt_digest`] of `document`.
259    pub digest: Sha256Digest,
260    pub document: Value,
261}
262
263/// `sha256:<hex>` of `document` with object keys sorted and no whitespace
264/// (numbers as `serde_json` prints them; receipts are not signed here).
265pub fn receipt_digest(document: &Value) -> Result<Sha256Digest, ValidationError> {
266    use sha2::{Digest, Sha256};
267    use std::fmt::Write as _;
268    fn sorted(value: &Value) -> Value {
269        match value {
270            Value::Object(fields) => Value::Object(
271                fields
272                    .iter()
273                    .map(|(key, value)| (key.clone(), sorted(value)))
274                    .collect::<std::collections::BTreeMap<_, _>>()
275                    .into_iter()
276                    .collect(),
277            ),
278            Value::Array(values) => Value::Array(values.iter().map(sorted).collect()),
279            scalar => scalar.clone(),
280        }
281    }
282    let bytes = serde_json::to_vec(&sorted(document))
283        .map_err(|error| ValidationError::new(format!("receipt: {error}")))?;
284    let mut digest = String::from("sha256:");
285    for byte in Sha256::digest(bytes) {
286        write!(&mut digest, "{byte:02x}").expect("writing to String cannot fail");
287    }
288    Sha256Digest::new(digest)
289}
290
291/// `POST /v1/receipts`.
292#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
293#[serde(deny_unknown_fields)]
294pub struct ReceiptsUploadV1 {
295    pub schema: String,
296    pub receipts: Vec<ReceiptItemV1>,
297}
298
299impl ReceiptsUploadV1 {
300    pub fn validate(&self) -> Result<(), ValidationError> {
301        require_schema(&self.schema, RECEIPTS_SCHEMA)?;
302        if self.receipts.is_empty() || self.receipts.len() > MAX_RECEIPTS {
303            return Err(ValidationError::new("receipts must hold 1-256 items"));
304        }
305        for item in &self.receipts {
306            validate_id(&item.receipt_id, "receipt_id")?;
307            super::validate_vault_id(&item.vault_id)?;
308            if !matches!(item.kind.as_str(), "engine" | "agent_connector") {
309                return Err(ValidationError::new(
310                    "receipt kind must be engine or agent_connector",
311                ));
312            }
313            if !item.document.is_object() {
314                return Err(ValidationError::new("receipt document must be an object"));
315            }
316            if receipt_digest(&item.document)? != item.digest {
317                return Err(ValidationError::new(
318                    "receipt digest does not match document",
319                ));
320            }
321        }
322        let size = serde_json::to_vec(self)
323            .map_err(|error| ValidationError::new(format!("receipts: {error}")))?
324            .len();
325        if size > MAX_BODY_BYTES {
326            return Err(ValidationError::new("receipts body exceeds 1 MiB"));
327        }
328        Ok(())
329    }
330}
331
332/// Response of `POST /v1/receipts`.
333#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
334pub struct ReceiptsAckV1 {
335    pub schema: String,
336    pub accepted: u32,
337    pub duplicates: u32,
338}
lean_ctx_protocol/hive/telemetry.rs

lean_ctx_protocol/hive/
telemetry.rs
