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::McpServerReady { name, tools } => serde_json::json!({
206            "name": name,
207            "tools": tools.iter().map(|tool| serde_json::json!({
208                "name": tool.name,
209                "description": tool.description,
210                "input_schema": tool.input_schema,
211            })).collect::<Vec<_>>(),
212        }),
213        Msg::McpServerErrored { name, reason } => serde_json::json!({
214            "name": name,
215            "reason": reason,
216        }),
217        Msg::McpServerStopped { name } => serde_json::json!({"name": name}),
218        Msg::InstructionsChanged(loaded) => match loaded {
219            Some(loaded) => serde_json::json!({
220                "path": loaded.path,
221                "byte_len": loaded.byte_len,
222                "truncated": loaded.truncated,
223            }),
224            None => serde_json::json!({"path": null}),
225        },
226        Msg::SessionSaved => serde_json::json!({}),
227        Msg::ConversationLoaded(history) => serde_json::json!({
228            "id": history.id,
229            "message_count": history.messages.len(),
230            "title": history.title,
231        }),
232        Msg::ConversationsListed(summaries) => serde_json::json!({
233            "count": summaries.len(),
234            "ids": summaries.iter().map(|summary| summary.id.as_str()).collect::<Vec<_>>(),
235        }),
236        Msg::RuntimeTasksListed(tasks) => serde_json::json!({
237            "count": tasks.len(),
238            "ids": tasks.iter().map(|task| task.id.as_str()).collect::<Vec<_>>(),
239        }),
240        Msg::RuntimeTaskLoaded { task, events } => serde_json::json!({
241            "id": task.as_ref().map(|task| task.id.as_str()),
242            "found": task.is_some(),
243            "event_count": events.len(),
244        }),
245        Msg::RuntimeProcessesListed(processes) => serde_json::json!({
246            "count": processes.len(),
247            "ids": processes.iter().map(|process| process.id.as_str()).collect::<Vec<_>>(),
248        }),
249        Msg::RuntimeText(text) => serde_json::json!({"chars": text.len()}),
250        Msg::RuntimeApprovalsListed(approvals) => serde_json::json!({
251            "count": approvals.len(),
252            "ids": approvals.iter().map(|approval| approval.id.as_str()).collect::<Vec<_>>(),
253        }),
254        Msg::RuntimeCheckpointsListed(checkpoints) => serde_json::json!({
255            "count": checkpoints.len(),
256            "ids": checkpoints.iter().map(|checkpoint| checkpoint.id.as_str()).collect::<Vec<_>>(),
257        }),
258        Msg::RuntimeMemoryListed(memory) => serde_json::json!({
259            "count": memory.len(),
260            "ids": memory.iter().map(|entry| entry.id.as_str()).collect::<Vec<_>>(),
261        }),
262        Msg::RuntimePluginsListed(plugins) => serde_json::json!({
263            "count": plugins.len(),
264            "ids": plugins.iter().map(|plugin| plugin.id.as_str()).collect::<Vec<_>>(),
265        }),
266        Msg::ModelPullFinished { model } => serde_json::json!({"model": model}),
267        Msg::ModelPullProgress(line) => serde_json::json!({"line": line}),
268        Msg::Tick => serde_json::json!({}),
269        Msg::StatusDismiss => serde_json::json!({}),
270        Msg::Resize { width, height } => serde_json::json!({
271            "width": width,
272            "height": height,
273        }),
274        Msg::TransientStatus {
275            text,
276            kind,
277            dismiss_ms,
278        } => serde_json::json!({
279            "text": text,
280            "kind": format!("{:?}", kind),
281            "dismiss_ms": dismiss_ms,
282        }),
283        Msg::MouseScroll { delta } => serde_json::json!({"delta": delta}),
284        Msg::OpenImageAt {
285            message_index,
286            image_index,
287        } => serde_json::json!({
288            "message_index": message_index,
289            "image_index": image_index,
290        }),
291    }
292}
293
294fn key_code_body(code: KeyCode) -> serde_json::Value {
295    match code {
296        KeyCode::Char(c) => serde_json::json!({"char": c.to_string()}),
297        KeyCode::F(n) => serde_json::json!({"f": n}),
298        other => serde_json::json!(format!("{:?}", other)),
299    }
300}
301
302fn slash_body(cmd: &SlashCmd) -> serde_json::Value {
303    match cmd {
304        SlashCmd::Model(model) => serde_json::json!({"command": "model", "arg": model}),
305        SlashCmd::Reasoning(level) => serde_json::json!({
306            "command": "reasoning",
307            "arg": level.map(|level| level.as_str()),
308        }),
309        SlashCmd::VisibleReasoning(arg) => {
310            serde_json::json!({"command": "visible-reasoning", "arg": arg})
311        },
312        SlashCmd::Safety(mode) => serde_json::json!({
313            "command": "safety",
314            "arg": mode.map(|m| m.as_str()),
315        }),
316        SlashCmd::Clear => serde_json::json!({"command": "clear"}),
317        SlashCmd::Save(name) => serde_json::json!({"command": "save", "arg": name}),
318        SlashCmd::Load(name) => serde_json::json!({"command": "load", "arg": name}),
319        SlashCmd::List => serde_json::json!({"command": "list"}),
320        SlashCmd::Usage => serde_json::json!({"command": "usage"}),
321        SlashCmd::Context => serde_json::json!({"command": "context"}),
322        SlashCmd::Compact(instructions) => {
323            serde_json::json!({"command": "compact", "arg": instructions})
324        },
325        SlashCmd::Doctor => serde_json::json!({"command": "doctor"}),
326        SlashCmd::Tasks => serde_json::json!({"command": "tasks"}),
327        SlashCmd::Task(id) => serde_json::json!({"command": "task", "arg": id}),
328        SlashCmd::Pause(id) => serde_json::json!({"command": "pause", "arg": id}),
329        SlashCmd::Resume(id) => serde_json::json!({"command": "resume", "arg": id}),
330        SlashCmd::Cancel(id) => serde_json::json!({"command": "cancel", "arg": id}),
331        SlashCmd::Handoff(id) => serde_json::json!({"command": "handoff", "arg": id}),
332        SlashCmd::Report(id) => serde_json::json!({"command": "report", "arg": id}),
333        SlashCmd::Processes => serde_json::json!({"command": "processes"}),
334        SlashCmd::Logs(id) => serde_json::json!({"command": "logs", "arg": id}),
335        SlashCmd::Stop(id) => serde_json::json!({"command": "stop", "arg": id}),
336        SlashCmd::Restart(id) => serde_json::json!({"command": "restart", "arg": id}),
337        SlashCmd::Open(target) => serde_json::json!({"command": "open", "arg": target}),
338        SlashCmd::Ports => serde_json::json!({"command": "ports"}),
339        SlashCmd::Approvals => serde_json::json!({"command": "approvals"}),
340        SlashCmd::Approve(id) => serde_json::json!({"command": "approve", "arg": id}),
341        SlashCmd::Deny(id) => serde_json::json!({"command": "deny", "arg": id}),
342        SlashCmd::Checkpoint(paths) => serde_json::json!({"command": "checkpoint", "arg": paths}),
343        SlashCmd::Checkpoints => serde_json::json!({"command": "checkpoints"}),
344        SlashCmd::Restore(id) => serde_json::json!({"command": "restore", "arg": id}),
345        SlashCmd::Memory => serde_json::json!({"command": "memory"}),
346        SlashCmd::MemoryEdit(input) => serde_json::json!({"command": "memory edit", "arg": input}),
347        SlashCmd::Remember(entry) => serde_json::json!({"command": "remember", "arg": entry}),
348        SlashCmd::Forget(id) => serde_json::json!({"command": "forget", "arg": id}),
349        SlashCmd::ModelInfo(model) => serde_json::json!({"command": "model-info", "arg": model}),
350        SlashCmd::Plugins => serde_json::json!({"command": "plugins"}),
351        SlashCmd::CloudSetup => serde_json::json!({"command": "cloud-setup"}),
352        SlashCmd::Help => serde_json::json!({"command": "help"}),
353        SlashCmd::Quit => serde_json::json!({"command": "quit"}),
354        SlashCmd::Unknown(name) => serde_json::json!({"command": "unknown", "name": name}),
355    }
356}
357
358fn progress_body(event: &ProgressEvent) -> serde_json::Value {
359    match event {
360        ProgressEvent::Output(text) => serde_json::json!({"type": "output", "text": text}),
361        ProgressEvent::Status(text) => serde_json::json!({"type": "status", "text": text}),
362        ProgressEvent::Bytes { done, total } => serde_json::json!({
363            "type": "bytes",
364            "done": done,
365            "total": total,
366        }),
367        ProgressEvent::Artifact {
368            mime,
369            data,
370            caption,
371        } => serde_json::json!({
372            "recordable": false,
373            "type": "artifact",
374            "mime": mime,
375            "caption": caption,
376            "size_bytes": data.len(),
377        }),
378        ProgressEvent::SubagentToolCall {
379            child_call_id,
380            tool_name,
381            phase,
382        } => serde_json::json!({
383            "type": "subagent_tool_call",
384            "child_call_id": child_call_id.0,
385            "tool_name": tool_name,
386            "phase": subagent_phase(*phase),
387        }),
388        ProgressEvent::SubagentText(text) => {
389            serde_json::json!({"type": "subagent_text", "text": text})
390        },
391    }
392}
393
394fn subagent_phase(phase: SubagentPhase) -> &'static str {
395    match phase {
396        SubagentPhase::Started => "started",
397        SubagentPhase::Finished => "finished",
398        SubagentPhase::Errored => "errored",
399    }
400}
401
402fn outcome_body(outcome: &ToolOutcome) -> serde_json::Value {
403    serde_json::json!({
404        "status": match outcome.status {
405            crate::domain::ToolStatus::Success => "success",
406            crate::domain::ToolStatus::Error => "error",
407            crate::domain::ToolStatus::Cancelled => "cancelled",
408        },
409        "summary": outcome.summary,
410        "model_content": outcome.model_content,
411        "error": outcome.error,
412        "image_count": outcome.images().map(|images| images.len()).unwrap_or(0),
413        "duration_secs": outcome.duration_secs,
414        "metadata": outcome.metadata,
415        "artifacts": outcome.artifacts,
416    })
417}
418
419fn unsupported(reason: &str) -> serde_json::Value {
420    serde_json::json!({
421        "recordable": false,
422        "reason": reason,
423    })
424}
425
426/// Read a JSONL log back. Iterates one line at a time so a huge
427/// replay doesn't allocate the whole file upfront.
428pub struct Replay {
429    lines: std::io::Lines<BufReader<File>>,
430    path: PathBuf,
431}
432
433impl Replay {
434    pub fn open(path: impl Into<PathBuf>) -> Result<Self> {
435        let path = path.into();
436        let file =
437            File::open(&path).with_context(|| format!("open {} for replay", path.display()))?;
438        Ok(Self {
439            lines: BufReader::new(file).lines(),
440            path,
441        })
442    }
443
444    pub fn path(&self) -> &Path {
445        &self.path
446    }
447}
448
449impl Iterator for Replay {
450    type Item = Result<ReplayEntry>;
451
452    fn next(&mut self) -> Option<Self::Item> {
453        let line = self.lines.next()?;
454        Some(match line {
455            Ok(raw) => serde_json::from_str::<ReplayEntry>(&raw)
456                .with_context(|| format!("parse replay line: {}", raw)),
457            Err(e) => Err(anyhow::Error::from(e)),
458        })
459    }
460}
461
462/// Parsed JSONL entry. Fields mirror what `Recorder::record_kind`
463/// writes.
464#[derive(Debug, serde::Serialize, serde::Deserialize)]
465pub struct ReplayEntry {
466    pub ts: String,
467    pub kind: String,
468    pub turn: Option<u64>,
469    pub body: serde_json::Value,
470}
471
472#[cfg(test)]
473mod tests {
474    use super::*;
475
476    fn tmpfile(name: &str) -> PathBuf {
477        let dir = std::env::temp_dir().join("mermaid_recorder_tests");
478        let _ = std::fs::create_dir_all(&dir);
479        dir.join(name)
480    }
481
482    #[test]
483    fn record_and_replay_roundtrip() {
484        let path = tmpfile("roundtrip.jsonl");
485        let _ = std::fs::remove_file(&path);
486
487        {
488            let mut r = Recorder::open(&path).expect("open");
489            r.record_kind(MsgKind::Tick, None, serde_json::json!({}))
490                .expect("record");
491            r.record_kind(
492                MsgKind::SubmitPrompt,
493                None,
494                serde_json::json!({"text": "hello"}),
495            )
496            .expect("record");
497            r.record_kind(
498                MsgKind::StreamText,
499                Some(TurnId(7)),
500                serde_json::json!({"chunk": "partial"}),
501            )
502            .expect("record");
503            r.flush().expect("flush");
504        }
505
506        let replay = Replay::open(&path).expect("open replay");
507        let entries: Vec<_> = replay.collect::<Result<_>>().expect("all parse");
508        assert_eq!(entries.len(), 3);
509        assert_eq!(entries[0].kind, "Tick");
510        assert_eq!(entries[1].body["text"], "hello");
511        assert_eq!(entries[2].turn, Some(7));
512
513        let _ = std::fs::remove_file(&path);
514    }
515
516    #[test]
517    fn replay_parses_malformed_line_as_err() {
518        let path = tmpfile("bad.jsonl");
519        std::fs::write(&path, "not-json\n").expect("write");
520        let mut replay = Replay::open(&path).expect("open");
521        let first = replay.next().expect("first entry");
522        assert!(first.is_err());
523        let _ = std::fs::remove_file(&path);
524    }
525
526    #[test]
527    fn record_creates_file_on_open() {
528        let path = tmpfile("creates.jsonl");
529        let _ = std::fs::remove_file(&path);
530        assert!(!path.exists());
531        let _ = Recorder::open(&path).expect("open");
532        assert!(path.exists());
533        let _ = std::fs::remove_file(&path);
534    }
535
536    #[test]
537    fn record_append_preserves_existing_lines() {
538        let path = tmpfile("append.jsonl");
539        let _ = std::fs::remove_file(&path);
540        {
541            let mut r = Recorder::open(&path).expect("open");
542            r.record_kind(MsgKind::Tick, None, serde_json::json!({}))
543                .expect("record");
544        }
545        {
546            let mut r = Recorder::open(&path).expect("reopen");
547            r.record_kind(MsgKind::Quit, None, serde_json::json!({}))
548                .expect("record");
549        }
550        let replay = Replay::open(&path).expect("replay");
551        let entries: Vec<_> = replay.collect::<Result<_>>().expect("all parse");
552        assert_eq!(entries.len(), 2);
553        assert_eq!(entries[0].kind, "Tick");
554        assert_eq!(entries[1].kind, "Quit");
555        let _ = std::fs::remove_file(&path);
556    }
557
558    #[test]
559    fn record_msg_body_submit_prompt_keeps_text_and_attachments() {
560        let body = record_msg_body(&crate::domain::Msg::SubmitPrompt {
561            text: "explain main.rs".to_string(),
562            attachment_ids: vec![3, 9],
563        });
564        assert_eq!(body["text"], "explain main.rs");
565        assert_eq!(body["attachment_ids"][0], 3);
566        assert_eq!(body["attachment_ids"][1], 9);
567    }
568
569    #[test]
570    fn record_msg_body_slash_model_keeps_command_and_arg() {
571        let body = record_msg_body(&crate::domain::Msg::Slash(crate::domain::SlashCmd::Model(
572            Some("anthropic/opus".to_string()),
573        )));
574        assert_eq!(body["command"], "model");
575        assert_eq!(body["arg"], "anthropic/opus");
576    }
577
578    #[test]
579    fn record_msg_body_runtime_signal_keeps_signal_name() {
580        let body = record_msg_body(&crate::domain::Msg::RuntimeSignal(
581            crate::domain::RuntimeSignal::Terminate,
582        ));
583        assert_eq!(body["signal"], "terminate");
584    }
585
586    #[test]
587    fn record_msg_body_marks_binary_paste_image_unrecordable() {
588        let body = record_msg_body(&crate::domain::Msg::Paste(crate::domain::Paste::Image {
589            bytes: vec![1, 2, 3],
590            format: "png".to_string(),
591        }));
592        assert_eq!(body["recordable"], false);
593        assert_eq!(body["type"], "image");
594        assert_eq!(body["size_bytes"], 3);
595    }
596}
mermaid_cli/app/recorder.rs

mermaid_cli/app/
recorder.rs
