mermaid_cli/domain/
run_event.rs

1//! Public, versioned event stream for `mermaid run --format ndjson`.
2//!
3//! [`RunEvent`] is the stable SDK surface: a lossy but frozen projection of the
4//! internal turn/tool/approval lifecycle ([`Msg`]) into one JSON object per
5//! line. Unlike `Msg` — which is deliberately loose so `--record`/`--replay`
6//! can grow variants freely — this wire format is a contract. The golden test
7//! in this module pins every variant's serialization so it cannot drift
8//! silently.
9//!
10//! Purity: this module is serde-only (no I/O, no wall clock), so it lives in
11//! `domain` and stays inside the purity guard. The impure emission (writing the
12//! lines to stdout) lives in the headless driver, `app::run_non_interactive`.
13
14use serde::{Deserialize, Serialize};
15
16use super::msg::Msg;
17use super::runtime::{ToolMetadata, ToolStatus};
18use crate::models::FinishReason;
19
20/// Wire-format version of the `RunEvent` stream. Bump only on a breaking change
21/// to an existing variant's shape; additive variants keep version 1.
22pub const RUN_EVENT_PROTOCOL_VERSION: u32 = 1;
23
24/// One line of the `mermaid run --format ndjson` stream. Internally tagged on
25/// `type` (snake_case), matching the house style for stable wire unions
26/// (`ToolMetadata`).
27#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
28#[serde(tag = "type", rename_all = "snake_case")]
29pub enum RunEvent {
30    /// First line of every stream: protocol version + run identity.
31    SessionStarted {
32        /// [`RUN_EVENT_PROTOCOL_VERSION`] at the time of emission.
33        protocol_version: u32,
34        /// Mermaid version that produced the stream.
35        cli_version: String,
36        /// Resolved model id driving the run.
37        model: String,
38        /// Durable runtime task id, when the run is task-backed.
39        #[serde(default)]
40        task_id: Option<String>,
41        /// Conversation/session id owning this run — pass it to
42        /// `mermaid run --resume <id>` to continue the session. Additive
43        /// (defaulted) so pre-existing recordings still deserialize.
44        #[serde(default)]
45        session_id: String,
46    },
47    /// A chunk of assistant answer text.
48    Text {
49        /// The appended text.
50        delta: String,
51    },
52    /// A chunk of model reasoning / thinking.
53    Reasoning {
54        /// The appended reasoning text.
55        delta: String,
56    },
57    /// A tool began executing (bracketed by a later [`RunEvent::ToolFinished`]
58    /// with the same `call_id`).
59    ToolStarted {
60        /// Stable per-run tool-call id.
61        call_id: String,
62    },
63    /// A tool finished. `name` is derived from the run metadata; `status` is
64    /// `success` / `error` / `cancelled`.
65    ToolFinished {
66        /// Stable per-run tool-call id (matches the `tool_started` line).
67        call_id: String,
68        /// Tool name, e.g. `execute_command`, `read_file`, `<server>/<tool>`.
69        name: String,
70        /// `success`, `error`, or `cancelled`.
71        status: String,
72        /// One-line human summary of the outcome.
73        summary: String,
74        /// Error detail, when the tool failed.
75        #[serde(default)]
76        error: Option<String>,
77        /// Present when this call is `exit_plan_mode` resolving with an
78        /// APPROVED plan — first-class plan visibility for SDK/daemon
79        /// subscribers without breaking the started/finished pairing.
80        /// Additive: absent for every other tool, and omitted from the wire
81        /// when `None`.
82        #[serde(default, skip_serializing_if = "Option::is_none")]
83        plan: Option<PlanApproved>,
84    },
85    /// A gated tool is waiting for approval. Headless runs surface this so a
86    /// supervising process can decide.
87    ApprovalRequired {
88        /// Stable per-run tool-call id.
89        call_id: String,
90        /// Tool name being gated.
91        tool: String,
92        /// Risk classification (e.g. `network`, `mutation`).
93        risk: String,
94        /// Human-readable approval prompt.
95        prompt: String,
96    },
97    /// The session task checklist changed (`task_create` / `task_update` /
98    /// a user `/tasks` edit). Full snapshot of the visible list, so consumers
99    /// never need to correlate diffs. Additive — protocol stays v1.
100    TasksUpdated {
101        /// Every non-deleted task, in creation order.
102        tasks: Vec<TaskLine>,
103        /// Count of completed tasks (numerator of "Tasks m/n").
104        completed: u32,
105        /// Count of visible tasks (denominator of "Tasks m/n").
106        total: u32,
107    },
108    /// The turn hit a recoverable or terminal upstream error.
109    Error {
110        /// Human-readable error message.
111        message: String,
112    },
113    /// A model turn completed (token usage + why it stopped, when known).
114    TurnDone {
115        /// Total tokens for the turn, when the provider reported them.
116        #[serde(default)]
117        total_tokens: Option<u64>,
118        /// Why the turn stopped (`stop`, `length`, `tool_use`, …), when known.
119        #[serde(default)]
120        stop_reason: Option<String>,
121    },
122    /// Terminal line of the stream: the aggregated run result.
123    Result {
124        /// Final assistant response text.
125        response: String,
126        /// Final reasoning text, when the model exposed any.
127        #[serde(default)]
128        reasoning: Option<String>,
129        /// Cumulative token usage for the whole run.
130        total_tokens: u64,
131        /// Errors encountered during the run (empty on success).
132        errors: Vec<String>,
133        /// Conversation/session id owning this run (same as the
134        /// `session_started` line; repeated here so a consumer that only
135        /// reads the terminal line still gets it). Additive (defaulted).
136        #[serde(default)]
137        session_id: String,
138        /// `--output-schema` runs: the response parsed as JSON, present only
139        /// when it parsed AND validated against the schema. Additive
140        /// (defaulted) — protocol_version stays 1.
141        #[serde(default, skip_serializing_if = "Option::is_none")]
142        structured_output: Option<serde_json::Value>,
143    },
144}
145
146/// Plan payload on a [`RunEvent::ToolFinished`] for `exit_plan_mode`: the
147/// approved plan's location and the execution disposition the user chose.
148#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
149pub struct PlanApproved {
150    /// Plan-file path, project-relative.
151    pub path: String,
152    /// Implementation starts immediately.
153    #[serde(default)]
154    pub start: bool,
155    /// Execution continues in a fresh conversation.
156    #[serde(default)]
157    pub fresh: bool,
158    /// Execution continues in a forked conversation.
159    #[serde(default)]
160    pub fork: bool,
161}
162
163impl RunEvent {
164    /// Project a lifecycle [`Msg`] into a public `RunEvent`, or `None` for the
165    /// many messages that have no place in the SDK stream.
166    ///
167    /// Pure and stateless: tool identity is read from the finished outcome's
168    /// metadata rather than correlated across messages, so no projector state
169    /// is required. The wildcard arm is intentional here (this is a lossy
170    /// projection, not the reducer — most `Msg`s are deliberately dropped).
171    pub fn from_msg(msg: &Msg) -> Option<RunEvent> {
172        Some(match msg {
173            Msg::StreamText { chunk, .. } => RunEvent::Text {
174                delta: chunk.clone(),
175            },
176            Msg::StreamReasoning { chunk, .. } => RunEvent::Reasoning {
177                delta: chunk.text.clone(),
178            },
179            Msg::ToolStarted { call_id, .. } => RunEvent::ToolStarted {
180                call_id: call_id.to_string(),
181            },
182            Msg::ToolFinished {
183                call_id, outcome, ..
184            } => RunEvent::ToolFinished {
185                call_id: call_id.to_string(),
186                name: tool_name(&outcome.metadata.detail),
187                status: status_str(outcome.status).to_string(),
188                summary: outcome.summary.clone(),
189                error: outcome.error.clone(),
190                plan: match &outcome.metadata.detail {
191                    ToolMetadata::Plan {
192                        path,
193                        start,
194                        fresh,
195                        fork,
196                        ..
197                    } => Some(PlanApproved {
198                        path: path.clone(),
199                        start: *start,
200                        fresh: *fresh,
201                        fork: *fork,
202                    }),
203                    _ => None,
204                },
205            },
206            Msg::ApprovalRequested {
207                call_id,
208                tool,
209                risk,
210                prompt,
211                ..
212            } => RunEvent::ApprovalRequired {
213                call_id: call_id.to_string(),
214                tool: tool.clone(),
215                risk: risk.clone(),
216                prompt: prompt.clone(),
217            },
218            Msg::TasksUpdated { store } => {
219                let (completed, total) = store.counts();
220                RunEvent::TasksUpdated {
221                    tasks: store.visible().map(TaskLine::from).collect(),
222                    completed: completed as u32,
223                    total: total as u32,
224                }
225            },
226            Msg::UpstreamError { error, .. } => RunEvent::Error {
227                message: error.message.clone(),
228            },
229            Msg::StreamDone {
230                usage, stop_reason, ..
231            } => RunEvent::TurnDone {
232                total_tokens: usage.as_ref().map(|u| u.total_tokens() as u64),
233                stop_reason: stop_reason.as_ref().map(finish_reason_str),
234            },
235            _ => return None,
236        })
237    }
238}
239
240/// One checklist row on the `tasks_updated` line. Flat and stringly-statused
241/// (wire contract — the internal enum can grow without breaking consumers).
242#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
243pub struct TaskLine {
244    pub id: u32,
245    pub subject: String,
246    /// `pending` / `in_progress` / `completed`.
247    pub status: String,
248    pub active_form: String,
249    /// Seconds spent in_progress→completed, when both stamps exist.
250    #[serde(default)]
251    pub elapsed_secs: Option<u64>,
252    /// Completion tokens attributed while in progress, when known.
253    #[serde(default)]
254    pub tokens_spent: Option<u64>,
255    /// `model` or `user` (a `/tasks add` entry).
256    #[serde(default)]
257    pub origin: Option<String>,
258}
259
260impl From<&crate::domain::TaskItem> for TaskLine {
261    fn from(task: &crate::domain::TaskItem) -> Self {
262        Self {
263            id: task.id,
264            subject: task.subject.clone(),
265            status: task.status.as_str().to_string(),
266            active_form: task.active_form.clone(),
267            elapsed_secs: task.elapsed_secs(),
268            tokens_spent: task.tokens_spent,
269            origin: Some(
270                match task.origin {
271                    crate::domain::TaskOrigin::Model => "model",
272                    crate::domain::TaskOrigin::User => "user",
273                }
274                .to_string(),
275            ),
276        }
277    }
278}
279
280/// Snake_case name of a finished tool, from its run-metadata tag. The exhaustive
281/// match doubles as a drift guard: a new `ToolMetadata` variant forces a name
282/// mapping here.
283fn tool_name(detail: &ToolMetadata) -> String {
284    match detail {
285        ToolMetadata::None => "tool".to_string(),
286        ToolMetadata::ReadFile { .. } => "read_file".to_string(),
287        ToolMetadata::WriteFile { .. } => "write_file".to_string(),
288        ToolMetadata::ApplyPatch { .. } => "apply_patch".to_string(),
289        ToolMetadata::DeleteFile { .. } => "delete_file".to_string(),
290        ToolMetadata::CreateDirectory { .. } => "create_directory".to_string(),
291        ToolMetadata::WebSearch { .. } => "web_search".to_string(),
292        ToolMetadata::WebFetch { .. } => "web_fetch".to_string(),
293        ToolMetadata::ExecuteCommand { .. } => "execute_command".to_string(),
294        ToolMetadata::ComputerUse { .. } => "computer_use".to_string(),
295        ToolMetadata::Mcp { server, tool } => format!("{server}/{tool}"),
296        ToolMetadata::Subagent { .. } => "agent".to_string(),
297        ToolMetadata::Tasks { action, .. } => format!("task_{action}"),
298        ToolMetadata::Questions { .. } => "ask_user_question".to_string(),
299        ToolMetadata::Plan { .. } => "exit_plan_mode".to_string(),
300        ToolMetadata::Custom { name, .. } => name.clone(),
301    }
302}
303
304/// Stable string form of a tool status.
305fn status_str(status: ToolStatus) -> &'static str {
306    match status {
307        ToolStatus::Success => "success",
308        ToolStatus::Error => "error",
309        ToolStatus::Cancelled => "cancelled",
310    }
311}
312
313/// Stable string form of a finish reason (mirrors the `FinishReason` serde tags).
314fn finish_reason_str(reason: &FinishReason) -> String {
315    match reason {
316        FinishReason::Stop => "stop".to_string(),
317        FinishReason::ToolUse => "tool_use".to_string(),
318        FinishReason::Length => "length".to_string(),
319        FinishReason::ContentFilter => "content_filter".to_string(),
320        FinishReason::Other(other) => other.clone(),
321    }
322}
323
324#[cfg(test)]
325mod tests {
326    use super::*;
327    use crate::domain::ids::{ToolCallId, TurnId};
328    use crate::domain::runtime::ToolRunMetadata;
329    use crate::domain::state::ToolOutcome;
330    use crate::models::TokenUsage;
331
332    /// One canonical value per variant, in declaration order.
333    fn samples() -> Vec<RunEvent> {
334        vec![
335            RunEvent::SessionStarted {
336                protocol_version: RUN_EVENT_PROTOCOL_VERSION,
337                cli_version: "9.9.9".to_string(),
338                model: "anthropic/claude-x".to_string(),
339                task_id: None,
340                session_id: "20260709_120000_000".to_string(),
341            },
342            RunEvent::Text {
343                delta: "hello".to_string(),
344            },
345            RunEvent::Reasoning {
346                delta: "thinking".to_string(),
347            },
348            RunEvent::ToolStarted {
349                call_id: "tool#3".to_string(),
350            },
351            RunEvent::ToolFinished {
352                call_id: "tool#3".to_string(),
353                name: "execute_command".to_string(),
354                status: "success".to_string(),
355                summary: "command completed".to_string(),
356                error: None,
357                plan: None,
358            },
359            RunEvent::ApprovalRequired {
360                call_id: "tool#4".to_string(),
361                tool: "execute_command".to_string(),
362                risk: "network".to_string(),
363                prompt: "Run curl?".to_string(),
364            },
365            RunEvent::TasksUpdated {
366                tasks: vec![TaskLine {
367                    id: 1,
368                    subject: "wire the broker".to_string(),
369                    status: "completed".to_string(),
370                    active_form: "wiring the broker".to_string(),
371                    elapsed_secs: Some(130),
372                    tokens_spent: Some(8400),
373                    origin: Some("model".to_string()),
374                }],
375                completed: 1,
376                total: 1,
377            },
378            RunEvent::Error {
379                message: "connection failed".to_string(),
380            },
381            RunEvent::TurnDone {
382                total_tokens: Some(1234),
383                stop_reason: Some("stop".to_string()),
384            },
385            RunEvent::Result {
386                response: "Hi there".to_string(),
387                reasoning: None,
388                total_tokens: 1234,
389                errors: vec![],
390                session_id: "20260709_120000_000".to_string(),
391                structured_output: None,
392            },
393        ]
394    }
395
396    /// The frozen wire form of each variant. The exhaustive match (no `_ =>`) is
397    /// the compile-time drift guard: a new variant forces a pinned string here,
398    /// exactly mirroring `app::recorder`'s per-`MsgKind` sample guard.
399    fn golden(ev: &RunEvent) -> &'static str {
400        match ev {
401            RunEvent::SessionStarted { .. } => {
402                r#"{"type":"session_started","protocol_version":1,"cli_version":"9.9.9","model":"anthropic/claude-x","task_id":null,"session_id":"20260709_120000_000"}"#
403            },
404            RunEvent::Text { .. } => r#"{"type":"text","delta":"hello"}"#,
405            RunEvent::Reasoning { .. } => r#"{"type":"reasoning","delta":"thinking"}"#,
406            RunEvent::ToolStarted { .. } => r#"{"type":"tool_started","call_id":"tool#3"}"#,
407            RunEvent::ToolFinished { .. } => {
408                r#"{"type":"tool_finished","call_id":"tool#3","name":"execute_command","status":"success","summary":"command completed","error":null}"#
409            },
410            RunEvent::ApprovalRequired { .. } => {
411                r#"{"type":"approval_required","call_id":"tool#4","tool":"execute_command","risk":"network","prompt":"Run curl?"}"#
412            },
413            RunEvent::TasksUpdated { .. } => {
414                r#"{"type":"tasks_updated","tasks":[{"id":1,"subject":"wire the broker","status":"completed","active_form":"wiring the broker","elapsed_secs":130,"tokens_spent":8400,"origin":"model"}],"completed":1,"total":1}"#
415            },
416            RunEvent::Error { .. } => r#"{"type":"error","message":"connection failed"}"#,
417            RunEvent::TurnDone { .. } => {
418                r#"{"type":"turn_done","total_tokens":1234,"stop_reason":"stop"}"#
419            },
420            RunEvent::Result { .. } => {
421                r#"{"type":"result","response":"Hi there","reasoning":null,"total_tokens":1234,"errors":[],"session_id":"20260709_120000_000"}"#
422            },
423        }
424    }
425
426    #[test]
427    fn wire_format_is_frozen() {
428        for ev in samples() {
429            assert_eq!(
430                serde_json::to_string(&ev).unwrap(),
431                golden(&ev),
432                "RunEvent wire format drifted for {ev:?}"
433            );
434        }
435    }
436
437    #[test]
438    fn every_variant_round_trips() {
439        for ev in samples() {
440            let wire = serde_json::to_string(&ev).unwrap();
441            let back: RunEvent = serde_json::from_str(&wire).unwrap();
442            assert_eq!(back, ev);
443        }
444    }
445
446    #[test]
447    fn every_variant_has_a_sample() {
448        // Backstop for `golden`'s compile-time guard: keep one sample per
449        // variant. Bump the count when a variant lands (and add its golden
450        // line above, which won't compile otherwise).
451        assert_eq!(samples().len(), 10);
452    }
453
454    #[test]
455    fn result_structured_output_is_additive() {
456        // Present -> serialized; absent -> omitted entirely (golden above
457        // stays frozen); old wire lines without the field still parse.
458        let ev = RunEvent::Result {
459            response: "{\"answer\":42}".to_string(),
460            reasoning: None,
461            total_tokens: 10,
462            errors: vec![],
463            session_id: "s".to_string(),
464            structured_output: Some(serde_json::json!({"answer": 42})),
465        };
466        let wire = serde_json::to_string(&ev).unwrap();
467        assert!(
468            wire.contains("\"structured_output\":{\"answer\":42}"),
469            "{wire}"
470        );
471        let back: RunEvent = serde_json::from_str(&wire).unwrap();
472        assert_eq!(back, ev);
473        let old = r#"{"type":"result","response":"x","reasoning":null,"total_tokens":1,"errors":[],"session_id":"s"}"#;
474        let parsed: RunEvent = serde_json::from_str(old).unwrap();
475        assert!(matches!(
476            parsed,
477            RunEvent::Result {
478                structured_output: None,
479                ..
480            }
481        ));
482    }
483
484    #[test]
485    fn protocol_version_is_pinned() {
486        assert_eq!(RUN_EVENT_PROTOCOL_VERSION, 1);
487    }
488
489    #[test]
490    fn tool_name_maps_each_metadata_kind() {
491        assert_eq!(tool_name(&ToolMetadata::None), "tool");
492        assert_eq!(
493            tool_name(&ToolMetadata::Mcp {
494                server: "srv".to_string(),
495                tool: "do".to_string(),
496            }),
497            "srv/do"
498        );
499        assert_eq!(
500            tool_name(&ToolMetadata::Custom {
501                name: "weird".to_string(),
502                data: serde_json::Value::Null,
503            }),
504            "weird"
505        );
506    }
507
508    #[test]
509    fn from_msg_projects_streamed_and_drops_the_rest() {
510        let text = Msg::StreamText {
511            turn: TurnId(1),
512            chunk: "hi".to_string(),
513        };
514        assert_eq!(
515            RunEvent::from_msg(&text),
516            Some(RunEvent::Text {
517                delta: "hi".to_string()
518            })
519        );
520
521        // A message with no SDK projection returns None.
522        assert_eq!(RunEvent::from_msg(&Msg::Tick), None);
523
524        let done = Msg::StreamDone {
525            turn: TurnId(1),
526            usage: Some(TokenUsage::provider(10, 20)),
527            provider_continuation: None,
528            stop_reason: Some(FinishReason::Stop),
529        };
530        assert_eq!(
531            RunEvent::from_msg(&done),
532            Some(RunEvent::TurnDone {
533                total_tokens: Some(30),
534                stop_reason: Some("stop".to_string()),
535            })
536        );
537    }
538
539    #[test]
540    fn from_msg_projects_tool_finished_with_name_from_metadata() {
541        let outcome = ToolOutcome {
542            status: ToolStatus::Success,
543            summary: "command completed".to_string(),
544            model_content: "out".to_string(),
545            error: None,
546            metadata: Box::new(ToolRunMetadata {
547                detail: ToolMetadata::ExecuteCommand {
548                    command: "ls".to_string(),
549                    working_dir: None,
550                    exit_code: Some(0),
551                    timed_out: false,
552                    background: false,
553                    stdout_lines: 1,
554                    stderr_lines: 0,
555                    detected_urls: vec![],
556                    pid: None,
557                    log_path: None,
558                    denied_by_sandbox: false,
559                },
560                ..ToolRunMetadata::default()
561            }),
562            artifacts: vec![],
563            duration_secs: Some(0.0),
564        };
565        let finished = Msg::ToolFinished {
566            turn: TurnId(1),
567            call_id: ToolCallId(3),
568            outcome,
569        };
570        assert_eq!(
571            RunEvent::from_msg(&finished),
572            Some(RunEvent::ToolFinished {
573                call_id: "tool#3".to_string(),
574                name: "execute_command".to_string(),
575                status: "success".to_string(),
576                summary: "command completed".to_string(),
577                error: None,
578                plan: None,
579            })
580        );
581    }
582
583    #[test]
584    fn approved_plan_rides_tool_finished_as_an_additive_payload() {
585        let outcome =
586            ToolOutcome::success("approved", "plan approved", 0.1).with_metadata(ToolRunMetadata {
587                detail: ToolMetadata::Plan {
588                    path: ".mermaid/plans/x.md".to_string(),
589                    body: "## Summary".to_string(),
590                    start: true,
591                    fresh: true,
592                    fork: false,
593                    model: None,
594                },
595                ..ToolRunMetadata::default()
596            });
597        let event = RunEvent::from_msg(&Msg::ToolFinished {
598            turn: TurnId(1),
599            call_id: ToolCallId(7),
600            outcome,
601        })
602        .expect("mapped");
603        let RunEvent::ToolFinished { name, plan, .. } = &event else {
604            panic!("expected ToolFinished, got {event:?}");
605        };
606        assert_eq!(name, "exit_plan_mode");
607        let plan = plan.as_ref().expect("plan payload");
608        assert_eq!(plan.path, ".mermaid/plans/x.md");
609        assert!(plan.start && plan.fresh && !plan.fork);
610        // The wire stays clean for every other tool: `plan` is omitted, not
611        // null, so existing consumers see byte-identical lines.
612        let json = serde_json::to_string(&RunEvent::ToolFinished {
613            call_id: "tool#1".to_string(),
614            name: "read_file".to_string(),
615            status: "success".to_string(),
616            summary: "read".to_string(),
617            error: None,
618            plan: None,
619        })
620        .unwrap();
621        assert!(!json.contains("\"plan\""));
622    }
623}
mermaid_cli/domain/run_event.rs

mermaid_cli/domain/
run_event.rs
