rightkit_agent_engine_protocol/event/
envelope.rs

1use super::ProtocolEvent;
2use crate::TranscriptMessage;
3use serde::{Deserialize, Serialize};
4
5pub const EVENT_ENVELOPE_SCHEMA_V1: u16 = 1;
6
7/// Optional causal/cross-runtime correlation carried beside a canonical event.
8/// These fields identify relationships only; session/sequence/kind remain owned
9/// by `ProtocolEvent` and SQLite rather than being duplicated in the envelope.
10#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
11#[serde(rename_all = "camelCase")]
12pub struct EventCorrelationV1 {
13    #[serde(default, skip_serializing_if = "Option::is_none")]
14    pub run_id: Option<String>,
15    #[serde(default, skip_serializing_if = "Option::is_none")]
16    pub turn_id: Option<String>,
17    #[serde(default, skip_serializing_if = "Option::is_none")]
18    pub task_id: Option<String>,
19    #[serde(default, skip_serializing_if = "Option::is_none")]
20    pub tool_call_id: Option<String>,
21    #[serde(default, skip_serializing_if = "Option::is_none")]
22    pub attempt_id: Option<String>,
23    #[serde(default, skip_serializing_if = "Option::is_none")]
24    pub causal_parent_event_id: Option<String>,
25}
26
27impl EventCorrelationV1 {
28    /// Derive the normalized correlation dimensions already carried by the
29    /// canonical event. Routing identity (`session_id`, `seq`, kind) remains on
30    /// `ProtocolEvent`; this only exposes cross-event relationships without
31    /// duplicating the event payload.
32    pub fn from_event(event: &ProtocolEvent) -> Self {
33        let mut correlation = Self::default();
34        match event {
35            ProtocolEvent::SessionTurnFinished { turn_id, .. } => {
36                correlation.turn_id = Some(turn_id.clone());
37            }
38            ProtocolEvent::ToolCallStarted { call, .. }
39            | ProtocolEvent::ToolApprovalRequested { call, .. } => {
40                correlation.tool_call_id = Some(call.id.clone());
41            }
42            ProtocolEvent::ToolApprovalResolved { tool_call_id, .. }
43            | ProtocolEvent::ToolApprovalReviewed { tool_call_id, .. } => {
44                correlation.tool_call_id = Some(tool_call_id.clone());
45            }
46            ProtocolEvent::CredentialRequested { tool_call_id, .. } => {
47                correlation.tool_call_id = Some(tool_call_id.clone());
48            }
49            ProtocolEvent::CredentialResolved { tool_call_id, .. } => {
50                correlation.tool_call_id = Some(tool_call_id.clone());
51            }
52            ProtocolEvent::ToolResultAdded {
53                result: TranscriptMessage::Tool { tool_call_id, .. },
54                ..
55            } => {
56                correlation.tool_call_id = Some(tool_call_id.clone());
57            }
58            ProtocolEvent::TaskStatusUpdated { task_id, .. }
59            | ProtocolEvent::TaskArtifactUpdated { task_id, .. }
60            | ProtocolEvent::ArtifactLineageUpdated { task_id, .. } => {
61                correlation.task_id = Some(task_id.clone());
62            }
63            ProtocolEvent::SessionContextFolded { subtask_id, .. } => {
64                correlation.task_id = Some(subtask_id.clone());
65            }
66            ProtocolEvent::RunGovernanceRejected {
67                tool_call_id,
68                budget,
69                ownership,
70                origin,
71                ..
72            } => {
73                correlation.tool_call_id = Some(tool_call_id.clone());
74                correlation.run_id = budget
75                    .as_ref()
76                    .map(|value| value.run_id.clone())
77                    .or_else(|| origin.as_ref().map(|value| value.run_id.clone()))
78                    .or_else(|| ownership.as_ref().map(|value| value.target_run_id.clone()));
79            }
80            _ => {}
81        }
82        correlation
83    }
84}
85
86/// Stable metadata wrapper for one durable the host app event.
87///
88/// `event_id` is deterministically derived from the already-authoritative
89/// `(session_id, seq)` pair, so replay/reopen produces the same identity. The
90/// event payload remains unchanged and therefore backward-compatible.
91#[derive(Debug, Clone, Serialize, Deserialize)]
92#[serde(rename_all = "camelCase")]
93pub struct EventEnvelopeV1 {
94    pub schema_version: u16,
95    pub event_id: String,
96    #[serde(default)]
97    pub correlation: EventCorrelationV1,
98    pub event: ProtocolEvent,
99}
100
101impl EventEnvelopeV1 {
102    /// Build the canonical normalized representation for a durable fact using
103    /// correlation already present on the event. Because `event_id` derives
104    /// from `(session_id, seq)`, replay/reopen reconstructs the same identity
105    /// without a duplicate database column that could drift.
106    pub fn canonical(event: ProtocolEvent) -> Result<Self, String> {
107        let correlation = EventCorrelationV1::from_event(&event);
108        Self::for_durable_event(event, correlation)
109    }
110
111    pub fn for_durable_event(
112        event: ProtocolEvent,
113        correlation: EventCorrelationV1,
114    ) -> Result<Self, String> {
115        if !event.is_durable_fact() {
116            return Err("live projection events do not receive durable event envelopes".into());
117        }
118        let meta = event.meta();
119        if meta.session_id.trim().is_empty() {
120            return Err("durable event envelope requires session_id".into());
121        }
122        if meta.seq < 1 {
123            return Err("durable event envelope requires committed sequence >= 1".into());
124        }
125        Ok(Self {
126            schema_version: EVENT_ENVELOPE_SCHEMA_V1,
127            event_id: stable_event_id(&meta.session_id, meta.seq),
128            correlation,
129            event,
130        })
131    }
132}
133
134pub fn stable_event_id(session_id: &str, seq: i64) -> String {
135    format!("evt-v1:{session_id}:{seq}")
136}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141    use crate::TranscriptMessage;
142
143    fn durable(seq: i64) -> ProtocolEvent {
144        ProtocolEvent::AssistantMessageCompleted {
145            session_id: "session-a".into(),
146            seq,
147            created_at: "t".into(),
148            message: TranscriptMessage::Assistant {
149                id: "m".into(),
150                created_at: "t".into(),
151                content: "done".into(),
152                reasoning_content: None,
153                tool_calls: Vec::new(),
154                pinned: false,
155            },
156        }
157    }
158
159    #[test]
160    fn replay_identity_is_stable_for_same_session_sequence() {
161        let first =
162            EventEnvelopeV1::for_durable_event(durable(7), EventCorrelationV1::default()).unwrap();
163        let second =
164            EventEnvelopeV1::for_durable_event(durable(7), EventCorrelationV1::default()).unwrap();
165        assert_eq!(first.event_id, second.event_id);
166        assert_eq!(first.event_id, "evt-v1:session-a:7");
167        assert_eq!(first.schema_version, EVENT_ENVELOPE_SCHEMA_V1);
168    }
169
170    #[test]
171    fn correlation_is_additive_and_does_not_replace_event_routing() {
172        let envelope = EventEnvelopeV1::for_durable_event(
173            durable(9),
174            EventCorrelationV1 {
175                run_id: Some("run-1".into()),
176                turn_id: Some("turn-1".into()),
177                tool_call_id: Some("tool-1".into()),
178                ..EventCorrelationV1::default()
179            },
180        )
181        .unwrap();
182        assert_eq!(envelope.correlation.run_id.as_deref(), Some("run-1"));
183        assert_eq!(envelope.event.meta().seq, 9);
184        assert_eq!(envelope.event.meta().session_id, "session-a");
185    }
186
187    #[test]
188    fn canonical_extracts_turn_task_and_tool_correlation() {
189        let turn = ProtocolEvent::SessionTurnFinished {
190            session_id: "session-a".into(),
191            seq: 10,
192            created_at: "t".into(),
193            turn_id: "turn-10".into(),
194            status: "completed".into(),
195        };
196        let envelope = EventEnvelopeV1::canonical(turn).unwrap();
197        assert_eq!(envelope.correlation.turn_id.as_deref(), Some("turn-10"));
198
199        let tool = ProtocolEvent::ToolCallStarted {
200            session_id: "session-a".into(),
201            seq: 11,
202            created_at: "t".into(),
203            call: crate::ToolCall {
204                id: "tool-11".into(),
205                name: "read_file".into(),
206                arguments_json: "{}".into(),
207            },
208        };
209        let envelope = EventEnvelopeV1::canonical(tool).unwrap();
210        assert_eq!(
211            envelope.correlation.tool_call_id.as_deref(),
212            Some("tool-11")
213        );
214
215        let task = ProtocolEvent::TaskStatusUpdated {
216            session_id: "session-a".into(),
217            seq: 12,
218            created_at: "t".into(),
219            task_id: "task-12".into(),
220            context_id: "ctx".into(),
221            state: "running".into(),
222            task: serde_json::json!({}),
223        };
224        let envelope = EventEnvelopeV1::canonical(task).unwrap();
225        assert_eq!(envelope.correlation.task_id.as_deref(), Some("task-12"));
226    }
227
228    #[test]
229    fn canonical_extracts_tool_result_and_run_governance_correlation() {
230        let result = ProtocolEvent::ToolResultAdded {
231            session_id: "session-a".into(),
232            seq: 13,
233            created_at: "t".into(),
234            result: TranscriptMessage::Tool {
235                id: "result".into(),
236                created_at: "t".into(),
237                tool_call_id: "tool-13".into(),
238                summary: "ok".into(),
239                output: None,
240                trust: None,
241                is_error: false,
242                pinned: false,
243                process_id: None,
244            },
245        };
246        let envelope = EventEnvelopeV1::canonical(result).unwrap();
247        assert_eq!(
248            envelope.correlation.tool_call_id.as_deref(),
249            Some("tool-13")
250        );
251
252        let rejected = ProtocolEvent::RunGovernanceRejected {
253            session_id: "session-a".into(),
254            seq: 14,
255            created_at: "t".into(),
256            kind: crate::event::RunFailureKind::OriginMissing,
257            tool_call_id: "tool-14".into(),
258            budget: None,
259            ownership: None,
260            origin: Some(crate::event::RunOriginFailure {
261                run_id: "run-14".into(),
262                tool_call_id: "tool-14".into(),
263                reason: "missing".into(),
264            }),
265        };
266        let envelope = EventEnvelopeV1::canonical(rejected).unwrap();
267        assert_eq!(envelope.correlation.run_id.as_deref(), Some("run-14"));
268        assert_eq!(
269            envelope.correlation.tool_call_id.as_deref(),
270            Some("tool-14")
271        );
272    }
273
274    #[test]
275    fn live_projection_is_rejected() {
276        let delta = ProtocolEvent::AssistantDelta {
277            session_id: "session-a".into(),
278            seq: 3,
279            created_at: "t".into(),
280            text: "partial".into(),
281            is_reasoning: false,
282        };
283        assert!(EventEnvelopeV1::for_durable_event(delta, EventCorrelationV1::default()).is_err());
284    }
285}
rightkit_agent_engine_protocol/event/envelope.rs

rightkit_agent_engine_protocol/event/
envelope.rs
