kaizen/sync/
outbound.rs

1// SPDX-License-Identifier: AGPL-3.0-or-later
2//! Typed outbound JSON matching `POST /v1/events` (single-event shape used in outbox rows).
3
4use crate::core::event::{Event, EventKind, EventSource, SessionRecord};
5use blake3::Hasher;
6use serde::{Deserialize, Serialize};
7use std::path::Path;
8
9const BLAKE3_PREFIX: &str = "blake3:";
10
11/// Full batch body for `POST /v1/events`.
12#[derive(Debug, Clone, Serialize, Deserialize)]
13pub struct EventsBatchBody {
14    pub team_id: String,
15    pub workspace_hash: String,
16    pub events: Vec<OutboundEvent>,
17}
18
19/// One event in the ingest API shape (after redaction).
20#[derive(Debug, Clone, Serialize, Deserialize)]
21pub struct OutboundEvent {
22    pub session_id_hash: String,
23    pub event_seq: u64,
24    pub ts_ms: u64,
25    pub agent: String,
26    pub model: String,
27    pub kind: String,
28    pub source: String,
29    #[serde(skip_serializing_if = "Option::is_none")]
30    pub tool: Option<String>,
31    #[serde(skip_serializing_if = "Option::is_none")]
32    pub tool_call_id: Option<String>,
33    #[serde(skip_serializing_if = "Option::is_none")]
34    pub tokens_in: Option<u32>,
35    #[serde(skip_serializing_if = "Option::is_none")]
36    pub tokens_out: Option<u32>,
37    #[serde(skip_serializing_if = "Option::is_none")]
38    pub reasoning_tokens: Option<u32>,
39    #[serde(skip_serializing_if = "Option::is_none")]
40    pub cost_usd_e6: Option<i64>,
41    pub payload: serde_json::Value,
42}
43
44pub fn hash_with_salt(team_salt: &[u8; 32], material: &[u8]) -> String {
45    let mut h = Hasher::new();
46    h.update(team_salt);
47    h.update(material);
48    format!("{BLAKE3_PREFIX}{}", hex::encode(h.finalize().as_bytes()))
49}
50
51pub fn workspace_hash(team_salt: &[u8; 32], workspace_abs: &Path) -> String {
52    let normalized = workspace_abs.to_string_lossy();
53    hash_with_salt(team_salt, normalized.as_bytes())
54}
55
56pub fn outbound_event_from_row(
57    e: &Event,
58    session: &SessionRecord,
59    team_salt: &[u8; 32],
60) -> OutboundEvent {
61    OutboundEvent {
62        session_id_hash: hash_with_salt(team_salt, e.session_id.as_bytes()),
63        event_seq: e.seq,
64        ts_ms: e.ts_ms,
65        agent: session.agent.clone(),
66        model: session
67            .model
68            .clone()
69            .unwrap_or_else(|| "unknown".to_string()),
70        kind: kind_api(&e.kind),
71        source: source_api(&e.source),
72        tool: e.tool.clone(),
73        tool_call_id: e.tool_call_id.clone(),
74        tokens_in: e.tokens_in,
75        tokens_out: e.tokens_out,
76        reasoning_tokens: e.reasoning_tokens,
77        cost_usd_e6: e.cost_usd_e6,
78        payload: e.payload.clone(),
79    }
80}
81
82fn kind_api(k: &EventKind) -> String {
83    match k {
84        EventKind::ToolCall => "tool_call",
85        EventKind::ToolResult => "tool_result",
86        EventKind::Message => "message",
87        EventKind::Error => "error",
88        EventKind::Cost => "cost",
89        EventKind::Hook => "hook",
90        EventKind::Lifecycle => "lifecycle",
91    }
92    .to_string()
93}
94
95fn source_api(s: &EventSource) -> String {
96    match s {
97        EventSource::Tail => "tail",
98        EventSource::Hook => "hook",
99        EventSource::Proxy => "proxy",
100    }
101    .to_string()
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107    use serde_json::json;
108
109    #[test]
110    fn workspace_hash_stable_for_same_salt_and_path() {
111        let salt = [7u8; 32];
112        let p = Path::new("/tmp/ws");
113        let a = workspace_hash(&salt, p);
114        let b = workspace_hash(&salt, p);
115        assert_eq!(a, b);
116        assert!(a.starts_with(BLAKE3_PREFIX));
117    }
118
119    #[test]
120    fn outbound_maps_kind_snake_case() {
121        let salt = [0u8; 32];
122        let session = SessionRecord {
123            id: "sid".into(),
124            agent: "cursor".into(),
125            model: Some("m1".into()),
126            workspace: "/w".into(),
127            started_at_ms: 0,
128            ended_at_ms: None,
129            status: crate::core::event::SessionStatus::Running,
130            trace_path: "".into(),
131            start_commit: None,
132            end_commit: None,
133            branch: None,
134            dirty_start: None,
135            dirty_end: None,
136            repo_binding_source: None,
137            prompt_fingerprint: None,
138            parent_session_id: None,
139            agent_version: None,
140            os: None,
141            arch: None,
142            repo_file_count: None,
143            repo_total_loc: None,
144        };
145        let ev = Event {
146            session_id: "sid".into(),
147            seq: 3,
148            ts_ms: 99,
149            ts_exact: false,
150            kind: EventKind::ToolCall,
151            source: EventSource::Hook,
152            tool: Some("Edit".into()),
153            tool_call_id: Some("call_1".into()),
154            tokens_in: None,
155            tokens_out: None,
156            reasoning_tokens: None,
157            cost_usd_e6: None,
158            stop_reason: None,
159            latency_ms: None,
160            ttft_ms: None,
161            retry_count: None,
162            context_used_tokens: None,
163            context_max_tokens: None,
164            cache_creation_tokens: None,
165            cache_read_tokens: None,
166            system_prompt_tokens: None,
167            payload: json!({}),
168        };
169        let o = outbound_event_from_row(&ev, &session, &salt);
170        assert_eq!(o.kind, "tool_call");
171        assert_eq!(o.source, "hook");
172        assert_eq!(o.event_seq, 3);
173    }
174}
kaizen/sync/outbound.rs

kaizen/sync/
outbound.rs
