phi_agent/render/
json_stream.rs

1#![allow(missing_docs)]
2
3use std::io::{self, Write};
4
5use agent_base::{AgentResult, RuntimeEvent, UserEvent};
6use serde_json::{Value, json};
7
8use crate::render::EventRenderer;
9
10/// JSON stream renderer: outputs one JSON line per event (JSONL format).
11/// Suitable for IDE integrations and programmatic consumers.
12pub struct JsonStreamRenderer {
13    writer: Box<dyn Write + Send>,
14    turn_start: Option<std::time::Instant>,
15    tool_call_count: u32,
16    last_assistant_text: String,
17}
18
19impl JsonStreamRenderer {
20    /// Create a new JSON stream renderer writing to the given writer.
21    pub fn new(writer: Box<dyn Write + Send>) -> Self {
22        Self { writer, turn_start: None, tool_call_count: 0, last_assistant_text: String::new() }
23    }
24
25    /// Create a renderer that writes to stdout.
26    pub fn stdout() -> Self {
27        Self::new(Box::new(io::stdout()))
28    }
29
30    fn emit(&mut self, value: &Value) -> AgentResult<()> {
31        let line = serde_json::to_string(value)
32            .map_err(|e| agent_base::AgentError::internal(format!("JSON serialize error: {e}")))?;
33        writeln!(self.writer, "{}", line).map_err(|e| agent_base::AgentError::internal(format!("write error: {e}")))?;
34        Ok(())
35    }
36
37    /// Emit an event line, attaching `agent_id` when the event carries one.
38    fn emit_event(&mut self, event: &RuntimeEvent, mut value: Value) -> AgentResult<()> {
39        if let Some(agent_id) = event.agent_id() {
40            value["agent_id"] = json!(agent_id);
41        }
42        self.emit(&value)
43    }
44}
45
46impl EventRenderer for JsonStreamRenderer {
47    fn render(&mut self, event: RuntimeEvent) -> AgentResult<()> {
48        if self.turn_start.is_none() {
49            self.turn_start = Some(std::time::Instant::now());
50        }
51
52        match &event {
53            RuntimeEvent::ThoughtDelta { text, .. } => {
54                self.emit_event(&event, json!({ "type": "thought_delta", "text": text }))?;
55            },
56            RuntimeEvent::TextDelta { text, .. } => {
57                if event.agent_id().is_none() {
58                    self.last_assistant_text.push_str(text);
59                }
60                self.emit_event(&event, json!({ "type": "text_delta", "text": text }))?;
61            },
62            RuntimeEvent::ToolCallStarted { tool_name, args_json, .. } => {
63                self.tool_call_count += 1;
64                let args: Value = serde_json::from_str(args_json).unwrap_or(Value::Null);
65                self.emit_event(
66                    &event,
67                    json!({
68                        "type": "tool_call_started",
69                        "tool": tool_name,
70                        "args": args,
71                    }),
72                )?;
73            },
74            RuntimeEvent::ToolCallFinished { tool_name, summary, denied, .. } => {
75                self.emit_event(
76                    &event,
77                    json!({
78                        "type": "tool_call_finished",
79                        "tool": tool_name,
80                        "summary": summary,
81                        "denied": denied,
82                    }),
83                )?;
84            },
85            RuntimeEvent::AwaitingApproval { request, .. } => {
86                self.emit_event(
87                    &event,
88                    json!({
89                        "type": "approval_request",
90                        "title": request.title,
91                        "risk": format!("{:?}", request.risk_level),
92                        "message": request.message,
93                    }),
94                )?;
95            },
96            RuntimeEvent::PlanUpdated { explanation, plan, .. } => {
97                self.emit_event(
98                    &event,
99                    json!({
100                        "type": "plan_updated",
101                        "explanation": explanation,
102                        "plan": plan,
103                    }),
104                )?;
105            },
106            RuntimeEvent::UserEvent { event: UserEvent::Structured { event_type, data }, .. } => {
107                self.emit_event(
108                    &event,
109                    json!({
110                        "type": "user_event",
111                        "event_type": event_type,
112                        "data": data,
113                    }),
114                )?;
115            },
116            RuntimeEvent::UserEvent { .. } => {},
117            RuntimeEvent::Checkpoint { .. } => {},
118            RuntimeEvent::RunFinished { .. } => {},
119            RuntimeEvent::RunCancelled { .. } => {
120                self.emit_event(&event, json!({ "type": "run_cancelled" }))?;
121            },
122        }
123
124        Ok(())
125    }
126
127    fn finish_turn(&mut self) -> AgentResult<()> {
128        let duration_ms = self.turn_start.map(|s| s.elapsed().as_millis() as u64).unwrap_or(0);
129
130        self.emit(&json!({
131            "type": "turn_finished",
132            "duration_ms": duration_ms,
133            "tool_call_count": self.tool_call_count,
134            "assistant_text": self.last_assistant_text.trim(),
135        }))?;
136
137        self.turn_start = None;
138        self.tool_call_count = 0;
139        self.last_assistant_text.clear();
140
141        Ok(())
142    }
143}
144
145#[cfg(test)]
146mod tests {
147    use super::*;
148    use agent_base::{ApprovalRequest, PlanItem, PlanStepStatus, RiskLevel, SessionId, UserEvent};
149    use std::io::Write;
150    use std::sync::{Arc, Mutex};
151
152    struct SharedWriter {
153        inner: Arc<Mutex<Vec<u8>>>,
154    }
155
156    impl Write for SharedWriter {
157        fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
158            self.inner.lock().unwrap().extend_from_slice(data);
159            Ok(data.len())
160        }
161        fn flush(&mut self) -> std::io::Result<()> {
162            Ok(())
163        }
164    }
165
166    impl SharedWriter {
167        fn new() -> (Self, Arc<Mutex<Vec<u8>>>) {
168            let inner = Arc::new(Mutex::new(Vec::new()));
169            (Self { inner: inner.clone() }, inner)
170        }
171    }
172
173    fn session_id() -> SessionId {
174        SessionId { id: 1, external_id: None }
175    }
176
177    fn render_one(event: RuntimeEvent) -> Vec<String> {
178        let (writer, buf) = SharedWriter::new();
179        let mut r = JsonStreamRenderer::new(Box::new(writer));
180        r.render(event).unwrap();
181        drop(r);
182        let text = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
183        if text.is_empty() { vec![] } else { text.lines().map(|l| l.to_string()).collect() }
184    }
185
186    fn render_and_finish(events: &[RuntimeEvent]) -> Vec<String> {
187        let (writer, buf) = SharedWriter::new();
188        let mut r = JsonStreamRenderer::new(Box::new(writer));
189        for e in events {
190            r.render(e.clone()).unwrap();
191        }
192        r.finish_turn().unwrap();
193        drop(r);
194        let text = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
195        text.lines().map(|l| l.to_string()).collect()
196    }
197
198    #[test]
199    fn test_text_delta_produces_valid_json() {
200        let lines = render_one(RuntimeEvent::TextDelta {
201            session_id: session_id(),
202            text: "hello".into(),
203            agent_id: None,
204            trace_id: None,
205        });
206        assert_eq!(lines.len(), 1);
207        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
208        assert_eq!(v["type"], "text_delta");
209        assert_eq!(v["text"], "hello");
210    }
211
212    #[test]
213    fn test_thought_delta_produces_valid_json() {
214        let lines = render_one(RuntimeEvent::ThoughtDelta {
215            session_id: session_id(),
216            text: "thinking...".into(),
217            agent_id: None,
218            trace_id: None,
219        });
220        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
221        assert_eq!(v["type"], "thought_delta");
222    }
223
224    #[test]
225    fn test_tool_call_started_parses_args() {
226        let lines = render_one(RuntimeEvent::ToolCallStarted {
227            session_id: session_id(),
228            tool_name: "shell".into(),
229            args_json: r#"{"cmd":"ls"}"#.into(),
230            agent_id: None,
231            trace_id: None,
232        });
233        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
234        assert_eq!(v["type"], "tool_call_started");
235        assert_eq!(v["tool"], "shell");
236        assert_eq!(v["args"]["cmd"], "ls");
237    }
238
239    #[test]
240    fn test_tool_call_finished_produces_valid_json() {
241        let lines = render_one(RuntimeEvent::ToolCallFinished {
242            session_id: session_id(),
243            tool_name: "shell".into(),
244            summary: "done".into(),
245            agent_id: None,
246            trace_id: None,
247            denied: false,
248            details: None,
249        });
250        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
251        assert_eq!(v["type"], "tool_call_finished");
252        assert_eq!(v["tool"], "shell");
253        assert_eq!(v["summary"], "done");
254        assert_eq!(v["denied"], false);
255    }
256
257    #[test]
258    fn test_awaiting_approval_produces_valid_json() {
259        let lines = render_one(RuntimeEvent::AwaitingApproval {
260            session_id: session_id(),
261            request: ApprovalRequest {
262                title: "Delete".into(),
263                message: "Dangerous".into(),
264                action_key: None,
265                risk_level: RiskLevel::Destructive,
266                raw: None,
267                source: None,
268            },
269            agent_id: None,
270            trace_id: None,
271        });
272        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
273        assert_eq!(v["type"], "approval_request");
274        assert_eq!(v["title"], "Delete");
275    }
276
277    #[test]
278    fn test_plan_updated_produces_valid_json() {
279        let lines = render_one(RuntimeEvent::PlanUpdated {
280            session_id: session_id(),
281            objective: "test".into(),
282            explanation: Some("step 1 done".into()),
283            plan: vec![PlanItem { step: "Step 1".into(), status: PlanStepStatus::Completed }],
284            agent_id: None,
285            trace_id: None,
286        });
287        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
288        assert_eq!(v["type"], "plan_updated");
289    }
290
291    #[test]
292    fn test_user_event_structured() {
293        let lines = render_one(RuntimeEvent::UserEvent {
294            session_id: session_id(),
295            event: UserEvent::Structured { event_type: "custom".into(), data: serde_json::json!({"key": "value"}) },
296            agent_id: None,
297            trace_id: None,
298        });
299        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
300        assert_eq!(v["type"], "user_event");
301        assert_eq!(v["event_type"], "custom");
302        assert_eq!(v["data"]["key"], "value");
303    }
304
305    #[test]
306    fn test_user_event_progress_ignored() {
307        let lines = render_one(RuntimeEvent::UserEvent {
308            session_id: session_id(),
309            event: UserEvent::Progress { text: "loading...".into() },
310            agent_id: None,
311            trace_id: None,
312        });
313        assert!(lines.is_empty());
314    }
315
316    #[test]
317    fn test_run_finished_no_output() {
318        let lines = render_one(RuntimeEvent::RunFinished { session_id: session_id(), agent_id: None, trace_id: None });
319        assert!(lines.is_empty());
320    }
321
322    #[test]
323    fn test_run_cancelled_produces_valid_json() {
324        let lines = render_one(RuntimeEvent::RunCancelled { session_id: session_id(), agent_id: None, trace_id: None });
325        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
326        assert_eq!(v["type"], "run_cancelled");
327    }
328
329    #[test]
330    fn test_finish_turn_emits_summary() {
331        let lines = render_and_finish(&[RuntimeEvent::TextDelta {
332            session_id: session_id(),
333            text: "hello".into(),
334            agent_id: None,
335            trace_id: None,
336        }]);
337        let last: serde_json::Value = serde_json::from_str(lines.last().unwrap()).unwrap();
338        assert_eq!(last["type"], "turn_finished");
339        assert!(last["duration_ms"].as_u64().is_some());
340        assert_eq!(last["tool_call_count"], 0);
341    }
342
343    #[test]
344    fn test_tool_call_count_incremented() {
345        let lines = render_and_finish(&[
346            RuntimeEvent::ToolCallStarted {
347                session_id: session_id(),
348                tool_name: "a".into(),
349                args_json: "{}".into(),
350                agent_id: None,
351                trace_id: None,
352            },
353            RuntimeEvent::ToolCallStarted {
354                session_id: session_id(),
355                tool_name: "b".into(),
356                args_json: "{}".into(),
357                agent_id: None,
358                trace_id: None,
359            },
360        ]);
361        let last: serde_json::Value = serde_json::from_str(lines.last().unwrap()).unwrap();
362        assert_eq!(last["tool_call_count"], 2);
363    }
364
365    #[test]
366    fn test_assistant_text_accumulated() {
367        let lines = render_and_finish(&[
368            RuntimeEvent::TextDelta { session_id: session_id(), text: "Hello ".into(), agent_id: None, trace_id: None },
369            RuntimeEvent::TextDelta { session_id: session_id(), text: "World".into(), agent_id: None, trace_id: None },
370        ]);
371        let last: serde_json::Value = serde_json::from_str(lines.last().unwrap()).unwrap();
372        assert_eq!(last["assistant_text"], "Hello World");
373    }
374
375    // ── Sub-agent (agent_id) tests ──
376
377    #[test]
378    fn test_subagent_text_delta_includes_agent_id() {
379        let lines = render_one(RuntimeEvent::TextDelta {
380            session_id: session_id(),
381            text: "found 3 items".into(),
382            agent_id: Some("root/searcher".into()),
383            trace_id: None,
384        });
385        assert_eq!(lines.len(), 1);
386        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
387        assert_eq!(v["type"], "text_delta");
388        assert_eq!(v["text"], "found 3 items");
389        assert_eq!(v["agent_id"], "root/searcher");
390    }
391
392    #[test]
393    fn test_subagent_tool_call_includes_agent_id() {
394        let lines = render_one(RuntimeEvent::ToolCallStarted {
395            session_id: session_id(),
396            tool_name: "shell".into(),
397            args_json: r#"{"cmd":"ls"}"#.into(),
398            agent_id: Some("root/worker".into()),
399            trace_id: None,
400        });
401        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
402        assert_eq!(v["type"], "tool_call_started");
403        assert_eq!(v["tool"], "shell");
404        assert_eq!(v["agent_id"], "root/worker");
405    }
406
407    #[test]
408    fn test_root_event_omits_agent_id() {
409        let lines = render_one(RuntimeEvent::TextDelta {
410            session_id: session_id(),
411            text: "hello".into(),
412            agent_id: None,
413            trace_id: None,
414        });
415        let v: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
416        assert!(v.get("agent_id").is_none());
417    }
418
419    #[test]
420    fn test_subagent_text_does_not_pollute_assistant_text() {
421        let lines = render_and_finish(&[
422            RuntimeEvent::TextDelta { session_id: session_id(), text: "root ".into(), agent_id: None, trace_id: None },
423            RuntimeEvent::TextDelta {
424                session_id: session_id(),
425                text: "child".into(),
426                agent_id: Some("root/child".into()),
427                trace_id: None,
428            },
429        ]);
430        let last: serde_json::Value = serde_json::from_str(lines.last().unwrap()).unwrap();
431        assert_eq!(last["assistant_text"], "root");
432    }
433}
phi_agent/render/json_stream.rs

phi_agent/render/
json_stream.rs
