claudex Public

Watch 0 Fork 0 Star 0
Rust · 21332 bytes Raw Blame History
1 //! Full session reader. Streams a `.jsonl` file, converts each event
2 //! to a frontend-facing [`Message`], and assembles a [`SessionDetail`]
3 //! alongside the cheap metadata summary.
4 //!
5 //! Session-level metadata events (`permission-mode`, `custom-title`,
6 //! `agent-name`, `file-history-snapshot`, `queue-operation`) are
7 //! filtered out — they don't belong in the viewer timeline.
8 //!
9 //! Sidechain events are also filtered in v0. Partial/broken lines are
10 //! skipped silently; the reader never fails on malformed data.
11
12 use std::fs::File;
13 use std::io::{BufRead, BufReader};
14 use std::path::Path;
15
16 use chrono::{DateTime, Utc};
17 use serde_json::Value;
18
19 use crate::core::error::CoreResult;
20 use crate::core::metadata::summarize;
21 use crate::core::schema::{ContentBlock, Message, RawEvent, SessionDetail, Usage};
22
23 pub fn read_session(path: &Path, project_id: &str) -> CoreResult<SessionDetail> {
24 read_session_limited(path, project_id, None)
25 }
26
27 /// Read a session, optionally capping the result to the most recent
28 /// `limit` messages. Required for host performance on very large
29 /// sessions — one user's armfortas session is 171 MB of JSONL and
30 /// that's enough to hang the main thread for seconds during IPC
31 /// deserialization. Passing `Some(N)` streams the whole file but
32 /// only retains the tail, so the wire payload stays bounded.
33 ///
34 /// The returned [`SessionDetail`] is identical in shape regardless
35 /// of whether `limit` was set; truncation is invisible to the
36 /// frontend except that `messages.len() < summary.message_count`.
37 pub fn read_session_limited(
38 path: &Path,
39 project_id: &str,
40 limit: Option<usize>,
41 ) -> CoreResult<SessionDetail> {
42 let summary = summarize(path, project_id)?;
43 let messages = read_messages(path, limit)?;
44 Ok(SessionDetail { summary, messages })
45 }
46
47 fn read_messages(path: &Path, limit: Option<usize>) -> CoreResult<Vec<Message>> {
48 let file = File::open(path)?;
49 let reader = BufReader::new(file);
50 let mut fallback_counter: u32 = 0;
51
52 // When a cap is set, keep a rolling tail so we never allocate a
53 // `Vec<Message>` bigger than the cap + 1. This keeps memory
54 // bounded on multi-hundred-megabyte files.
55 match limit {
56 Some(cap) if cap > 0 => {
57 let mut tail: std::collections::VecDeque<Message> =
58 std::collections::VecDeque::with_capacity(cap);
59 for line in reader.lines() {
60 let line = match line {
61 Ok(l) => l,
62 Err(_) => continue,
63 };
64 if line.is_empty() {
65 continue;
66 }
67 let ev: RawEvent = match serde_json::from_str(&line) {
68 Ok(e) => e,
69 Err(_) => continue,
70 };
71 if ev.is_sidechain.unwrap_or(false) {
72 continue;
73 }
74 if let Some(msg) = raw_to_message(ev, &mut fallback_counter) {
75 if tail.len() == cap {
76 tail.pop_front();
77 }
78 tail.push_back(msg);
79 }
80 }
81 Ok(tail.into_iter().collect())
82 }
83 _ => {
84 let mut out = Vec::new();
85 for line in reader.lines() {
86 let line = match line {
87 Ok(l) => l,
88 Err(_) => continue,
89 };
90 if line.is_empty() {
91 continue;
92 }
93 let ev: RawEvent = match serde_json::from_str(&line) {
94 Ok(e) => e,
95 Err(_) => continue,
96 };
97 if ev.is_sidechain.unwrap_or(false) {
98 continue;
99 }
100 if let Some(msg) = raw_to_message(ev, &mut fallback_counter) {
101 out.push(msg);
102 }
103 }
104 Ok(out)
105 }
106 }
107 }
108
109 /// Convert one raw event into a timeline message. Returns `None` for
110 /// session-level metadata events and for anything we can't construct
111 /// a stable id/timestamp for.
112 pub(crate) fn raw_to_message(ev: RawEvent, fallback_counter: &mut u32) -> Option<Message> {
113 match ev.kind.as_str() {
114 "permission-mode"
115 | "custom-title"
116 | "agent-name"
117 | "ai-title"
118 | "file-history-snapshot"
119 | "queue-operation"
120 | "progress"
121 | "last-prompt"
122 | "pr-link" => return None,
123 _ => {}
124 }
125
126 let id = ev.uuid.clone().unwrap_or_else(|| {
127 *fallback_counter += 1;
128 format!("synthetic-{}", fallback_counter)
129 });
130 let at = ev.timestamp;
131
132 match ev.kind.as_str() {
133 "user" => {
134 let text = ev.message.as_ref().and_then(extract_user_text)?;
135 Some(Message::User {
136 id,
137 at: at?,
138 text,
139 is_meta: ev.is_meta.unwrap_or(false),
140 })
141 }
142
143 "assistant" => {
144 let (model, blocks, stop_reason, usage) = extract_assistant(ev.message.as_ref());
145 if blocks.is_empty() && model.is_none() {
146 // Skip empty assistant shells.
147 return None;
148 }
149 Some(Message::Assistant {
150 id,
151 at: at?,
152 model,
153 blocks,
154 stop_reason,
155 usage,
156 status: None,
157 })
158 }
159
160 "system" => {
161 let text = extract_system_text(&ev);
162 Some(Message::System {
163 id,
164 at: at?,
165 text,
166 subtype: ev.subtype.clone(),
167 })
168 }
169
170 "attachment" => {
171 let (attachment_type, hook_name, text) = extract_attachment(ev.attachment.as_ref());
172 Some(Message::Attachment {
173 id,
174 at: at?,
175 attachment_type,
176 hook_name,
177 text,
178 })
179 }
180
181 raw_type => Some(Message::Unknown {
182 id,
183 at,
184 raw_type: raw_type.to_string(),
185 raw: raw_event_to_value(&ev),
186 }),
187 }
188 }
189
190 fn extract_user_text(msg: &Value) -> Option<String> {
191 match msg.get("content") {
192 Some(Value::String(s)) => Some(s.clone()),
193 Some(Value::Array(blocks)) => {
194 // A tool_result block looks like:
195 // { "type": "tool_result", "tool_use_id": "...", "content": "...", "is_error": ... }
196 // Mixed blocks are rendered as concatenated text for v0.
197 let mut parts = Vec::new();
198 for b in blocks {
199 match b.get("type").and_then(Value::as_str) {
200 Some("text") => {
201 if let Some(t) = b.get("text").and_then(Value::as_str) {
202 parts.push(t.to_string());
203 }
204 }
205 Some("tool_result") => {
206 let is_error = b
207 .get("is_error")
208 .and_then(Value::as_bool)
209 .unwrap_or(false);
210 let prefix = if is_error { "[tool error] " } else { "[tool result] " };
211 let body = match b.get("content") {
212 Some(Value::String(s)) => s.clone(),
213 Some(Value::Array(arr)) => arr
214 .iter()
215 .filter_map(|c| c.get("text").and_then(Value::as_str))
216 .collect::<Vec<_>>()
217 .join("\n"),
218 _ => String::new(),
219 };
220 parts.push(format!("{prefix}{body}"));
221 }
222 _ => {
223 if let Some(t) = b.get("text").and_then(Value::as_str) {
224 parts.push(t.to_string());
225 }
226 }
227 }
228 }
229 if parts.is_empty() {
230 None
231 } else {
232 Some(parts.join("\n"))
233 }
234 }
235 _ => None,
236 }
237 }
238
239 fn extract_assistant(
240 msg: Option<&Value>,
241 ) -> (Option<String>, Vec<ContentBlock>, Option<String>, Option<Usage>) {
242 let Some(msg) = msg else {
243 return (None, Vec::new(), None, None);
244 };
245 let model = msg
246 .get("model")
247 .and_then(Value::as_str)
248 .map(str::to_owned);
249 let stop_reason = msg
250 .get("stop_reason")
251 .and_then(Value::as_str)
252 .map(str::to_owned);
253 let usage = msg
254 .get("usage")
255 .cloned()
256 .and_then(|v| serde_json::from_value::<Usage>(v).ok());
257 let blocks = msg
258 .get("content")
259 .and_then(Value::as_array)
260 .map(|arr| arr.iter().filter_map(parse_content_block).collect())
261 .unwrap_or_default();
262 (model, blocks, stop_reason, usage)
263 }
264
265 fn parse_content_block(block: &Value) -> Option<ContentBlock> {
266 let kind = block.get("type").and_then(Value::as_str)?;
267 match kind {
268 "text" => {
269 let text = block.get("text").and_then(Value::as_str)?.to_string();
270 Some(ContentBlock::Text { text })
271 }
272 "thinking" => {
273 // Observed field is `thinking` not `text`.
274 let text = block
275 .get("thinking")
276 .and_then(Value::as_str)
277 .or_else(|| block.get("text").and_then(Value::as_str))
278 .unwrap_or("")
279 .to_string();
280 Some(ContentBlock::Thinking { text })
281 }
282 "tool_use" => {
283 let id = block.get("id").and_then(Value::as_str)?.to_string();
284 let name = block.get("name").and_then(Value::as_str)?.to_string();
285 let input = block.get("input").cloned().unwrap_or(Value::Null);
286 Some(ContentBlock::ToolUse { id, name, input })
287 }
288 "tool_result" => {
289 // Rare inside assistant content but handle it anyway.
290 let tool_use_id = block
291 .get("tool_use_id")
292 .and_then(Value::as_str)?
293 .to_string();
294 let content = match block.get("content") {
295 Some(Value::String(s)) => s.clone(),
296 Some(Value::Array(arr)) => arr
297 .iter()
298 .filter_map(|c| c.get("text").and_then(Value::as_str))
299 .collect::<Vec<_>>()
300 .join("\n"),
301 _ => String::new(),
302 };
303 let is_error = block
304 .get("is_error")
305 .and_then(Value::as_bool)
306 .unwrap_or(false);
307 Some(ContentBlock::ToolResult {
308 tool_use_id,
309 content,
310 is_error,
311 })
312 }
313 _ => None,
314 }
315 }
316
317 fn extract_system_text(ev: &RawEvent) -> String {
318 // `system` events in the wild carry hookInfos, stopReason, etc. as
319 // top-level fields rather than a tidy `text`. Fall back to subtype
320 // as a label.
321 ev.subtype
322 .clone()
323 .unwrap_or_else(|| "system notice".to_string())
324 }
325
326 fn extract_attachment(att: Option<&Value>) -> (String, Option<String>, String) {
327 let Some(att) = att else {
328 return ("unknown".into(), None, String::new());
329 };
330 let attachment_type = att
331 .get("type")
332 .and_then(Value::as_str)
333 .unwrap_or("attachment")
334 .to_string();
335 let hook_name = att
336 .get("hookName")
337 .and_then(Value::as_str)
338 .map(str::to_owned);
339 // Prefer stdout, then content, then command.
340 let text = att
341 .get("stdout")
342 .and_then(Value::as_str)
343 .filter(|s| !s.is_empty())
344 .or_else(|| att.get("content").and_then(Value::as_str))
345 .or_else(|| att.get("command").and_then(Value::as_str))
346 .unwrap_or("")
347 .to_string();
348 (attachment_type, hook_name, text)
349 }
350
351 /// Best-effort raw event → Value for Unknown variant. Loses the fields
352 /// we didn't carry on RawEvent but preserves the recognisable shape.
353 fn raw_event_to_value(ev: &RawEvent) -> Value {
354 serde_json::json!({
355 "type": ev.kind,
356 "uuid": ev.uuid,
357 "parentUuid": ev.parent_uuid,
358 "timestamp": ev.timestamp.map(|t: DateTime<Utc>| t.to_rfc3339()),
359 "cwd": ev.cwd,
360 "subtype": ev.subtype,
361 })
362 }
363
364 #[cfg(test)]
365 mod tests {
366 use super::*;
367 use std::io::Write;
368 use tempfile::tempdir;
369
370 fn write_fixture(path: &Path, lines: &[&str]) {
371 let mut f = File::create(path).unwrap();
372 for line in lines {
373 writeln!(f, "{line}").unwrap();
374 }
375 }
376
377 #[test]
378 fn reads_user_assistant_pair() {
379 let tmp = tempdir().unwrap();
380 let path = tmp.path().join("s.jsonl");
381 write_fixture(
382 &path,
383 &[
384 r#"{"type":"user","uuid":"u1","timestamp":"2026-04-11T00:55:35.000Z","cwd":"/Users/me/repo","sessionId":"abc","version":"2.1.101","gitBranch":"main","message":{"role":"user","content":"hello"}}"#,
385 r#"{"type":"assistant","uuid":"u2","timestamp":"2026-04-11T00:55:40.000Z","cwd":"/Users/me/repo","sessionId":"abc","version":"2.1.101","gitBranch":"main","message":{"model":"claude-opus-4-6","content":[{"type":"text","text":"hi there"}],"usage":{"input_tokens":5,"output_tokens":10,"cache_creation_input_tokens":0,"cache_read_input_tokens":0}}}"#,
386 ],
387 );
388
389 let detail = read_session(&path, "-Users-me-repo").unwrap();
390 assert_eq!(detail.messages.len(), 2);
391 match &detail.messages[0] {
392 Message::User { text, is_meta, .. } => {
393 assert_eq!(text, "hello");
394 assert!(!is_meta);
395 }
396 other => panic!("expected user, got {other:?}"),
397 }
398 match &detail.messages[1] {
399 Message::Assistant {
400 model,
401 blocks,
402 usage,
403 ..
404 } => {
405 assert_eq!(model.as_deref(), Some("claude-opus-4-6"));
406 assert_eq!(blocks.len(), 1);
407 assert!(matches!(&blocks[0], ContentBlock::Text { text } if text == "hi there"));
408 assert_eq!(usage.as_ref().unwrap().output_tokens, 10);
409 }
410 other => panic!("expected assistant, got {other:?}"),
411 }
412 }
413
414 #[test]
415 fn parses_tool_use_blocks() {
416 let tmp = tempdir().unwrap();
417 let path = tmp.path().join("s.jsonl");
418 write_fixture(
419 &path,
420 &[
421 r#"{"type":"assistant","uuid":"u1","timestamp":"2026-04-11T00:55:40.000Z","sessionId":"abc","message":{"model":"claude-opus-4-6","content":[{"type":"thinking","thinking":"hmm"},{"type":"text","text":"Let me check."},{"type":"tool_use","id":"tu_1","name":"Read","input":{"file_path":"/tmp/x"}}]}}"#,
422 ],
423 );
424
425 let detail = read_session(&path, "-Users-me-repo").unwrap();
426 let Message::Assistant { blocks, .. } = &detail.messages[0] else {
427 panic!("expected assistant");
428 };
429 assert_eq!(blocks.len(), 3);
430 assert!(matches!(&blocks[0], ContentBlock::Thinking { text } if text == "hmm"));
431 assert!(matches!(&blocks[1], ContentBlock::Text { .. }));
432 assert!(matches!(&blocks[2], ContentBlock::ToolUse { name, .. } if name == "Read"));
433 }
434
435 #[test]
436 fn tool_result_inside_user_content_folds_to_text() {
437 let tmp = tempdir().unwrap();
438 let path = tmp.path().join("s.jsonl");
439 write_fixture(
440 &path,
441 &[
442 r#"{"type":"user","uuid":"u1","timestamp":"2026-04-11T00:55:35.000Z","sessionId":"abc","isMeta":true,"message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"tu_1","content":"file contents here","is_error":false}]}}"#,
443 ],
444 );
445 let detail = read_session(&path, "-Users-me-repo").unwrap();
446 match &detail.messages[0] {
447 Message::User { text, is_meta, .. } => {
448 assert!(*is_meta);
449 assert!(text.starts_with("[tool result]"));
450 assert!(text.contains("file contents here"));
451 }
452 other => panic!("expected user/meta, got {other:?}"),
453 }
454 }
455
456 #[test]
457 fn filters_session_metadata_events() {
458 let tmp = tempdir().unwrap();
459 let path = tmp.path().join("s.jsonl");
460 write_fixture(
461 &path,
462 &[
463 r#"{"type":"permission-mode","permissionMode":"default","sessionId":"abc"}"#,
464 r#"{"type":"custom-title","customTitle":"t","sessionId":"abc"}"#,
465 r#"{"type":"agent-name","agentName":"a","sessionId":"abc"}"#,
466 r#"{"type":"file-history-snapshot","messageId":"m","snapshot":{}}"#,
467 r#"{"type":"queue-operation","operation":"enqueue","sessionId":"abc"}"#,
468 r#"{"type":"user","uuid":"u1","timestamp":"2026-04-11T00:55:35.000Z","sessionId":"abc","message":{"role":"user","content":"hi"}}"#,
469 ],
470 );
471 let detail = read_session(&path, "-Users-me-repo").unwrap();
472 assert_eq!(detail.messages.len(), 1);
473 assert!(matches!(detail.messages[0], Message::User { .. }));
474 }
475
476 #[test]
477 fn unknown_event_becomes_unknown_variant() {
478 let tmp = tempdir().unwrap();
479 let path = tmp.path().join("s.jsonl");
480 write_fixture(
481 &path,
482 &[
483 r#"{"type":"brand-new-event-kind","uuid":"u1","timestamp":"2026-04-11T00:55:35.000Z","sessionId":"abc"}"#,
484 ],
485 );
486 let detail = read_session(&path, "-Users-me-repo").unwrap();
487 match &detail.messages[0] {
488 Message::Unknown { raw_type, .. } => {
489 assert_eq!(raw_type, "brand-new-event-kind");
490 }
491 other => panic!("expected unknown, got {other:?}"),
492 }
493 }
494
495 #[test]
496 fn skips_sidechain_events() {
497 let tmp = tempdir().unwrap();
498 let path = tmp.path().join("s.jsonl");
499 write_fixture(
500 &path,
501 &[
502 r#"{"type":"user","uuid":"u1","timestamp":"2026-04-11T00:55:35.000Z","isSidechain":true,"sessionId":"abc","message":{"role":"user","content":"subagent"}}"#,
503 r#"{"type":"user","uuid":"u2","timestamp":"2026-04-11T00:55:36.000Z","isSidechain":false,"sessionId":"abc","message":{"role":"user","content":"main"}}"#,
504 ],
505 );
506 let detail = read_session(&path, "-Users-me-repo").unwrap();
507 assert_eq!(detail.messages.len(), 1);
508 let Message::User { text, .. } = &detail.messages[0] else {
509 panic!("expected user");
510 };
511 assert_eq!(text, "main");
512 }
513
514 #[test]
515 fn limit_returns_tail_and_preserves_order() {
516 let tmp = tempdir().unwrap();
517 let path = tmp.path().join("s.jsonl");
518 let mut lines: Vec<String> = Vec::new();
519 for i in 0..10 {
520 lines.push(format!(
521 "{{\"type\":\"user\",\"uuid\":\"u{i}\",\"timestamp\":\"2026-04-11T00:55:{:02}.000Z\",\"sessionId\":\"abc\",\"message\":{{\"role\":\"user\",\"content\":\"msg{i}\"}}}}",
522 i
523 ));
524 }
525 write_fixture(
526 &path,
527 &lines.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
528 );
529
530 let detail = read_session_limited(&path, "-Users-me-repo", Some(3)).unwrap();
531 assert_eq!(detail.messages.len(), 3);
532 // Summary still reflects total message count.
533 assert_eq!(detail.summary.message_count, 10);
534 let texts: Vec<&str> = detail
535 .messages
536 .iter()
537 .filter_map(|m| match m {
538 Message::User { text, .. } => Some(text.as_str()),
539 _ => None,
540 })
541 .collect();
542 assert_eq!(texts, vec!["msg7", "msg8", "msg9"]);
543 }
544
545 #[test]
546 fn limit_zero_is_treated_as_unlimited() {
547 // 0 is a weird edge case; we treat it as "no cap" so a
548 // caller mishandling its limit arg still gets data.
549 let tmp = tempdir().unwrap();
550 let path = tmp.path().join("s.jsonl");
551 write_fixture(
552 &path,
553 &[
554 r#"{"type":"user","uuid":"u1","timestamp":"2026-04-11T00:55:35.000Z","sessionId":"abc","message":{"role":"user","content":"hi"}}"#,
555 ],
556 );
557 let detail = read_session_limited(&path, "-Users-me-repo", Some(0)).unwrap();
558 assert_eq!(detail.messages.len(), 1);
559 }
560
561 #[test]
562 fn survives_partial_last_line() {
563 let tmp = tempdir().unwrap();
564 let path = tmp.path().join("s.jsonl");
565 let mut f = File::create(&path).unwrap();
566 writeln!(f, r#"{{"type":"user","uuid":"u1","timestamp":"2026-04-11T00:55:35.000Z","sessionId":"abc","message":{{"role":"user","content":"valid"}}}}"#).unwrap();
567 f.write_all(br#"{"type":"assistant","uuid":"u2","timestamp":"2026-04"#).unwrap();
568 drop(f);
569
570 let detail = read_session(&path, "-Users-me-repo").unwrap();
571 assert_eq!(detail.messages.len(), 1);
572 }
573 }
574