phi_agent/
event_log.rs

1//! Event log: serialize turn events to JSONL files.
2//!
3//! Provides event persistence integrated with SessionContext,
4//! shared across CLI, Web, and other consumers.
5//!
6//! Note: `save_turn_log` performs synchronous file I/O. Consumers should
7//! call it via `tokio::task::spawn_blocking` in async contexts to avoid
8//! blocking the runtime.
9
10use std::io::Write;
11
12use agent_base::{AgentResult, RuntimeEvent, UserEvent};
13
14use crate::session::SessionContext;
15
16/// Save all events from a turn to a JSONL file.
17///
18/// Performs synchronous file I/O. Callers should invoke this via
19/// `tokio::task::spawn_blocking`.
20pub fn save_turn_log(
21    session_ctx: &SessionContext,
22    turn: u32,
23    events: &[RuntimeEvent],
24    user_input: &str,
25) -> AgentResult<()> {
26    let turn_path = session_ctx.turn_path(turn as usize);
27    let mut file = std::fs::OpenOptions::new().create(true).append(true).open(&turn_path)?;
28
29    let meta = serde_json::json!({
30        "turn": turn,
31        "timestamp": chrono::Utc::now().to_rfc3339(),
32        "user_input": user_input,
33    });
34    writeln!(file, "{}", serde_json::to_string(&meta)?)?;
35
36    for event in events {
37        let line = event_to_jsonl(event);
38        writeln!(file, "{}", line)?;
39    }
40
41    writeln!(file, "{}", serde_json::to_string(&serde_json::json!({"type": "turn_end", "turn": turn}))?)?;
42
43    file.flush()?;
44
45    Ok(())
46}
47
48/// Convert a RuntimeEvent to a JSON Value (shared by event_log and render).
49pub fn event_to_value(event: &RuntimeEvent) -> serde_json::Value {
50    match event {
51        RuntimeEvent::ThoughtDelta { text, .. } => {
52            serde_json::json!({"type": "thought_delta", "text": text})
53        },
54        RuntimeEvent::TextDelta { text, .. } => {
55            serde_json::json!({"type": "text_delta", "text": text})
56        },
57        RuntimeEvent::ToolCallStarted { tool_name, args_json, .. } => {
58            let args: serde_json::Value = serde_json::from_str(args_json).unwrap_or(serde_json::Value::Null);
59            serde_json::json!({"type": "tool_call_started", "tool": tool_name, "args": args})
60        },
61        RuntimeEvent::ToolCallFinished { tool_name, summary, .. } => {
62            serde_json::json!({"type": "tool_call_finished", "tool": tool_name, "summary": summary})
63        },
64        RuntimeEvent::AwaitingApproval { request, .. } => {
65            serde_json::json!({"type": "approval_request", "title": request.title, "message": request.message})
66        },
67        RuntimeEvent::PlanUpdated { explanation, plan, .. } => {
68            serde_json::json!({"type": "plan_updated", "explanation": explanation, "plan": plan})
69        },
70        RuntimeEvent::UserEvent { event: UserEvent::Structured { event_type, data }, .. } => {
71            serde_json::json!({"type": "user_event", "event_type": event_type, "data": data})
72        },
73        RuntimeEvent::UserEvent { event: UserEvent::SubAgentEvent { subagent, event: inner }, .. } => {
74            let inner = event_to_value(inner);
75            serde_json::json!({"type": "subagent_event", "subagent": subagent, "event": inner})
76        },
77        RuntimeEvent::UserEvent { .. } => serde_json::json!({"type": "other"}),
78        RuntimeEvent::RunCancelled { .. } => serde_json::json!({"type": "run_cancelled"}),
79        RuntimeEvent::RunFinished { .. } => serde_json::json!({"type": "run_finished"}),
80        RuntimeEvent::Checkpoint { .. } => serde_json::json!({"type": "checkpoint"}),
81    }
82}
83
84/// Convert a RuntimeEvent to a JSONL line.
85pub fn event_to_jsonl(event: &RuntimeEvent) -> String {
86    let value = event_to_value(event);
87    serde_json::to_string(&value).unwrap_or_else(|_| r#"{"type":"serialize_error"}"#.to_string())
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93    use agent_base::{SessionId, UserEvent};
94    use tempfile::TempDir;
95
96    fn session_id() -> SessionId {
97        SessionId { id: 1, external_id: None }
98    }
99
100    // ── event_to_value tests ──
101
102    #[test]
103    fn test_event_to_value_text_delta() {
104        let v = event_to_value(&RuntimeEvent::TextDelta {
105            session_id: session_id(),
106            text: "hello".into(),
107            agent_id: None,
108            trace_id: None,
109        });
110        assert_eq!(v["type"], "text_delta");
111        assert_eq!(v["text"], "hello");
112    }
113
114    #[test]
115    fn test_event_to_value_thought_delta() {
116        let v = event_to_value(&RuntimeEvent::ThoughtDelta {
117            session_id: session_id(),
118            text: "thinking".into(),
119            agent_id: None,
120            trace_id: None,
121        });
122        assert_eq!(v["type"], "thought_delta");
123    }
124
125    #[test]
126    fn test_event_to_value_tool_call_started() {
127        let v = event_to_value(&RuntimeEvent::ToolCallStarted {
128            session_id: session_id(),
129            tool_name: "shell".into(),
130            args_json: r#"{"cmd":"ls"}"#.into(),
131            agent_id: None,
132            trace_id: None,
133        });
134        assert_eq!(v["type"], "tool_call_started");
135        assert_eq!(v["tool"], "shell");
136        assert_eq!(v["args"]["cmd"], "ls");
137    }
138
139    #[test]
140    fn test_event_to_value_tool_call_finished() {
141        let v = event_to_value(&RuntimeEvent::ToolCallFinished {
142            session_id: session_id(),
143            tool_name: "shell".into(),
144            summary: "done".into(),
145            agent_id: None,
146            trace_id: None,
147        });
148        assert_eq!(v["type"], "tool_call_finished");
149        assert_eq!(v["summary"], "done");
150    }
151
152    #[test]
153    fn test_event_to_value_run_finished() {
154        let v = event_to_value(&RuntimeEvent::RunFinished { session_id: session_id(), agent_id: None, trace_id: None });
155        assert_eq!(v["type"], "run_finished");
156    }
157
158    #[test]
159    fn test_event_to_value_run_cancelled() {
160        let v =
161            event_to_value(&RuntimeEvent::RunCancelled { session_id: session_id(), agent_id: None, trace_id: None });
162        assert_eq!(v["type"], "run_cancelled");
163    }
164
165    #[test]
166    fn test_event_to_value_user_event_other() {
167        let v = event_to_value(&RuntimeEvent::UserEvent {
168            session_id: session_id(),
169            event: UserEvent::Progress { text: "loading".into() },
170            agent_id: None,
171            trace_id: None,
172        });
173        assert_eq!(v["type"], "other");
174    }
175
176    #[test]
177    fn test_event_to_value_user_event_structured() {
178        let v = event_to_value(&RuntimeEvent::UserEvent {
179            session_id: session_id(),
180            event: UserEvent::Structured { event_type: "custom".into(), data: serde_json::json!({"key": "value"}) },
181            agent_id: None,
182            trace_id: None,
183        });
184        assert_eq!(v["type"], "user_event");
185        assert_eq!(v["event_type"], "custom");
186    }
187
188    // ── event_to_jsonl tests ──
189
190    #[test]
191    fn test_event_to_jsonl_returns_valid_json() {
192        let line = event_to_jsonl(&RuntimeEvent::TextDelta {
193            session_id: session_id(),
194            text: "hello".into(),
195            agent_id: None,
196            trace_id: None,
197        });
198        let v: serde_json::Value = serde_json::from_str(&line).unwrap();
199        assert_eq!(v["type"], "text_delta");
200    }
201
202    #[test]
203    fn test_event_to_jsonl_all_variants_valid() {
204        let events: Vec<RuntimeEvent> = vec![
205            RuntimeEvent::TextDelta { session_id: session_id(), text: "t".into(), agent_id: None, trace_id: None },
206            RuntimeEvent::ThoughtDelta { session_id: session_id(), text: "t".into(), agent_id: None, trace_id: None },
207            RuntimeEvent::ToolCallStarted {
208                session_id: session_id(),
209                tool_name: "t".into(),
210                args_json: "{}".into(),
211                agent_id: None,
212                trace_id: None,
213            },
214            RuntimeEvent::ToolCallFinished {
215                session_id: session_id(),
216                tool_name: "t".into(),
217                summary: "s".into(),
218                agent_id: None,
219                trace_id: None,
220            },
221            RuntimeEvent::RunFinished { session_id: session_id(), agent_id: None, trace_id: None },
222            RuntimeEvent::RunCancelled { session_id: session_id(), agent_id: None, trace_id: None },
223            RuntimeEvent::UserEvent {
224                session_id: session_id(),
225                event: UserEvent::Progress { text: "t".into() },
226                agent_id: None,
227                trace_id: None,
228            },
229        ];
230        for e in &events {
231            let line = event_to_jsonl(e);
232            assert!(serde_json::from_str::<serde_json::Value>(&line).is_ok(), "Failed to parse JSONL for event");
233        }
234    }
235
236    // ── save_turn_log tests ──
237
238    fn make_session_ctx(tmp: &TempDir) -> SessionContext {
239        crate::session::resolve_session(Some("test-session"), tmp.path()).unwrap()
240    }
241
242    #[test]
243    fn test_save_turn_log_creates_file() {
244        let tmp = TempDir::new().unwrap();
245        let ctx = make_session_ctx(&tmp);
246        let events = vec![RuntimeEvent::TextDelta {
247            session_id: session_id(),
248            text: "hello".into(),
249            agent_id: None,
250            trace_id: None,
251        }];
252        save_turn_log(&ctx, 1, &events, "test input").unwrap();
253        let path = ctx.turn_path(1);
254        assert!(path.exists());
255        let content = std::fs::read_to_string(&path).unwrap();
256        assert!(!content.is_empty());
257    }
258
259    #[test]
260    fn test_save_turn_log_contains_metadata() {
261        let tmp = TempDir::new().unwrap();
262        let ctx = make_session_ctx(&tmp);
263        let events = vec![RuntimeEvent::TextDelta {
264            session_id: session_id(),
265            text: "hello".into(),
266            agent_id: None,
267            trace_id: None,
268        }];
269        save_turn_log(&ctx, 1, &events, "my input").unwrap();
270        let content = std::fs::read_to_string(ctx.turn_path(1)).unwrap();
271        let lines: Vec<&str> = content.lines().collect();
272        let first: serde_json::Value = serde_json::from_str(lines[0]).unwrap();
273        assert_eq!(first["turn"], 1);
274        assert_eq!(first["user_input"], "my input");
275        assert!(first["timestamp"].as_str().is_some());
276    }
277
278    #[test]
279    fn test_save_turn_log_contains_events() {
280        let tmp = TempDir::new().unwrap();
281        let ctx = make_session_ctx(&tmp);
282        let events = vec![
283            RuntimeEvent::TextDelta { session_id: session_id(), text: "hello".into(), agent_id: None, trace_id: None },
284            RuntimeEvent::RunFinished { session_id: session_id(), agent_id: None, trace_id: None },
285        ];
286        save_turn_log(&ctx, 1, &events, "input").unwrap();
287        let content = std::fs::read_to_string(ctx.turn_path(1)).unwrap();
288        let lines: Vec<&str> = content.lines().collect();
289        // Line 0: metadata, Line 1: text_delta, Line 2: run_finished, Line 3: turn_end
290        assert!(lines.len() >= 4);
291        let line1: serde_json::Value = serde_json::from_str(lines[1]).unwrap();
292        assert_eq!(line1["type"], "text_delta");
293        let line2: serde_json::Value = serde_json::from_str(lines[2]).unwrap();
294        assert_eq!(line2["type"], "run_finished");
295    }
296
297    #[test]
298    fn test_save_turn_log_contains_turn_end_marker() {
299        let tmp = TempDir::new().unwrap();
300        let ctx = make_session_ctx(&tmp);
301        let events = vec![RuntimeEvent::TextDelta {
302            session_id: session_id(),
303            text: "hello".into(),
304            agent_id: None,
305            trace_id: None,
306        }];
307        save_turn_log(&ctx, 1, &events, "input").unwrap();
308        let content = std::fs::read_to_string(ctx.turn_path(1)).unwrap();
309        let last_line = content.lines().last().unwrap();
310        let v: serde_json::Value = serde_json::from_str(last_line).unwrap();
311        assert_eq!(v["type"], "turn_end");
312        assert_eq!(v["turn"], 1);
313    }
314
315    #[test]
316    fn test_save_turn_log_appends() {
317        let tmp = TempDir::new().unwrap();
318        let ctx = make_session_ctx(&tmp);
319        let events1 = vec![RuntimeEvent::TextDelta {
320            session_id: session_id(),
321            text: "turn1".into(),
322            agent_id: None,
323            trace_id: None,
324        }];
325        let events2 = vec![RuntimeEvent::TextDelta {
326            session_id: session_id(),
327            text: "turn2".into(),
328            agent_id: None,
329            trace_id: None,
330        }];
331        save_turn_log(&ctx, 1, &events1, "input1").unwrap();
332        save_turn_log(&ctx, 1, &events2, "input2").unwrap();
333        let content = std::fs::read_to_string(ctx.turn_path(1)).unwrap();
334        assert!(content.contains("turn1"));
335        assert!(content.contains("turn2"));
336        // Should have two turn_end markers
337        let turn_end_count = content.lines().filter(|l| l.contains("turn_end")).count();
338        assert_eq!(turn_end_count, 2);
339    }
340
341    #[test]
342    fn test_save_turn_log_empty_events() {
343        let tmp = TempDir::new().unwrap();
344        let ctx = make_session_ctx(&tmp);
345        save_turn_log(&ctx, 1, &[], "no events").unwrap();
346        let content = std::fs::read_to_string(ctx.turn_path(1)).unwrap();
347        let lines: Vec<&str> = content.lines().collect();
348        // metadata header + turn_end marker
349        assert_eq!(lines.len(), 2);
350        let last: serde_json::Value = serde_json::from_str(lines[1]).unwrap();
351        assert_eq!(last["type"], "turn_end");
352        assert_eq!(last["turn"], 1);
353    }
354
355    // ── SubAgentEvent tests ──
356
357    #[test]
358    fn test_event_to_value_subagent_event() {
359        let v = event_to_value(&RuntimeEvent::UserEvent {
360            session_id: session_id(),
361            event: UserEvent::SubAgentEvent {
362                subagent: "root/searcher".into(),
363                event: Box::new(RuntimeEvent::TextDelta {
364                    session_id: session_id(),
365                    text: "result".into(),
366                    agent_id: None,
367                    trace_id: None,
368                }),
369            },
370            agent_id: None,
371            trace_id: None,
372        });
373        assert_eq!(v["type"], "subagent_event");
374        assert_eq!(v["subagent"], "root/searcher");
375        assert_eq!(v["event"]["type"], "text_delta");
376        assert_eq!(v["event"]["text"], "result");
377    }
378
379    #[test]
380    fn test_event_to_value_nested_subagent_event() {
381        // Subagent event wrapping another subagent event (nested)
382        let v = event_to_value(&RuntimeEvent::UserEvent {
383            session_id: session_id(),
384            event: UserEvent::SubAgentEvent {
385                subagent: "root/parent".into(),
386                event: Box::new(RuntimeEvent::UserEvent {
387                    session_id: session_id(),
388                    event: UserEvent::SubAgentEvent {
389                        subagent: "root/parent/child".into(),
390                        event: Box::new(RuntimeEvent::TextDelta {
391                            session_id: session_id(),
392                            text: "deep".into(),
393                            agent_id: None,
394                            trace_id: None,
395                        }),
396                    },
397                    agent_id: None,
398                    trace_id: None,
399                }),
400            },
401            agent_id: None,
402            trace_id: None,
403        });
404        assert_eq!(v["type"], "subagent_event");
405        assert_eq!(v["subagent"], "root/parent");
406        assert_eq!(v["event"]["type"], "subagent_event");
407        assert_eq!(v["event"]["subagent"], "root/parent/child");
408        assert_eq!(v["event"]["event"]["text"], "deep");
409    }
410}
phi_agent/event_log.rs

phi_agent/
event_log.rs
