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 /// Failure detail, when the step failed (additive schema field —
97 /// absent on passing steps, so pre-existing streams are unchanged).
98 #[serde(default, skip_serializing_if = "Option::is_none")]
99 detail: Option<String>,
100 /// Messages from earlier, failed attempts of a step that ultimately
101 /// passed — the flaky-failure detail (`JUnit` `<flakyFailure>`).
102 /// Additive schema field: empty (and unserialized) for the common
103 /// single-attempt step, so pre-existing streams are unchanged.
104 #[serde(default, skip_serializing_if = "Vec::is_empty")]
105 attempt_details: Vec<String>,
106 },
107 /// A scenario finished.
108 ScenarioFinished {
109 /// Scenario name as authored.
110 scenario: Arc<str>,
111 /// Feature file path — together with `scenario`, the run-wide
112 /// identity (scenario names are unique only within one file).
113 /// Defaults empty when replaying records that predate the field
114 /// (schema 1 is additive-only — ADR-0008).
115 #[serde(default = "unknown_file")]
116 file: Arc<str>,
117 /// Aggregate scenario status.
118 status: Status,
119 /// Milliseconds since the run began — injected at the CLI sink (the
120 /// sans-IO core leaves it `None`, like `run_id`). Absent on records
121 /// without timing; additive (ADR-0008, ADR-0015).
122 #[serde(default, skip_serializing_if = "Option::is_none")]
123 timestamp_ms: Option<u64>,
124 /// 0-based worker index this scenario ran on — injected at the sink.
125 #[serde(default, skip_serializing_if = "Option::is_none")]
126 worker: Option<u64>,
127 /// The `[run]` lifecycle phase this scenario belongs to (`"setup"` /
128 /// `"teardown"`), absent for an ordinary suite scenario.
129 ///
130 /// Without it a phase scenario is indistinguishable from a suite one
131 /// except by feature path, so every consumer had to re-derive phase
132 /// membership from `proef.toml` — and `explain`, `--rerun` and `diff`
133 /// each got it wrong in a different way. The record says so itself now.
134 /// Additive and optional (ADR-0008): records that predate the field
135 /// read as "no phase", which is what they were.
136 #[serde(default, skip_serializing_if = "Option::is_none")]
137 phase: Option<Arc<str>>,
138 },
139 /// The run finished. Tail of every stream.
140 RunFinished {
141 /// Scenarios that passed.
142 passed: usize,
143 /// Scenarios that failed.
144 failed: usize,
145 /// Scenarios that were skipped.
146 skipped: usize,
147 /// The run was cancelled before completing (additive schema field —
148 /// absent means `false`, and `false` is not serialized, so pre-existing
149 /// streams and snapshots are unchanged).
150 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
151 cancelled: bool,
152 },
153}
154
155/// Serde default for records that predate the `file` field on
156/// [`Event::ScenarioFinished`].
157fn unknown_file() -> Arc<str> {
158 Arc::from("")
159}
160
161/// Fan-out point for [`Event`]s — reporters subscribe here (borrowed
162/// events). Cheap to clone; threads share the same sink.
163#[derive(Clone)]
164pub struct EventSink(Arc<dyn Fn(&Event) + Send + Sync>);
165
166impl EventSink {
167 /// A sink invoking `f` for every emitted event.
168 pub fn new(f: impl Fn(&Event) + Send + Sync + 'static) -> Self {
169 Self(Arc::new(f))
170 }
171
172 /// A sink that discards every event (tests, `--dry-run`).
173 pub fn null() -> Self {
174 Self(Arc::new(|_| {}))
175 }
176
177 /// Emit one event to all consumers.
178 pub fn emit(&self, event: &Event) {
179 (self.0)(event);
180 }
181}
182
183impl fmt::Debug for EventSink {
184 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
185 f.write_str("EventSink(..)")
186 }
187}
188
189#[cfg(test)]
190mod tests {
191 use super::*;
192
193 /// The wire shape is a compatibility surface (ADR-0008): pin the JSON of the
194 /// stream head exactly. Additive changes only.
195 #[test]
196 fn run_started_wire_shape_is_stable() {
197 let event = Event::RunStarted {
198 schema: EVENT_SCHEMA_VERSION,
199 run_id: Arc::from("run-0001"),
200 };
201 let json = serde_json::to_string(&event).unwrap_or_default();
202 assert_eq!(
203 json,
204 r#"{"event":"run_started","schema":1,"run_id":"run-0001"}"#
205 );
206 }
207
208 #[test]
209 fn events_round_trip_through_jsonl() {
210 let event = Event::StepFinished {
211 scenario: Arc::from("Search finds a record"),
212 engine: Arc::from("http"),
213 step: StepRef {
214 file: Arc::from("tests/features/501_search.feature"),
215 line: 12,
216 text: Arc::from("the admin searches for \"Jansen\""),
217 },
218 status: Status::Passed,
219 attempts: 2,
220 duration_ms: 42,
221 captures: vec!["recordId".to_owned()],
222 detail: None,
223 attempt_details: vec!["attempt 1: HTTP 404 (retried)".to_owned()],
224 };
225 let json = serde_json::to_string(&event).unwrap_or_default();
226 let back: Event = serde_json::from_str(&json).unwrap_or(Event::RunFinished {
227 passed: 0,
228 failed: 0,
229 skipped: 0,
230 cancelled: false,
231 });
232 assert_eq!(back, event);
233 }
234
235 #[test]
236 fn sink_fans_out_borrowed_events() {
237 use std::sync::atomic::{AtomicUsize, Ordering};
238 let seen = Arc::new(AtomicUsize::new(0));
239 let counter = Arc::clone(&seen);
240 let sink = EventSink::new(move |_| {
241 counter.fetch_add(1, Ordering::SeqCst);
242 });
243 let clone = sink.clone();
244 clone.emit(&Event::RunFinished {
245 passed: 1,
246 failed: 0,
247 skipped: 0,
248 cancelled: false,
249 });
250 sink.emit(&Event::RunFinished {
251 passed: 1,
252 failed: 0,
253 skipped: 0,
254 cancelled: false,
255 });
256 assert_eq!(seen.load(Ordering::SeqCst), 2);
257 }
258}