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

phi_agent/render/
json_stream.rs
