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

lean_ctx_protocol/hive/
telemetry.rs
