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 },
47 /// A batch of contiguous same-engine steps was dispatched.
48 BatchStarted {
49 /// Scenario name as authored.
50 scenario: Arc<str>,
51 /// Engine executing the batch.
52 engine: Arc<str>,
53 /// Number of steps in the batch.
54 steps: usize,
55 },
56 /// An artifact entry began an execution attempt — the engine's live
57 /// progress signal (ADR-0001's `EventListener`, surfaced on the spine).
58 /// Additive schema variant: absent from pre-existing streams.
59 EntryRunning {
60 /// Scenario name as authored.
61 scenario: Arc<str>,
62 /// Engine executing the entry.
63 engine: Arc<str>,
64 /// 0-based entry ordinal within the scenario's artifact.
65 entry: usize,
66 /// Retry number of this attempt (`0` = first attempt).
67 retry: u32,
68 },
69 /// A step finished (in success or failure).
70 StepFinished {
71 /// Scenario name as authored.
72 scenario: Arc<str>,
73 /// Engine that executed the step.
74 engine: Arc<str>,
75 /// Anchor to the authored feature line.
76 step: StepRef,
77 /// Outcome status.
78 status: Status,
79 /// Number of attempts made.
80 attempts: u32,
81 /// Wall-clock duration in milliseconds.
82 duration_ms: u64,
83 /// Names (never values) of captures produced by this step.
84 captures: Vec<String>,
85 /// Failure detail, when the step failed (additive schema field —
86 /// absent on passing steps, so pre-existing streams are unchanged).
87 #[serde(default, skip_serializing_if = "Option::is_none")]
88 detail: Option<String>,
89 /// Messages from earlier, failed attempts of a step that ultimately
90 /// passed — the flaky-failure detail (`JUnit` `<flakyFailure>`).
91 /// Additive schema field: empty (and unserialized) for the common
92 /// single-attempt step, so pre-existing streams are unchanged.
93 #[serde(default, skip_serializing_if = "Vec::is_empty")]
94 attempt_details: Vec<String>,
95 },
96 /// A scenario finished.
97 ScenarioFinished {
98 /// Scenario name as authored.
99 scenario: Arc<str>,
100 /// Feature file path — together with `scenario`, the run-wide
101 /// identity (scenario names are unique only within one file).
102 /// Defaults empty when replaying records that predate the field
103 /// (schema 1 is additive-only — ADR-0008).
104 #[serde(default = "unknown_file")]
105 file: Arc<str>,
106 /// Aggregate scenario status.
107 status: Status,
108 /// Milliseconds since the run began — injected at the CLI sink (the
109 /// sans-IO core leaves it `None`, like `run_id`). Absent on records
110 /// without timing; additive (ADR-0008, ADR-0015).
111 #[serde(default, skip_serializing_if = "Option::is_none")]
112 timestamp_ms: Option<u64>,
113 /// 0-based worker index this scenario ran on — injected at the sink.
114 #[serde(default, skip_serializing_if = "Option::is_none")]
115 worker: Option<u64>,
116 },
117 /// The run finished. Tail of every stream.
118 RunFinished {
119 /// Scenarios that passed.
120 passed: usize,
121 /// Scenarios that failed.
122 failed: usize,
123 /// Scenarios that were skipped.
124 skipped: usize,
125 /// The run was cancelled before completing (additive schema field —
126 /// absent means `false`, and `false` is not serialized, so pre-existing
127 /// streams and snapshots are unchanged).
128 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
129 cancelled: bool,
130 },
131}
132
133/// Serde default for records that predate the `file` field on
134/// [`Event::ScenarioFinished`].
135fn unknown_file() -> Arc<str> {
136 Arc::from("")
137}
138
139/// Fan-out point for [`Event`]s — reporters subscribe here (borrowed
140/// events). Cheap to clone; threads share the same sink.
141#[derive(Clone)]
142pub struct EventSink(Arc<dyn Fn(&Event) + Send + Sync>);
143
144impl EventSink {
145 /// A sink invoking `f` for every emitted event.
146 pub fn new(f: impl Fn(&Event) + Send + Sync + 'static) -> Self {
147 Self(Arc::new(f))
148 }
149
150 /// A sink that discards every event (tests, `--dry-run`).
151 pub fn null() -> Self {
152 Self(Arc::new(|_| {}))
153 }
154
155 /// Emit one event to all consumers.
156 pub fn emit(&self, event: &Event) {
157 (self.0)(event);
158 }
159}
160
161impl fmt::Debug for EventSink {
162 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
163 f.write_str("EventSink(..)")
164 }
165}
166
167#[cfg(test)]
168mod tests {
169 use super::*;
170
171 /// The wire shape is a compatibility surface (ADR-0008): pin the JSON of the
172 /// stream head exactly. Additive changes only.
173 #[test]
174 fn run_started_wire_shape_is_stable() {
175 let event = Event::RunStarted {
176 schema: EVENT_SCHEMA_VERSION,
177 run_id: Arc::from("run-0001"),
178 };
179 let json = serde_json::to_string(&event).unwrap_or_default();
180 assert_eq!(
181 json,
182 r#"{"event":"run_started","schema":1,"run_id":"run-0001"}"#
183 );
184 }
185
186 #[test]
187 fn events_round_trip_through_jsonl() {
188 let event = Event::StepFinished {
189 scenario: Arc::from("Search finds a record"),
190 engine: Arc::from("http"),
191 step: StepRef {
192 file: Arc::from("tests/features/501_search.feature"),
193 line: 12,
194 text: Arc::from("the admin searches for \"Jansen\""),
195 },
196 status: Status::Passed,
197 attempts: 2,
198 duration_ms: 42,
199 captures: vec!["recordId".to_owned()],
200 detail: None,
201 attempt_details: vec!["attempt 1: HTTP 404 (retried)".to_owned()],
202 };
203 let json = serde_json::to_string(&event).unwrap_or_default();
204 let back: Event = serde_json::from_str(&json).unwrap_or(Event::RunFinished {
205 passed: 0,
206 failed: 0,
207 skipped: 0,
208 cancelled: false,
209 });
210 assert_eq!(back, event);
211 }
212
213 #[test]
214 fn sink_fans_out_borrowed_events() {
215 use std::sync::atomic::{AtomicUsize, Ordering};
216 let seen = Arc::new(AtomicUsize::new(0));
217 let counter = Arc::clone(&seen);
218 let sink = EventSink::new(move |_| {
219 counter.fetch_add(1, Ordering::SeqCst);
220 });
221 let clone = sink.clone();
222 clone.emit(&Event::RunFinished {
223 passed: 1,
224 failed: 0,
225 skipped: 0,
226 cancelled: false,
227 });
228 sink.emit(&Event::RunFinished {
229 passed: 1,
230 failed: 0,
231 skipped: 0,
232 cancelled: false,
233 });
234 assert_eq!(seen.load(Ordering::SeqCst), 2);
235 }
236}