release_kit/
events.rs

1//! The NDJSON event envelope for long-running commands.
2//!
3//! Bounded commands emit one JSON object; long-running ones emit one
4//! complete event per line, opening with a [`EventKind::Schema`] event
5//! that names the version, so a consumer knows what it is reading before
6//! the first step event arrives. The compatibility rules are stated once
7//! and held by test: a consumer ignores unknown fields and unknown event
8//! types, field names are never renamed, and new event types append.
9
10use serde::Serialize;
11
12use crate::diagnostic::Reason;
13
14/// The version of the event envelope's shape.
15pub const EVENTS_SCHEMA: &str = "rk.events/1";
16
17/// Every event type the stream can carry; additions append, and no
18/// variant is ever renamed.
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
20#[serde(rename_all = "snake_case")]
21pub enum EventKind {
22    /// The opening event naming the schema version.
23    Schema,
24    /// One step began.
25    StepStarted,
26    /// One step ended, with its status and duration.
27    StepFinished,
28    /// One chunk of a child process's output, tagged with its stream.
29    ChildOutput,
30    /// The run ended.
31    RunFinished,
32}
33
34/// Which of a child's streams a chunk came from.
35#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
36#[serde(rename_all = "snake_case")]
37pub enum ChildStream {
38    /// The child's stdout.
39    Stdout,
40    /// The child's stderr.
41    Stderr,
42}
43
44/// One event on the stream.
45///
46/// Every field is present on every event — absent values serialize as
47/// `null` rather than disappearing — so a consumer can parse one shape.
48#[derive(Debug, Serialize)]
49pub struct Event {
50    /// The envelope version, on every line.
51    pub schema: &'static str,
52    /// Monotonic sequence number within the run.
53    pub seq: u64,
54    /// Wall-clock UTC, RFC 3339.
55    pub time: String,
56    /// The run this event belongs to.
57    pub run_id: String,
58    /// The subcommand emitting the stream.
59    pub command: &'static str,
60    /// What happened.
61    #[serde(rename = "type")]
62    pub kind: EventKind,
63    /// The step concerned, where there is one.
64    pub step: Option<String>,
65    /// The step's result, on a finish event.
66    pub status: Option<String>,
67    /// The reason, on a failure.
68    pub reason: Option<Reason>,
69    /// The child's exit code, where one exists.
70    pub exit_code: Option<i32>,
71    /// How long the step took, on a finish event.
72    pub duration_ms: Option<u64>,
73    /// Which stream a `child_output` chunk came from.
74    pub stream: Option<ChildStream>,
75    /// The chunk's bytes, base64-encoded so invalid UTF-8 travels
76    /// losslessly; the journal transcript keeps the raw bytes in order.
77    pub data_b64: Option<String>,
78}
79
80impl Event {
81    /// The opening event of a stream: everything nullable is null.
82    #[must_use]
83    pub const fn opening(seq: u64, time: String, run_id: String, command: &'static str) -> Self {
84        Self {
85            schema: EVENTS_SCHEMA,
86            seq,
87            time,
88            run_id,
89            command,
90            kind: EventKind::Schema,
91            step: None,
92            status: None,
93            reason: None,
94            exit_code: None,
95            duration_ms: None,
96            stream: None,
97            data_b64: None,
98        }
99    }
100
101    /// A `child_output` event carrying one chunk of a child's output.
102    #[must_use]
103    pub fn child_output(mut self, stream: ChildStream, bytes: &[u8]) -> Self {
104        self.kind = EventKind::ChildOutput;
105        self.stream = Some(stream);
106        self.data_b64 = Some(base64(bytes));
107        self
108    }
109}
110
111/// Standard base64 with padding, encode only — the one direction this
112/// binary needs, so no dependency earns its place for it.
113fn base64(bytes: &[u8]) -> String {
114    const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
115    let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4);
116    for chunk in bytes.chunks(3) {
117        let b = [
118            chunk[0],
119            *chunk.get(1).unwrap_or(&0),
120            *chunk.get(2).unwrap_or(&0),
121        ];
122        let n = (u32::from(b[0]) << 16) | (u32::from(b[1]) << 8) | u32::from(b[2]);
123        for (idx, shift) in [18u32, 12, 6, 0].into_iter().enumerate() {
124            if idx <= chunk.len() {
125                out.push(char::from(ALPHABET[(n >> shift) as usize & 0x3f]));
126            } else {
127                out.push('=');
128            }
129        }
130    }
131    out
132}
133
134#[cfg(test)]
135mod tests {
136    #![allow(clippy::expect_used)]
137
138    use super::{ChildStream, Event, EventKind, base64};
139
140    /// The `rk.events/1` schema, held by snapshot: a field rename fails
141    /// here first and becomes a schema-version bump, not a silent parser
142    /// break at some agent.
143    #[test]
144    fn the_event_schema_snapshot_holds() {
145        let mut event =
146            Event::opening(0, "2026-08-29T14:10:31Z".into(), "01K5NQ7X".into(), "setup");
147        assert_eq!(
148            serde_json::to_string(&event).expect("an event serializes"),
149            r#"{"schema":"rk.events/1","seq":0,"time":"2026-08-29T14:10:31Z","run_id":"01K5NQ7X","command":"setup","type":"schema","step":null,"status":null,"reason":null,"exit_code":null,"duration_ms":null,"stream":null,"data_b64":null}"#
150        );
151        event.seq = 12;
152        event.kind = EventKind::StepFinished;
153        event.step = Some("protect-tags".into());
154        event.status = Some("satisfied".into());
155        event.exit_code = Some(0);
156        event.duration_ms = Some(418);
157        assert_eq!(
158            serde_json::to_string(&event).expect("an event serializes"),
159            r#"{"schema":"rk.events/1","seq":12,"time":"2026-08-29T14:10:31Z","run_id":"01K5NQ7X","command":"setup","type":"step_finished","step":"protect-tags","status":"satisfied","reason":null,"exit_code":0,"duration_ms":418,"stream":null,"data_b64":null}"#
160        );
161    }
162
163    /// A child's chunk travels with its stream tag and its bytes intact —
164    /// invalid UTF-8 included, which is why the payload is base64.
165    #[test]
166    fn a_child_output_event_carries_the_chunk_losslessly() {
167        let event = Event::opening(3, "2026-08-29T14:10:31Z".into(), "01K5NQ7X".into(), "setup")
168            .child_output(ChildStream::Stderr, &[0x66, 0x6f, 0x6f, 0xff, 0xfe]);
169        assert_eq!(
170            serde_json::to_string(&event).expect("an event serializes"),
171            r#"{"schema":"rk.events/1","seq":3,"time":"2026-08-29T14:10:31Z","run_id":"01K5NQ7X","command":"setup","type":"child_output","step":null,"status":null,"reason":null,"exit_code":null,"duration_ms":null,"stream":"stderr","data_b64":"Zm9v//4="}"#
172        );
173    }
174
175    /// The RFC 4648 vectors, so the hand-rolled encoder is checked against
176    /// values this crate did not compute.
177    #[test]
178    fn the_base64_encoder_matches_the_rfc_vectors() {
179        for (input, expected) in [
180            (&b""[..], ""),
181            (b"f", "Zg=="),
182            (b"fo", "Zm8="),
183            (b"foo", "Zm9v"),
184            (b"foob", "Zm9vYg=="),
185            (b"fooba", "Zm9vYmE="),
186            (b"foobar", "Zm9vYmFy"),
187        ] {
188            assert_eq!(base64(input), expected);
189        }
190    }
191}
release_kit/events.rs

release_kit/
events.rs
