eval_magic/adapters/claude_code/
stream_json.rs

1//! Claude Code `-p --output-format stream-json` transcript parsing.
2//!
3//! Parses the newline-delimited JSON event stream that `claude -p
4//! --output-format stream-json --verbose` writes (captured per task as
5//! `outputs/claude-events.jsonl`). The `assistant`/`user` events wrap a full
6//! Anthropic Messages object under `message`, so tool-call extraction is shared
7//! with the [`transcript`](super::transcript) record
8//! types. The differences are all in the envelope: there are no per-line
9//! timestamps, and a terminal `result` event carries the authoritative final
10//! text, wall-clock duration, and token usage. `system`, `rate_limit_event`, and
11//! any other non-message events are ignored (they don't deserialize into an
12//! assistant/user record, so the shared extractor skips them).
13
14use std::io;
15use std::path::Path;
16
17use serde::Deserialize;
18use serde_json::Value;
19
20use crate::core::ToolInvocation;
21
22use crate::adapters::TranscriptSummary;
23use crate::adapters::transcript::{TranscriptEvent, read_jsonl};
24
25use super::transcript::{
26    TranscriptRecord, UsageRecord, extract_events, extract_invocations, last_assistant_text,
27};
28
29/// The terminal `{"type":"result", …}` event of a `-p` stream-json run.
30#[derive(Debug, serde::Deserialize)]
31struct ResultEvent {
32    #[serde(default)]
33    result: Option<String>,
34    #[serde(default)]
35    duration_ms: Option<i64>,
36    #[serde(default)]
37    is_error: Option<bool>,
38    #[serde(default)]
39    usage: Option<UsageRecord>,
40}
41
42/// Parse the event stream into ordered tool invocations. Reuses the shared
43/// extractor: non-message events deserialize into records the extractor skips.
44pub fn parse_claude_stream_json(path: &Path) -> io::Result<Vec<ToolInvocation>> {
45    Ok(extract_invocations(&read_jsonl::<TranscriptRecord>(path)?))
46}
47
48/// Parse the event stream into a full [`TranscriptSummary`]. Final text,
49/// duration, and token totals come from the terminal `result` event; on a
50/// missing or errored `result` the final text falls back to the last assistant
51/// message's text, and duration/tokens fall back to `None`.
52pub fn parse_claude_stream_json_full(path: &Path) -> io::Result<TranscriptSummary> {
53    let values = read_jsonl::<Value>(path)?;
54    let mut records: Vec<TranscriptRecord> = Vec::new();
55    let mut result_event: Option<ResultEvent> = None;
56    for value in &values {
57        // Skip lines that don't shape into records rather than failing the parse.
58        let Ok(record) = TranscriptRecord::deserialize(value) else {
59            continue;
60        };
61        if record.record_type.as_deref() == Some("result") {
62            result_event = ResultEvent::deserialize(value).ok();
63        }
64        records.push(record);
65    }
66
67    let total_tokens = result_event
68        .as_ref()
69        .and_then(|e| e.usage.as_ref())
70        .map(|u| {
71            u.input_tokens.unwrap_or(0)
72                + u.output_tokens.unwrap_or(0)
73                + u.cache_creation_input_tokens.unwrap_or(0)
74                + u.cache_read_input_tokens.unwrap_or(0)
75        });
76    let duration_ms = result_event.as_ref().and_then(|e| e.duration_ms);
77    let final_text = match &result_event {
78        Some(ev) if ev.is_error != Some(true) => {
79            ev.result.clone().or_else(|| last_assistant_text(&records))
80        }
81        _ => last_assistant_text(&records),
82    };
83    let session_id = values.iter().find_map(|value| {
84        (value.get("type").and_then(Value::as_str) == Some("system"))
85            .then(|| value.get("session_id").and_then(Value::as_str))
86            .flatten()
87            .map(str::to_string)
88    });
89    let mut events = extract_events(&records);
90    if let Some(text) = &final_text {
91        let last_matches = events.iter().rev().find_map(|event| match event {
92            TranscriptEvent::AssistantMessage { text, .. } => Some(text),
93            TranscriptEvent::ToolInvocation { .. } => None,
94        }) == Some(text);
95        if !last_matches {
96            events.push(TranscriptEvent::AssistantMessage {
97                ordinal: events.len() as u32,
98                text: text.clone(),
99            });
100        }
101    }
102
103    Ok(TranscriptSummary {
104        tool_invocations: extract_invocations(&records),
105        events,
106        session_id,
107        total_tokens,
108        duration_ms,
109        final_text,
110    })
111}
112
113#[cfg(test)]
114mod tests {
115    use serde_json::{Value, json};
116    use std::fs;
117    use std::path::Path;
118    use tempfile::TempDir;
119
120    fn write_jsonl(path: &Path, lines: &[Value]) {
121        let body = lines
122            .iter()
123            .map(|l| l.to_string())
124            .collect::<Vec<_>>()
125            .join("\n");
126        fs::write(path, format!("{body}\n")).unwrap();
127    }
128
129    fn usage() -> Value {
130        json!({
131            "input_tokens": 4932,
132            "output_tokens": 139,
133            "cache_creation_input_tokens": 8287,
134            "cache_read_input_tokens": 33490,
135        })
136    }
137
138    #[test]
139    fn extracts_tool_use_and_result_skipping_non_message_events() {
140        let dir = TempDir::new().unwrap();
141        let path = dir.path().join("events.jsonl");
142        write_jsonl(
143            &path,
144            &[
145                json!({"type": "system", "subtype": "init", "cwd": "/env", "model": "claude-opus-4-8", "tools": ["Bash"]}),
146                json!({"type": "assistant", "message": {"id": "msg_1", "role": "assistant", "usage": usage(), "content": [
147                    {"type": "tool_use", "id": "toolu_1", "name": "Bash", "input": {"command": "ls"}}
148                ]}}),
149                json!({"type": "rate_limit_event", "rate_limit_info": {}}),
150                json!({"type": "user", "message": {"role": "user", "content": [
151                    {"type": "tool_result", "tool_use_id": "toolu_1", "content": "a.txt"}
152                ]}}),
153                json!({"type": "result", "subtype": "success", "is_error": false, "result": "Done", "duration_ms": 5637, "usage": usage()}),
154            ],
155        );
156
157        let result = super::parse_claude_stream_json(&path).unwrap();
158        assert_eq!(result.len(), 1);
159        assert_eq!(result[0].name, "Bash");
160        assert_eq!(result[0].ordinal, 0);
161        assert_eq!(result[0].args, Some(json!({"command": "ls"})));
162        assert_eq!(result[0].result, Some(Value::String("a.txt".into())));
163    }
164
165    #[test]
166    fn result_event_supplies_final_text_duration_and_tokens() {
167        let dir = TempDir::new().unwrap();
168        let path = dir.path().join("events.jsonl");
169        write_jsonl(
170            &path,
171            &[
172                json!({"type": "system", "subtype": "init", "session_id": "session-claude-1"}),
173                json!({"type": "assistant", "message": {"id": "msg_1", "role": "assistant", "content": [{"type": "text", "text": "working"}]}}),
174                json!({"type": "result", "subtype": "success", "is_error": false, "result": "Done", "duration_ms": 5637, "usage": usage()}),
175            ],
176        );
177
178        let summary = super::parse_claude_stream_json_full(&path).unwrap();
179        assert_eq!(summary.session_id.as_deref(), Some("session-claude-1"));
180        assert_eq!(
181            serde_json::to_value(&summary.events).unwrap(),
182            json!([
183                {
184                    "type": "assistant_message",
185                    "ordinal": 0,
186                    "text": "working"
187                },
188                {
189                    "type": "assistant_message",
190                    "ordinal": 1,
191                    "text": "Done"
192                }
193            ])
194        );
195        assert_eq!(summary.final_text, Some("Done".into()));
196        assert_eq!(summary.duration_ms, Some(5637));
197        // 4932 + 139 + 8287 + 33490
198        assert_eq!(summary.total_tokens, Some(46848));
199    }
200
201    #[test]
202    fn skill_tool_use_is_preserved_for_meta_check() {
203        let dir = TempDir::new().unwrap();
204        let path = dir.path().join("events.jsonl");
205        write_jsonl(
206            &path,
207            &[
208                json!({"type": "assistant", "message": {"id": "msg_1", "role": "assistant", "content": [
209                    {"type": "tool_use", "id": "toolu_s", "name": "Skill", "input": {"command": "slow-powers-eval-mr-review"}}
210                ]}}),
211                json!({"type": "result", "subtype": "success", "is_error": false, "result": "ok", "duration_ms": 10, "usage": usage()}),
212            ],
213        );
214        let result = super::parse_claude_stream_json(&path).unwrap();
215        assert_eq!(result.len(), 1);
216        assert_eq!(result[0].name, "Skill");
217        assert_eq!(
218            result[0].args,
219            Some(json!({"command": "slow-powers-eval-mr-review"}))
220        );
221    }
222
223    #[test]
224    fn final_text_falls_back_to_last_assistant_text_when_result_is_error() {
225        let dir = TempDir::new().unwrap();
226        let path = dir.path().join("events.jsonl");
227        write_jsonl(
228            &path,
229            &[
230                json!({"type": "assistant", "message": {"id": "msg_1", "role": "assistant", "content": [{"type": "text", "text": "partial work"}]}}),
231                json!({"type": "result", "subtype": "error_during_execution", "is_error": true, "result": "Execution error", "duration_ms": 12, "usage": usage()}),
232            ],
233        );
234        let summary = super::parse_claude_stream_json_full(&path).unwrap();
235        assert_eq!(summary.final_text, Some("partial work".into()));
236    }
237
238    #[test]
239    fn skips_malformed_jsonl_lines() {
240        let dir = TempDir::new().unwrap();
241        let path = dir.path().join("events.jsonl");
242        let good = json!({"type": "assistant", "message": {"id": "msg_1", "role": "assistant", "content": [
243            {"type": "tool_use", "id": "toolu_1", "name": "Read", "input": {"file_path": "/tmp/x"}}
244        ]}});
245        let result_line = json!({"type": "result", "is_error": false, "result": "Done", "duration_ms": 1, "usage": usage()});
246        let body = format!("{good}\nnot valid json\n{result_line}\n");
247        fs::write(&path, body).unwrap();
248
249        let result = super::parse_claude_stream_json(&path).unwrap();
250        assert_eq!(result.len(), 1);
251        assert_eq!(result[0].name, "Read");
252    }
253
254    #[test]
255    fn null_duration_and_tokens_when_no_result_event() {
256        let dir = TempDir::new().unwrap();
257        let path = dir.path().join("events.jsonl");
258        write_jsonl(
259            &path,
260            &[
261                json!({"type": "system", "subtype": "init"}),
262                json!({"type": "assistant", "message": {"id": "msg_1", "role": "assistant", "content": [{"type": "text", "text": "incomplete"}]}}),
263            ],
264        );
265        let summary = super::parse_claude_stream_json_full(&path).unwrap();
266        assert_eq!(summary.duration_ms, None);
267        assert_eq!(summary.total_tokens, None);
268        assert_eq!(summary.final_text, Some("incomplete".into()));
269    }
270}
eval_magic/adapters/claude_code/stream_json.rs

eval_magic/adapters/claude_code/
stream_json.rs
