mermaid_cli/app/
recorder.rs

1//! `--record` / `--replay` support.
2//!
3//! The Elm/MVU architecture makes deterministic replay nearly free:
4//! if you capture every `Msg` the reducer sees, you can reconstruct
5//! the exact final `State` by folding over that log. This module
6//! implements both sides.
7//!
8//! Wire format: one JSON object per line (JSONL). Each object is
9//! `{ts, kind, body}`:
10//!   - `ts`: RFC3339 timestamp (for debugging, not replay).
11//!   - `kind`: `MsgKind` variant tag (matches `Msg::kind().into()`).
12//!   - `body`: best-effort structured payload from `record_msg_body`.
13//!
14//! Not every `Msg` field is safely serializable today — raw image
15//! bytes in `Paste::Image`, for example. Unsupported payloads are
16//! marked with `"recordable": false` and compact metadata. Replay is
17//! a best-effort reconstruction.
18//!
19//! For C6 this ships the on-disk shape + a `Recorder` type that
20//! writes; replay reading is available but opt-in (serialize
21//! support is wired on a subset of Msg variants that don't carry
22//! binary payloads). C9 rounds out coverage with the parity
23//! harness.
24
25use std::fs::{File, OpenOptions};
26use std::io::{BufRead, BufReader, BufWriter, Write};
27use std::path::{Path, PathBuf};
28
29use anyhow::{Context, Result};
30use chrono::Local;
31
32use crate::domain::{KeyCode, Msg, MsgKind, Paste, SlashCmd, ToolOutcome, TurnId};
33use crate::providers::{ProgressEvent, SubagentPhase};
34
35/// Append-only recorder. Writes one JSONL line per `Msg` the main
36/// loop chooses to log.
37pub struct Recorder {
38    writer: BufWriter<File>,
39    path: PathBuf,
40}
41
42impl Recorder {
43    /// Open `path` for append. Creates the file if it doesn't exist.
44    pub fn open(path: impl Into<PathBuf>) -> Result<Self> {
45        let path = path.into();
46        let file = OpenOptions::new()
47            .create(true)
48            .append(true)
49            .open(&path)
50            .with_context(|| format!("open {} for recording", path.display()))?;
51        Ok(Self {
52            writer: BufWriter::new(file),
53            path,
54        })
55    }
56
57    pub fn path(&self) -> &Path {
58        &self.path
59    }
60
61    /// Record a single `MsgKind` + optional body JSON. Meant for the
62    /// narrow subset of variants that survive round-trip. Full
63    /// Msg-graph coverage comes in C9.
64    pub fn record_kind(
65        &mut self,
66        kind: MsgKind,
67        turn: Option<TurnId>,
68        body: serde_json::Value,
69    ) -> Result<()> {
70        let entry = serde_json::json!({
71            "ts": Local::now().to_rfc3339(),
72            "kind": format!("{:?}", kind),
73            "turn": turn.map(|t| t.0),
74            "body": body,
75        });
76        writeln!(self.writer, "{}", entry).context("write jsonl line")?;
77        Ok(())
78    }
79
80    pub fn flush(&mut self) -> Result<()> {
81        self.writer.flush().context("flush recorder")
82    }
83}
84
85impl Drop for Recorder {
86    fn drop(&mut self) {
87        let _ = self.writer.flush();
88    }
89}
90
91/// Compact, best-effort JSON body for a reducer `Msg`.
92///
93/// This deliberately records useful payloads without claiming the full
94/// `Msg` graph can round-trip through serde today. Runtime-only or
95/// binary-heavy variants are marked explicitly so replay tooling can
96/// decide how to handle them.
97pub fn record_msg_body(msg: &Msg) -> serde_json::Value {
98    match msg {
99        Msg::Key(key) => serde_json::json!({
100            "code": key_code_body(key.code),
101            "modifiers": {
102                "ctrl": key.modifiers.ctrl,
103                "alt": key.modifiers.alt,
104                "shift": key.modifiers.shift,
105            },
106        }),
107        Msg::Paste(Paste::Text(text)) => serde_json::json!({
108            "type": "text",
109            "text": text,
110        }),
111        Msg::Paste(Paste::Image { bytes, format }) => serde_json::json!({
112            "recordable": false,
113            "reason": "binary paste image omitted",
114            "type": "image",
115            "format": format,
116            "size_bytes": bytes.len(),
117        }),
118        Msg::SubmitPrompt {
119            text,
120            attachment_ids,
121        } => serde_json::json!({
122            "text": text,
123            "attachment_ids": attachment_ids,
124        }),
125        Msg::Slash(cmd) => slash_body(cmd),
126        Msg::CancelTurn => serde_json::json!({}),
127        Msg::ConfirmAccepted => serde_json::json!({"accepted": true}),
128        Msg::ConfirmDeclined => serde_json::json!({"accepted": false}),
129        Msg::Quit => serde_json::json!({}),
130        Msg::RuntimeSignal(signal) => serde_json::json!({
131            "signal": signal.as_str(),
132        }),
133        Msg::StreamText { chunk, .. } => serde_json::json!({"chunk": chunk}),
134        Msg::StreamReasoning { chunk, .. } => serde_json::json!({
135            "text": chunk.text,
136            "signature": chunk.signature,
137        }),
138        Msg::StreamToolCall { call, .. } => serde_json::to_value(call)
139            .unwrap_or_else(|_| unsupported("tool call was not serializable")),
140        Msg::ContextUsageEstimated { snapshot, .. } => serde_json::json!({
141            "used_tokens": snapshot.used_tokens,
142            "max_tokens": snapshot.max_tokens,
143            "remaining_tokens": snapshot.remaining_tokens,
144            "used_percent": snapshot.used_percent,
145            "source": format!("{:?}", snapshot.source),
146            "breakdown": snapshot.breakdown.as_ref().map(|b| serde_json::json!({
147                "system_tokens": b.system_tokens,
148                "instructions_tokens": b.instructions_tokens,
149                "message_tokens": b.message_tokens,
150                "tool_schema_tokens": b.tool_schema_tokens,
151                "image_count": b.image_count,
152                "message_count": b.message_count,
153                "tool_count": b.tool_count,
154            })),
155        }),
156        Msg::CompactionFinished { result, .. } => serde_json::json!({
157            "id": result.record.id,
158            "trigger": result.record.trigger.as_str(),
159            "before_tokens": result.record.before_tokens,
160            "after_tokens": result.record.after_tokens,
161            "archived_message_count": result.record.archived_message_count,
162            "preserved_message_count": result.record.preserved_message_count,
163            "duration_secs": result.record.duration_secs,
164        }),
165        Msg::CompactionFailed {
166            trigger,
167            message,
168            kind,
169            ..
170        } => serde_json::json!({
171            "trigger": trigger.as_str(),
172            "message": message,
173            "kind": format!("{:?}", kind),
174        }),
175        Msg::StreamDone {
176            usage,
177            thinking_signature,
178            ..
179        } => serde_json::json!({
180            "usage": usage.as_ref().map(|u| serde_json::json!({
181                "prompt_tokens": u.prompt_tokens,
182                "completion_tokens": u.completion_tokens,
183                "total_tokens": u.total_tokens,
184                "cached_input_tokens": u.cached_input_tokens,
185                "cache_creation_input_tokens": u.cache_creation_input_tokens,
186                "reasoning_output_tokens": u.reasoning_output_tokens,
187                "source": format!("{:?}", u.source),
188            })),
189            "thinking_signature": thinking_signature,
190        }),
191        Msg::UpstreamError { error, .. } => serde_json::to_value(error)
192            .unwrap_or_else(|_| unsupported("upstream error was not serializable")),
193        Msg::TurnCancelled(_) => serde_json::json!({}),
194        Msg::ToolStarted { call_id, .. } => serde_json::json!({"call_id": call_id.0}),
195        Msg::ToolProgress { call_id, event, .. } => serde_json::json!({
196            "call_id": call_id.0,
197            "event": progress_body(event),
198        }),
199        Msg::ToolFinished {
200            call_id, outcome, ..
201        } => serde_json::json!({
202            "call_id": call_id.0,
203            "outcome": outcome_body(outcome),
204        }),
205        Msg::ApprovalRequested {
206            call_id,
207            tool,
208            risk,
209            ..
210        } => serde_json::json!({
211            "call_id": call_id.0,
212            "tool": tool,
213            "risk": risk,
214        }),
215        Msg::McpServerReady { name, tools } => serde_json::json!({
216            "name": name,
217            "tools": tools.iter().map(|tool| serde_json::json!({
218                "name": tool.name,
219                "description": tool.description,
220                "input_schema": tool.input_schema,
221            })).collect::<Vec<_>>(),
222        }),
223        Msg::McpServerErrored { name, reason } => serde_json::json!({
224            "name": name,
225            "reason": reason,
226        }),
227        Msg::McpServerStopped { name } => serde_json::json!({"name": name}),
228        Msg::InstructionsChanged(loaded) => match loaded {
229            Some(loaded) => serde_json::json!({
230                "path": loaded.path,
231                "byte_len": loaded.byte_len,
232                "truncated": loaded.truncated,
233            }),
234            None => serde_json::json!({"path": null}),
235        },
236        Msg::MemoryChanged(loaded) => match loaded {
237            Some(loaded) => serde_json::json!({
238                "entries": loaded.entries.len(),
239                "truncated": loaded.truncated,
240            }),
241            None => serde_json::json!({"entries": 0}),
242        },
243        Msg::SessionSaved => serde_json::json!({}),
244        Msg::ConversationLoaded(history) => serde_json::json!({
245            "id": history.id,
246            "message_count": history.messages.len(),
247            "title": history.title,
248        }),
249        Msg::ConversationsListed(summaries) => serde_json::json!({
250            "count": summaries.len(),
251            "ids": summaries.iter().map(|summary| summary.id.as_str()).collect::<Vec<_>>(),
252        }),
253        Msg::RuntimeTasksListed(tasks) => serde_json::json!({
254            "count": tasks.len(),
255            "ids": tasks.iter().map(|task| task.id.as_str()).collect::<Vec<_>>(),
256        }),
257        Msg::RuntimeTaskLoaded { task, events } => serde_json::json!({
258            "id": task.as_ref().map(|task| task.id.as_str()),
259            "found": task.is_some(),
260            "event_count": events.len(),
261        }),
262        Msg::RuntimeProcessesListed(processes) => serde_json::json!({
263            "count": processes.len(),
264            "ids": processes.iter().map(|process| process.id.as_str()).collect::<Vec<_>>(),
265        }),
266        Msg::RuntimeText(text) => serde_json::json!({"chars": text.len()}),
267        Msg::RuntimeApprovalsListed(approvals) => serde_json::json!({
268            "count": approvals.len(),
269            "ids": approvals.iter().map(|approval| approval.id.as_str()).collect::<Vec<_>>(),
270        }),
271        Msg::RuntimeCheckpointsListed(checkpoints) => serde_json::json!({
272            "count": checkpoints.len(),
273            "ids": checkpoints.iter().map(|checkpoint| checkpoint.id.as_str()).collect::<Vec<_>>(),
274        }),
275        Msg::RuntimePluginsListed(plugins) => serde_json::json!({
276            "count": plugins.len(),
277            "ids": plugins.iter().map(|plugin| plugin.id.as_str()).collect::<Vec<_>>(),
278        }),
279        Msg::ModelPullFinished { model } => serde_json::json!({"model": model}),
280        Msg::ModelPullProgress(line) => serde_json::json!({"line": line}),
281        Msg::Tick => serde_json::json!({}),
282        Msg::StatusDismiss => serde_json::json!({}),
283        Msg::Resize { width, height } => serde_json::json!({
284            "width": width,
285            "height": height,
286        }),
287        Msg::TransientStatus {
288            text,
289            kind,
290            dismiss_ms,
291        } => serde_json::json!({
292            "text": text,
293            "kind": format!("{:?}", kind),
294            "dismiss_ms": dismiss_ms,
295        }),
296        Msg::MouseScroll { delta } => serde_json::json!({"delta": delta}),
297        Msg::OpenImageAt {
298            message_index,
299            image_index,
300        } => serde_json::json!({
301            "message_index": message_index,
302            "image_index": image_index,
303        }),
304    }
305}
306
307fn key_code_body(code: KeyCode) -> serde_json::Value {
308    match code {
309        KeyCode::Char(c) => serde_json::json!({"char": c.to_string()}),
310        KeyCode::F(n) => serde_json::json!({"f": n}),
311        other => serde_json::json!(format!("{:?}", other)),
312    }
313}
314
315fn slash_body(cmd: &SlashCmd) -> serde_json::Value {
316    match cmd {
317        SlashCmd::Model(model) => serde_json::json!({"command": "model", "arg": model}),
318        SlashCmd::Reasoning(level) => serde_json::json!({
319            "command": "reasoning",
320            "arg": level.map(|level| level.as_str()),
321        }),
322        SlashCmd::VisibleReasoning(arg) => {
323            serde_json::json!({"command": "visible-reasoning", "arg": arg})
324        },
325        SlashCmd::Safety(mode) => serde_json::json!({
326            "command": "safety",
327            "arg": mode.map(|m| m.as_str()),
328        }),
329        SlashCmd::Clear => serde_json::json!({"command": "clear"}),
330        SlashCmd::Save(name) => serde_json::json!({"command": "save", "arg": name}),
331        SlashCmd::Load(name) => serde_json::json!({"command": "load", "arg": name}),
332        SlashCmd::List => serde_json::json!({"command": "list"}),
333        SlashCmd::Usage => serde_json::json!({"command": "usage"}),
334        SlashCmd::Context => serde_json::json!({"command": "context"}),
335        SlashCmd::Compact(instructions) => {
336            serde_json::json!({"command": "compact", "arg": instructions})
337        },
338        SlashCmd::Memory => serde_json::json!({"command": "memory"}),
339        SlashCmd::Remember(text) => serde_json::json!({"command": "remember", "arg": text}),
340        SlashCmd::Forget(id) => serde_json::json!({"command": "forget", "arg": id}),
341        SlashCmd::ConsolidateMemory => serde_json::json!({"command": "consolidate-memory"}),
342        SlashCmd::Doctor => serde_json::json!({"command": "doctor"}),
343        SlashCmd::Tasks => serde_json::json!({"command": "tasks"}),
344        SlashCmd::Task(id) => serde_json::json!({"command": "task", "arg": id}),
345        SlashCmd::Pause(id) => serde_json::json!({"command": "pause", "arg": id}),
346        SlashCmd::Resume(id) => serde_json::json!({"command": "resume", "arg": id}),
347        SlashCmd::Cancel(id) => serde_json::json!({"command": "cancel", "arg": id}),
348        SlashCmd::Handoff(id) => serde_json::json!({"command": "handoff", "arg": id}),
349        SlashCmd::Report(id) => serde_json::json!({"command": "report", "arg": id}),
350        SlashCmd::Processes => serde_json::json!({"command": "processes"}),
351        SlashCmd::Logs(id) => serde_json::json!({"command": "logs", "arg": id}),
352        SlashCmd::Stop(id) => serde_json::json!({"command": "stop", "arg": id}),
353        SlashCmd::Restart(id) => serde_json::json!({"command": "restart", "arg": id}),
354        SlashCmd::Open(target) => serde_json::json!({"command": "open", "arg": target}),
355        SlashCmd::Ports => serde_json::json!({"command": "ports"}),
356        SlashCmd::Approvals => serde_json::json!({"command": "approvals"}),
357        SlashCmd::Approve(id) => serde_json::json!({"command": "approve", "arg": id}),
358        SlashCmd::Deny(id) => serde_json::json!({"command": "deny", "arg": id}),
359        SlashCmd::Checkpoint(paths) => serde_json::json!({"command": "checkpoint", "arg": paths}),
360        SlashCmd::Checkpoints => serde_json::json!({"command": "checkpoints"}),
361        SlashCmd::Restore(id) => serde_json::json!({"command": "restore", "arg": id}),
362        SlashCmd::ModelInfo(model) => serde_json::json!({"command": "model-info", "arg": model}),
363        SlashCmd::Plugins => serde_json::json!({"command": "plugins"}),
364        SlashCmd::CloudSetup => serde_json::json!({"command": "cloud-setup"}),
365        SlashCmd::Help => serde_json::json!({"command": "help"}),
366        SlashCmd::Quit => serde_json::json!({"command": "quit"}),
367        SlashCmd::Unknown(name) => serde_json::json!({"command": "unknown", "name": name}),
368    }
369}
370
371fn progress_body(event: &ProgressEvent) -> serde_json::Value {
372    match event {
373        ProgressEvent::Output(text) => serde_json::json!({"type": "output", "text": text}),
374        ProgressEvent::Status(text) => serde_json::json!({"type": "status", "text": text}),
375        ProgressEvent::Bytes { done, total } => serde_json::json!({
376            "type": "bytes",
377            "done": done,
378            "total": total,
379        }),
380        ProgressEvent::Artifact {
381            mime,
382            data,
383            caption,
384        } => serde_json::json!({
385            "recordable": false,
386            "type": "artifact",
387            "mime": mime,
388            "caption": caption,
389            "size_bytes": data.len(),
390        }),
391        ProgressEvent::SubagentToolCall {
392            child_call_id,
393            tool_name,
394            phase,
395        } => serde_json::json!({
396            "type": "subagent_tool_call",
397            "child_call_id": child_call_id.0,
398            "tool_name": tool_name,
399            "phase": subagent_phase(*phase),
400        }),
401        ProgressEvent::SubagentText(text) => {
402            serde_json::json!({"type": "subagent_text", "text": text})
403        },
404    }
405}
406
407fn subagent_phase(phase: SubagentPhase) -> &'static str {
408    match phase {
409        SubagentPhase::Started => "started",
410        SubagentPhase::Finished => "finished",
411        SubagentPhase::Errored => "errored",
412    }
413}
414
415fn outcome_body(outcome: &ToolOutcome) -> serde_json::Value {
416    serde_json::json!({
417        "status": match outcome.status {
418            crate::domain::ToolStatus::Success => "success",
419            crate::domain::ToolStatus::Error => "error",
420            crate::domain::ToolStatus::Cancelled => "cancelled",
421        },
422        "summary": outcome.summary,
423        "model_content": outcome.model_content,
424        "error": outcome.error,
425        "image_count": outcome.images().map(|images| images.len()).unwrap_or(0),
426        "duration_secs": outcome.duration_secs,
427        "metadata": outcome.metadata,
428        "artifacts": outcome.artifacts,
429    })
430}
431
432fn unsupported(reason: &str) -> serde_json::Value {
433    serde_json::json!({
434        "recordable": false,
435        "reason": reason,
436    })
437}
438
439/// Read a JSONL log back. Iterates one line at a time so a huge
440/// replay doesn't allocate the whole file upfront.
441pub struct Replay {
442    lines: std::io::Lines<BufReader<File>>,
443    path: PathBuf,
444}
445
446impl Replay {
447    pub fn open(path: impl Into<PathBuf>) -> Result<Self> {
448        let path = path.into();
449        let file =
450            File::open(&path).with_context(|| format!("open {} for replay", path.display()))?;
451        Ok(Self {
452            lines: BufReader::new(file).lines(),
453            path,
454        })
455    }
456
457    pub fn path(&self) -> &Path {
458        &self.path
459    }
460}
461
462impl Iterator for Replay {
463    type Item = Result<ReplayEntry>;
464
465    fn next(&mut self) -> Option<Self::Item> {
466        let line = self.lines.next()?;
467        Some(match line {
468            Ok(raw) => serde_json::from_str::<ReplayEntry>(&raw)
469                .with_context(|| format!("parse replay line: {}", raw)),
470            Err(e) => Err(anyhow::Error::from(e)),
471        })
472    }
473}
474
475/// Parsed JSONL entry. Fields mirror what `Recorder::record_kind`
476/// writes.
477#[derive(Debug, serde::Serialize, serde::Deserialize)]
478pub struct ReplayEntry {
479    pub ts: String,
480    pub kind: String,
481    pub turn: Option<u64>,
482    pub body: serde_json::Value,
483}
484
485#[cfg(test)]
486mod tests {
487    use super::*;
488
489    fn tmpfile(name: &str) -> PathBuf {
490        let dir = std::env::temp_dir().join("mermaid_recorder_tests");
491        let _ = std::fs::create_dir_all(&dir);
492        dir.join(name)
493    }
494
495    #[test]
496    fn record_and_replay_roundtrip() {
497        let path = tmpfile("roundtrip.jsonl");
498        let _ = std::fs::remove_file(&path);
499
500        {
501            let mut r = Recorder::open(&path).expect("open");
502            r.record_kind(MsgKind::Tick, None, serde_json::json!({}))
503                .expect("record");
504            r.record_kind(
505                MsgKind::SubmitPrompt,
506                None,
507                serde_json::json!({"text": "hello"}),
508            )
509            .expect("record");
510            r.record_kind(
511                MsgKind::StreamText,
512                Some(TurnId(7)),
513                serde_json::json!({"chunk": "partial"}),
514            )
515            .expect("record");
516            r.flush().expect("flush");
517        }
518
519        let replay = Replay::open(&path).expect("open replay");
520        let entries: Vec<_> = replay.collect::<Result<_>>().expect("all parse");
521        assert_eq!(entries.len(), 3);
522        assert_eq!(entries[0].kind, "Tick");
523        assert_eq!(entries[1].body["text"], "hello");
524        assert_eq!(entries[2].turn, Some(7));
525
526        let _ = std::fs::remove_file(&path);
527    }
528
529    #[test]
530    fn replay_parses_malformed_line_as_err() {
531        let path = tmpfile("bad.jsonl");
532        std::fs::write(&path, "not-json\n").expect("write");
533        let mut replay = Replay::open(&path).expect("open");
534        let first = replay.next().expect("first entry");
535        assert!(first.is_err());
536        let _ = std::fs::remove_file(&path);
537    }
538
539    #[test]
540    fn record_creates_file_on_open() {
541        let path = tmpfile("creates.jsonl");
542        let _ = std::fs::remove_file(&path);
543        assert!(!path.exists());
544        let _ = Recorder::open(&path).expect("open");
545        assert!(path.exists());
546        let _ = std::fs::remove_file(&path);
547    }
548
549    #[test]
550    fn record_append_preserves_existing_lines() {
551        let path = tmpfile("append.jsonl");
552        let _ = std::fs::remove_file(&path);
553        {
554            let mut r = Recorder::open(&path).expect("open");
555            r.record_kind(MsgKind::Tick, None, serde_json::json!({}))
556                .expect("record");
557        }
558        {
559            let mut r = Recorder::open(&path).expect("reopen");
560            r.record_kind(MsgKind::Quit, None, serde_json::json!({}))
561                .expect("record");
562        }
563        let replay = Replay::open(&path).expect("replay");
564        let entries: Vec<_> = replay.collect::<Result<_>>().expect("all parse");
565        assert_eq!(entries.len(), 2);
566        assert_eq!(entries[0].kind, "Tick");
567        assert_eq!(entries[1].kind, "Quit");
568        let _ = std::fs::remove_file(&path);
569    }
570
571    #[test]
572    fn record_msg_body_submit_prompt_keeps_text_and_attachments() {
573        let body = record_msg_body(&crate::domain::Msg::SubmitPrompt {
574            text: "explain main.rs".to_string(),
575            attachment_ids: vec![3, 9],
576        });
577        assert_eq!(body["text"], "explain main.rs");
578        assert_eq!(body["attachment_ids"][0], 3);
579        assert_eq!(body["attachment_ids"][1], 9);
580    }
581
582    #[test]
583    fn record_msg_body_slash_model_keeps_command_and_arg() {
584        let body = record_msg_body(&crate::domain::Msg::Slash(crate::domain::SlashCmd::Model(
585            Some("anthropic/opus".to_string()),
586        )));
587        assert_eq!(body["command"], "model");
588        assert_eq!(body["arg"], "anthropic/opus");
589    }
590
591    #[test]
592    fn record_msg_body_runtime_signal_keeps_signal_name() {
593        let body = record_msg_body(&crate::domain::Msg::RuntimeSignal(
594            crate::domain::RuntimeSignal::Terminate,
595        ));
596        assert_eq!(body["signal"], "terminate");
597    }
598
599    #[test]
600    fn record_msg_body_marks_binary_paste_image_unrecordable() {
601        let body = record_msg_body(&crate::domain::Msg::Paste(crate::domain::Paste::Image {
602            bytes: vec![1, 2, 3],
603            format: "png".to_string(),
604        }));
605        assert_eq!(body["recordable"], false);
606        assert_eq!(body["type"], "image");
607        assert_eq!(body["size_bytes"], 3);
608    }
609}
mermaid_cli/app/recorder.rs

mermaid_cli/app/
recorder.rs
