muse_codes/io/
mod.rs

1//! Typed models of the `muse exec --json` JSONL event stream.
2//!
3//! Every stdout line is one [`MuseRecord`] — an event-sourced journal
4//! envelope with a lazily-typed payload. The envelope is fully typed; the
5//! payload stays raw JSON on the record (so round-trips are byte-faithful
6//! and unknown future payload types survive) and is lifted into a
7//! [`MusePayload`] on demand via [`MuseRecord::typed_payload`].
8//!
9//! Shapes in this module are derived from **captured real output** of
10//! Muse Code (see `test_cases/*.jsonl`), not from documentation — the wire
11//! is the contract. Payload types not yet observed (the journal also
12//! records approvals, edits, and subagent lifecycle under a live provider)
13//! deserialize as [`MusePayload::Unknown`] rather than failing.
14
15use serde::{Deserialize, Serialize};
16use serde_json::Value;
17
18/// One line of the `muse exec --json` stream: the journal envelope.
19#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
20pub struct MuseRecord {
21    /// Envelope schema version (observed: `1`).
22    pub schema_version: u32,
23    /// Unique record id (UUIDv7-style, monotonic within a stream).
24    pub id: String,
25    /// The stream this record belongs to.
26    pub stream: StreamRef,
27    /// 1-based position within `stream`.
28    pub sequence: u64,
29    /// Microseconds since the Unix epoch.
30    pub recorded_at: u64,
31    pub record_type: RecordType,
32    pub durability: Durability,
33    /// Id of the command that caused this record.
34    pub causation_id: String,
35    /// Dotted payload discriminator, e.g. `run.output.delta`.
36    pub payload_type: String,
37    /// Version of the payload's own schema (observed: `1`).
38    pub payload_schema_version: u32,
39    /// Raw payload — lift with [`MuseRecord::typed_payload`].
40    pub payload: Value,
41}
42
43impl MuseRecord {
44    /// Parse the payload into its typed form based on `payload_type`.
45    ///
46    /// Unknown payload types return [`MusePayload::Unknown`] carrying the
47    /// raw value; a payload that fails to match its expected shape is a
48    /// deserialization error (wire drift worth surfacing, not masking).
49    pub fn typed_payload(&self) -> serde_json::Result<MusePayload> {
50        MusePayload::from_parts(&self.payload_type, self.payload.clone())
51    }
52}
53
54/// Reference to a journal stream.
55#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
56pub struct StreamRef {
57    pub kind: StreamKind,
58    pub id: String,
59}
60
61/// Journal stream classes.
62#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
63#[serde(rename_all = "snake_case")]
64pub enum StreamKind {
65    Session,
66    Run,
67    Task,
68}
69
70/// Journal record classes.
71#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
72#[serde(rename_all = "snake_case")]
73pub enum RecordType {
74    /// Replay-exact state reconciliation (e.g. command acceptance).
75    Reconciliation,
76    /// Durable domain event.
77    Event,
78    /// Ephemeral progress/status (e.g. output deltas).
79    Status,
80}
81
82/// Whether the record survives restart/replay.
83#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
84#[serde(rename_all = "snake_case")]
85pub enum Durability {
86    Durable,
87    Ephemeral,
88}
89
90/// Typed payload of a [`MuseRecord`], discriminated by `payload_type`.
91#[derive(Debug, Clone, PartialEq)]
92pub enum MusePayload {
93    /// `runtime.command.accepted`
94    CommandAccepted(CommandAccepted),
95    /// `session.run.linked`
96    SessionRunLinked(SessionRunLinked),
97    /// `turn.input.user`
98    TurnInputUser(TurnInputUser),
99    /// `run.lifecycle.started`
100    RunStarted(RunStarted),
101    /// `run.output.delta`
102    RunOutputDelta(RunOutputDelta),
103    /// `run.terminal.completed` (and any future `run.terminal.*`)
104    RunTerminal(RunTerminal),
105    /// `task.stream.linked`
106    TaskStreamLinked(TaskStreamLinked),
107    /// `task.lifecycle.*`
108    TaskLifecycle(TaskLifecycle),
109    /// A payload type not yet known to this crate — preserved verbatim.
110    Unknown {
111        payload_type: String,
112        payload: Value,
113    },
114}
115
116impl MusePayload {
117    pub fn from_parts(payload_type: &str, payload: Value) -> serde_json::Result<Self> {
118        Ok(match payload_type {
119            "runtime.command.accepted" => {
120                MusePayload::CommandAccepted(serde_json::from_value(payload)?)
121            }
122            "session.run.linked" => MusePayload::SessionRunLinked(serde_json::from_value(payload)?),
123            "turn.input.user" => MusePayload::TurnInputUser(serde_json::from_value(payload)?),
124            "run.lifecycle.started" => MusePayload::RunStarted(serde_json::from_value(payload)?),
125            "run.output.delta" => MusePayload::RunOutputDelta(serde_json::from_value(payload)?),
126            t if t.starts_with("run.terminal.") => {
127                MusePayload::RunTerminal(serde_json::from_value(payload)?)
128            }
129            "task.stream.linked" => MusePayload::TaskStreamLinked(serde_json::from_value(payload)?),
130            t if t.starts_with("task.lifecycle.") => {
131                MusePayload::TaskLifecycle(serde_json::from_value(payload)?)
132            }
133            other => MusePayload::Unknown {
134                payload_type: other.to_string(),
135                payload,
136            },
137        })
138    }
139}
140
141/// `runtime.command.accepted` — the runtime took ownership of a submitted
142/// command (`command_kind`, e.g. `turn.submit`).
143#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
144pub struct CommandAccepted {
145    pub kind: String,
146    pub command_id: String,
147    pub command_kind: String,
148    pub client_id: Option<String>,
149    #[serde(flatten, default, skip_serializing_if = "serde_json::Map::is_empty")]
150    pub extra: serde_json::Map<String, Value>,
151}
152
153/// `session.run.linked` — a run stream was attached to the session.
154#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
155pub struct SessionRunLinked {
156    pub kind: String,
157    pub command_id: String,
158    pub run_stream: StreamRef,
159    #[serde(flatten, default, skip_serializing_if = "serde_json::Map::is_empty")]
160    pub extra: serde_json::Map<String, Value>,
161}
162
163/// `turn.input.user` — the user prompt as the runtime recorded it.
164#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
165pub struct TurnInputUser {
166    pub kind: String,
167    pub command_id: String,
168    pub prompt: String,
169    pub run_stream: StreamRef,
170    #[serde(flatten, default, skip_serializing_if = "serde_json::Map::is_empty")]
171    pub extra: serde_json::Map<String, Value>,
172}
173
174/// `run.lifecycle.started`
175#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
176pub struct RunStarted {
177    pub kind: String,
178    pub command_id: String,
179    pub prompt: String,
180    pub run_stream: StreamRef,
181    #[serde(flatten, default, skip_serializing_if = "serde_json::Map::is_empty")]
182    pub extra: serde_json::Map<String, Value>,
183}
184
185/// `run.output.delta` — streamed model/agent output text.
186#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
187pub struct RunOutputDelta {
188    pub kind: String,
189    pub command_id: String,
190    pub run_stream: StreamRef,
191    pub text: String,
192    #[serde(flatten, default, skip_serializing_if = "serde_json::Map::is_empty")]
193    pub extra: serde_json::Map<String, Value>,
194}
195
196/// `run.terminal.*` — the run reached a terminal state. `terminal` carries
197/// the state (`completed` observed); `text` the final output; `reason` is
198/// populated on abnormal endings.
199#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
200pub struct RunTerminal {
201    pub kind: String,
202    pub command_id: String,
203    pub run_stream: StreamRef,
204    pub terminal: String,
205    pub reason: Option<String>,
206    #[serde(default, skip_serializing_if = "Option::is_none")]
207    pub text: Option<String>,
208    #[serde(flatten, default, skip_serializing_if = "serde_json::Map::is_empty")]
209    pub extra: serde_json::Map<String, Value>,
210}
211
212/// `task.stream.linked` — a task stream was attached to a run.
213#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
214pub struct TaskStreamLinked {
215    pub kind: String,
216    pub command_id: String,
217    pub run_stream: StreamRef,
218    pub task_id: String,
219    pub task_stream: StreamRef,
220    #[serde(flatten, default, skip_serializing_if = "serde_json::Map::is_empty")]
221    pub extra: serde_json::Map<String, Value>,
222}
223
224/// `task.lifecycle.*` — one step in a task's lifecycle state machine.
225#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
226pub struct TaskLifecycle {
227    pub kind: String,
228    pub command_id: String,
229    pub run_stream: StreamRef,
230    pub task_id: String,
231    pub task_stream: StreamRef,
232    pub event: TaskLifecycleEvent,
233    #[serde(flatten, default, skip_serializing_if = "serde_json::Map::is_empty")]
234    pub extra: serde_json::Map<String, Value>,
235}
236
237/// The `event` member of [`TaskLifecycle`], tagged by `kind`.
238///
239/// Observed lifecycle: `proposed → accepted → started → (scheduled →
240/// side_effect_intent →) completed | failed`.
241#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
242#[serde(tag = "kind", rename_all = "snake_case")]
243pub enum TaskLifecycleEvent {
244    Proposed {
245        task_id: String,
246        /// Dotted task class, e.g. `model.unknown.response` or
247        /// `reminder.agent.plugin:<plugin>:<name>`.
248        task_kind: String,
249    },
250    Accepted {
251        task_id: String,
252    },
253    Started {
254        task_id: String,
255    },
256    Scheduled {
257        task_id: String,
258        idempotency_key: String,
259    },
260    SideEffectIntent {
261        task_id: String,
262        idempotency_key: String,
263        operation: String,
264        policy_decision: String,
265        parent_task_id: Option<String>,
266        cancellation_handle: Option<Value>,
267    },
268    Completed {
269        task_id: String,
270    },
271    Failed {
272        task_id: String,
273        reason: String,
274    },
275    /// A lifecycle kind not yet known to this crate.
276    #[serde(untagged)]
277    Unknown(Value),
278}
279
280#[cfg(test)]
281mod tests {
282    use super::*;
283    use serde_json::json;
284
285    #[test]
286    fn unknown_payload_type_is_preserved_not_error() {
287        let p = MusePayload::from_parts("subagent.lifecycle.spawned", json!({"x": 1})).unwrap();
288        match p {
289            MusePayload::Unknown {
290                payload_type,
291                payload,
292            } => {
293                assert_eq!(payload_type, "subagent.lifecycle.spawned");
294                assert_eq!(payload, json!({"x": 1}));
295            }
296            other => panic!("expected Unknown, got {other:?}"),
297        }
298    }
299
300    #[test]
301    fn task_lifecycle_failed_carries_reason() {
302        let e: TaskLifecycleEvent = serde_json::from_value(json!({
303            "kind": "failed",
304            "task_id": "t1",
305            "reason": "provider does not support base instructions"
306        }))
307        .unwrap();
308        assert!(matches!(e, TaskLifecycleEvent::Failed { ref reason, .. }
309            if reason.contains("base instructions")));
310    }
311}
muse_codes/io/mod.rs

muse_codes/io/
mod.rs
