proef_core/event.rs
1//! The serde event spine (ADR-0008).
2//!
3//! One versioned, serde-able [`Event`] enum is the single source of truth for live
4//! progress *and* persistence: the JSONL run record **is** the appended event stream —
5//! there is no second record format. Changes are **additive-only**; the stream head
6//! ([`Event::RunStarted`]) declares [`EVENT_SCHEMA_VERSION`].
7//!
8//! Secret values never enter events — captures are reported by *name* only
9//! (redaction invariant, ADR-0005).
10
11use std::fmt;
12use std::sync::Arc;
13
14use serde::{Deserialize, Serialize};
15
16use crate::step::{Status, StepRef};
17
18/// Version of the event schema, declared once per stream in [`Event::RunStarted`].
19pub const EVENT_SCHEMA_VERSION: u32 = 1;
20
21/// One event in a run's stream. Serialized as JSONL, tagged by `event`.
22#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
23#[serde(tag = "event", rename_all = "snake_case")]
24pub enum Event {
25 /// The run began. Head of every stream; declares the schema version.
26 RunStarted {
27 /// Event schema version ([`EVENT_SCHEMA_VERSION`]).
28 schema: u32,
29 /// Injected run identifier (uuid-v7-derived; core never generates it).
30 run_id: Arc<str>,
31 },
32 /// A scenario began executing.
33 ScenarioStarted {
34 /// Scenario name as authored.
35 scenario: Arc<str>,
36 /// Feature file the scenario comes from.
37 file: Arc<str>,
38 /// Milliseconds since the run began — injected at the CLI sink (the
39 /// sans-IO core leaves it `None`, like `run_id`). Absent on records
40 /// without timing; additive (ADR-0008, ADR-0015).
41 #[serde(default, skip_serializing_if = "Option::is_none")]
42 timestamp_ms: Option<u64>,
43 /// 0-based worker index this scenario ran on — injected at the sink.
44 #[serde(default, skip_serializing_if = "Option::is_none")]
45 worker: Option<u64>,
46 /// The `[run]` lifecycle phase this scenario belongs to (`"setup"` /
47 /// `"teardown"`), absent for an ordinary suite scenario.
48 ///
49 /// Without it a phase scenario is indistinguishable from a suite one
50 /// except by feature path, so every consumer had to re-derive phase
51 /// membership from `proef.toml` — and `explain`, `--rerun` and `diff`
52 /// each got it wrong in a different way. The record says so itself now.
53 /// Additive and optional (ADR-0008): records that predate the field
54 /// read as "no phase", which is what they were.
55 #[serde(default, skip_serializing_if = "Option::is_none")]
56 phase: Option<Arc<str>>,
57 },
58 /// A batch of contiguous same-engine steps was dispatched.
59 BatchStarted {
60 /// Scenario name as authored.
61 scenario: Arc<str>,
62 /// Engine executing the batch.
63 engine: Arc<str>,
64 /// Number of steps in the batch.
65 steps: usize,
66 },
67 /// An artifact entry began an execution attempt — the engine's live
68 /// progress signal (ADR-0001's `EventListener`, surfaced on the spine).
69 /// Additive schema variant: absent from pre-existing streams.
70 EntryRunning {
71 /// Scenario name as authored.
72 scenario: Arc<str>,
73 /// Engine executing the entry.
74 engine: Arc<str>,
75 /// 0-based entry ordinal within the scenario's artifact.
76 entry: usize,
77 /// Retry number of this attempt (`0` = first attempt).
78 retry: u32,
79 },
80 /// A step finished (in success or failure).
81 StepFinished {
82 /// Scenario name as authored.
83 scenario: Arc<str>,
84 /// Engine that executed the step.
85 engine: Arc<str>,
86 /// Anchor to the authored feature line.
87 step: StepRef,
88 /// Outcome status.
89 status: Status,
90 /// Number of attempts made.
91 attempts: u32,
92 /// Wall-clock duration in milliseconds.
93 duration_ms: u64,
94 /// Names (never values) of captures produced by this step.
95 captures: Vec<String>,
96 /// The fragment that supplied this step's request, as `file.hurl#name`
97 /// (ADR-0018). Additive schema field: absent for an inline `hurl:`
98 /// block, which is every step in every stream written before
99 /// fragments existed.
100 #[serde(default, skip_serializing_if = "Option::is_none")]
101 fragment: Option<String>,
102 /// Failure detail, when the step failed (additive schema field —
103 /// absent on passing steps, so pre-existing streams are unchanged).
104 #[serde(default, skip_serializing_if = "Option::is_none")]
105 detail: Option<String>,
106 /// Messages from earlier, failed attempts of a step that ultimately
107 /// passed — the flaky-failure detail (`JUnit` `<flakyFailure>`).
108 /// Additive schema field: empty (and unserialized) for the common
109 /// single-attempt step, so pre-existing streams are unchanged.
110 #[serde(default, skip_serializing_if = "Vec::is_empty")]
111 attempt_details: Vec<String>,
112 },
113 /// A scenario finished.
114 ScenarioFinished {
115 /// Scenario name as authored.
116 scenario: Arc<str>,
117 /// Feature file path — together with `scenario`, the run-wide
118 /// identity (scenario names are unique only within one file).
119 /// Defaults empty when replaying records that predate the field
120 /// (schema 1 is additive-only — ADR-0008).
121 #[serde(default = "unknown_file")]
122 file: Arc<str>,
123 /// Aggregate scenario status.
124 status: Status,
125 /// Milliseconds since the run began — injected at the CLI sink (the
126 /// sans-IO core leaves it `None`, like `run_id`). Absent on records
127 /// without timing; additive (ADR-0008, ADR-0015).
128 #[serde(default, skip_serializing_if = "Option::is_none")]
129 timestamp_ms: Option<u64>,
130 /// 0-based worker index this scenario ran on — injected at the sink.
131 #[serde(default, skip_serializing_if = "Option::is_none")]
132 worker: Option<u64>,
133 /// The `[run]` lifecycle phase this scenario belongs to (`"setup"` /
134 /// `"teardown"`), absent for an ordinary suite scenario.
135 ///
136 /// Without it a phase scenario is indistinguishable from a suite one
137 /// except by feature path, so every consumer had to re-derive phase
138 /// membership from `proef.toml` — and `explain`, `--rerun` and `diff`
139 /// each got it wrong in a different way. The record says so itself now.
140 /// Additive and optional (ADR-0008): records that predate the field
141 /// read as "no phase", which is what they were.
142 #[serde(default, skip_serializing_if = "Option::is_none")]
143 phase: Option<Arc<str>>,
144 },
145 /// The run finished. Tail of every stream.
146 RunFinished {
147 /// Scenarios that passed.
148 passed: usize,
149 /// Scenarios that failed.
150 failed: usize,
151 /// Scenarios that were skipped.
152 skipped: usize,
153 /// The run was cancelled before completing (additive schema field —
154 /// absent means `false`, and `false` is not serialized, so pre-existing
155 /// streams and snapshots are unchanged).
156 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
157 cancelled: bool,
158 },
159}
160
161/// Serde default for records that predate the `file` field on
162/// [`Event::ScenarioFinished`].
163fn unknown_file() -> Arc<str> {
164 Arc::from("")
165}
166
167/// Fan-out point for [`Event`]s — reporters subscribe here (borrowed
168/// events). Cheap to clone; threads share the same sink.
169#[derive(Clone)]
170pub struct EventSink(Arc<dyn Fn(&Event) + Send + Sync>);
171
172impl EventSink {
173 /// A sink invoking `f` for every emitted event.
174 pub fn new(f: impl Fn(&Event) + Send + Sync + 'static) -> Self {
175 Self(Arc::new(f))
176 }
177
178 /// A sink that discards every event (tests, `--dry-run`).
179 pub fn null() -> Self {
180 Self(Arc::new(|_| {}))
181 }
182
183 /// Emit one event to all consumers.
184 pub fn emit(&self, event: &Event) {
185 (self.0)(event);
186 }
187}
188
189impl fmt::Debug for EventSink {
190 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
191 f.write_str("EventSink(..)")
192 }
193}
194
195#[cfg(test)]
196mod tests {
197 use super::*;
198
199 /// The wire shape is a compatibility surface (ADR-0008): pin the JSON of the
200 /// stream head exactly. Additive changes only.
201 #[test]
202 fn run_started_wire_shape_is_stable() {
203 let event = Event::RunStarted {
204 schema: EVENT_SCHEMA_VERSION,
205 run_id: Arc::from("run-0001"),
206 };
207 let json = serde_json::to_string(&event).unwrap_or_default();
208 assert_eq!(
209 json,
210 r#"{"event":"run_started","schema":1,"run_id":"run-0001"}"#
211 );
212 }
213
214 #[test]
215 fn events_round_trip_through_jsonl() {
216 let event = Event::StepFinished {
217 scenario: Arc::from("Search finds a record"),
218 engine: Arc::from("http"),
219 step: StepRef {
220 file: Arc::from("tests/features/501_search.feature"),
221 line: 12,
222 text: Arc::from("the admin searches for \"Jansen\""),
223 },
224 status: Status::Passed,
225 attempts: 2,
226 duration_ms: 42,
227 captures: vec!["recordId".to_owned()],
228 fragment: Some("tests/hurl/admin.hurl#admin.search".to_owned()),
229 detail: None,
230 attempt_details: vec!["attempt 1: HTTP 404 (retried)".to_owned()],
231 };
232 let json = serde_json::to_string(&event).unwrap_or_default();
233 let back: Event = serde_json::from_str(&json).unwrap_or(Event::RunFinished {
234 passed: 0,
235 failed: 0,
236 skipped: 0,
237 cancelled: false,
238 });
239 assert_eq!(back, event);
240 }
241
242 #[test]
243 fn sink_fans_out_borrowed_events() {
244 use std::sync::atomic::{AtomicUsize, Ordering};
245 let seen = Arc::new(AtomicUsize::new(0));
246 let counter = Arc::clone(&seen);
247 let sink = EventSink::new(move |_| {
248 counter.fetch_add(1, Ordering::SeqCst);
249 });
250 let clone = sink.clone();
251 clone.emit(&Event::RunFinished {
252 passed: 1,
253 failed: 0,
254 skipped: 0,
255 cancelled: false,
256 });
257 sink.emit(&Event::RunFinished {
258 passed: 1,
259 failed: 0,
260 skipped: 0,
261 cancelled: false,
262 });
263 assert_eq!(seen.load(Ordering::SeqCst), 2);
264 }
265}