synapse/ledger/
event.rs

1//! talos-aligned published usage event (camelCase JSON). Decouples the wire
2//! format from the DB-shaped `UsageEntry`. Mirrors the conventions of
3//! `talos-core/src/tool_events.rs` (`ToolEvent`): `rename_all = "camelCase"`,
4//! tenancy as `namespace`, a `type` discriminator, RFC3339 timestamp.
5
6use chrono::{DateTime, Utc};
7use serde::Serialize;
8
9use crate::ledger::UsageEntry;
10
11/// Marketplace ledger subscription discriminator (`attributes.EventType`).
12pub const LEDGER_EVENT_TYPE: &str = "Ledger.LLMTokensConsumed";
13
14/// One terminal usage event, published to a topic per request.
15#[derive(Debug, Clone, Serialize)]
16#[serde(rename_all = "camelCase")]
17pub struct UsageEvent {
18    /// Tenancy. talos calls this `namespace`; synapse maps its `tenant` onto it.
19    pub namespace: String,
20    #[serde(skip_serializing_if = "Option::is_none")]
21    pub workspace: Option<String>,
22    pub request_id: String,
23    pub timestamp: DateTime<Utc>,
24    #[serde(rename = "type")]
25    pub event_type: &'static str,
26    pub route: String,
27    pub provider: String,
28    pub model: String,
29    pub lane: String,
30    pub input_tokens: u64,
31    pub output_tokens: u64,
32    pub cost_usd: f64,
33    pub status: String,
34    /// Lane discriminator: "chat" or "embedding".
35    pub op: String,
36}
37
38impl From<&UsageEntry> for UsageEvent {
39    fn from(e: &UsageEntry) -> Self {
40        Self {
41            namespace: e.tenant.clone(),
42            workspace: e.workspace.clone(),
43            request_id: e.request_id.clone(),
44            timestamp: e.ts,
45            event_type: "usage",
46            route: e.route.clone(),
47            provider: e.provider.clone(),
48            model: e.model.clone(),
49            lane: e.lane.clone(),
50            input_tokens: e.input_tokens,
51            output_tokens: e.output_tokens,
52            cost_usd: e.cost_usd,
53            status: e.status.clone(),
54            op: e.op.clone(),
55        }
56    }
57}
58
59impl UsageEvent {
60    /// Broker message attributes for subscription filtering (talos-aligned keys
61    /// plus synapse-useful `provider`/`status`). Returned as ordered pairs so
62    /// both backends can convert to their SDK-specific attribute type.
63    pub fn attributes(&self) -> Vec<(&'static str, String)> {
64        vec![
65            ("EventType", LEDGER_EVENT_TYPE.to_string()),
66            ("namespace", self.namespace.clone()),
67            ("requestId", self.request_id.clone()),
68            ("type", self.event_type.to_string()),
69            ("provider", self.provider.clone()),
70            ("status", self.status.clone()),
71        ]
72    }
73}
74
75#[cfg(test)]
76mod tests {
77    use super::*;
78    use chrono::TimeZone;
79
80    fn entry() -> UsageEntry {
81        UsageEntry {
82            ts: Utc.with_ymd_and_hms(2026, 6, 10, 15, 30, 45).unwrap(),
83            tenant: "acme".into(),
84            workspace: None,
85            route: "gemini-pro".into(),
86            provider: "vertex".into(),
87            model: "gemini-3-pro".into(),
88            lane: "standard".into(),
89            input_tokens: 3,
90            output_tokens: 5,
91            cost_usd: 0.001,
92            request_id: "req-1".into(),
93            status: "ok".into(),
94            op: "chat".into(),
95        }
96    }
97
98    #[test]
99    fn serializes_talos_aligned_camelcase() {
100        let v = serde_json::to_value(UsageEvent::from(&entry())).unwrap();
101        assert_eq!(v["namespace"], "acme");
102        assert_eq!(v["type"], "usage");
103        assert_eq!(v["requestId"], "req-1");
104        assert_eq!(v["inputTokens"], 3);
105        assert_eq!(v["outputTokens"], 5);
106        assert_eq!(v["costUsd"], 0.001);
107        assert_eq!(v["lane"], "standard");
108        assert_eq!(v["op"], "chat");
109        assert!(v.get("workspace").is_none());
110        assert!(v.get("request_id").is_none());
111        assert!(v.get("input_tokens").is_none());
112    }
113
114    #[test]
115    fn serializes_embedding_op() {
116        let mut e = entry();
117        e.op = "embedding".into();
118        let v = serde_json::to_value(UsageEvent::from(&e)).unwrap();
119        assert_eq!(v["op"], "embedding");
120    }
121
122    #[test]
123    fn attributes_have_talos_keys() {
124        let attrs = UsageEvent::from(&entry()).attributes();
125        let keys: Vec<&str> = attrs.iter().map(|(k, _)| *k).collect();
126        assert_eq!(
127            keys,
128            vec![
129                "EventType",
130                "namespace",
131                "requestId",
132                "type",
133                "provider",
134                "status"
135            ]
136        );
137        assert_eq!(attrs[0].1, LEDGER_EVENT_TYPE);
138        assert_eq!(attrs[1].1, "acme");
139    }
140}
synapse/ledger/event.rs

synapse/ledger/
event.rs
