Skip to main content

proef_core/
runner.rs

1//! The run orchestrator (TECH-SPEC §1, §12; ADR-0007): scenario-per-OS-thread,
2//! `--jobs`-bounded, cooperative cancellation at batch boundaries, and a
3//! heartbeat watchdog that **abandons** over-budget scenario threads (recording
4//! a `System` failure and detaching — the process reaps them at exit).
5//!
6//! Scenarios *prepare* lazily at dispatch time: `${global:key}` reads happen at
7//! lower time of the scenario (ADR-0005), so lowering + emission run inside the
8//! scenario thread against a snapshot of the shared global store; `saveAs:
9//! global` promotions merge back through the store lock when the scenario ends.
10//!
11//! Clock note: the pipeline stays clock-free (core purity); the orchestrator's
12//! monotonic-clock use is confined to budget enforcement and never enters
13//! events (durations are engine-measured).
14
15use std::collections::{BTreeMap, HashSet};
16use std::sync::mpsc;
17use std::sync::{Arc, Mutex, PoisonError};
18use std::time::{Duration, Instant};
19
20use crate::cancel::CancellationToken;
21use crate::diag::Diag;
22use crate::engine::{ArtifactRef, EngineFactory, HttpDefaults, ScenarioCtx};
23use crate::error::ExitCode;
24use crate::event::{EVENT_SCHEMA_VERSION, Event, EventSink};
25use crate::step::{Status, StepBatch, StepOutcome};
26use crate::world::{GlobalStore, World};
27
28/// Everything a scenario needs at dispatch time. Built by the CLI edge from
29/// owned/`Arc`ed data so scenario threads are `'static` (abandonable).
30pub struct ScenarioSpec {
31    /// Feature file path.
32    pub file: Arc<str>,
33    /// Scenario name (post-expansion).
34    pub name: Arc<str>,
35    /// 1-based header line.
36    pub line: usize,
37    /// Root for file bodies referenced by the scenario's entries (the feature
38    /// file's directory — hurl `context_dir` confinement, TECH-SPEC §13).
39    pub file_root: Option<std::path::PathBuf>,
40    /// Lower + emit against the live World snapshot (pure; runs in-thread).
41    pub prepare: PrepareFn,
42}
43
44/// The dispatch-time preparation: World snapshot in, batches + artifact out.
45pub type PrepareFn = Box<dyn FnOnce(&World) -> Result<Prepared, Vec<Diag>> + Send>;
46
47/// A scenario ready to execute.
48pub struct Prepared {
49    /// Engine batches in authored order.
50    pub batches: Vec<StepBatch>,
51    /// The emitted artifact (the hurl engine's executed input, ADR-0010).
52    pub artifact: Option<ArtifactRef>,
53}
54
55/// Run-level configuration.
56#[derive(Clone)]
57pub struct RunConfig {
58    /// Injected run identifier.
59    pub run_id: Arc<str>,
60    /// Parallel scenario workers.
61    pub jobs: usize,
62    /// Watchdog budget for a batch whose engine cannot estimate one.
63    pub default_batch_budget: Duration,
64    /// Secret name → value (engines inject via their redacting mechanisms).
65    pub secrets: Arc<BTreeMap<String, String>>,
66    /// Batch-level HTTP defaults.
67    pub http: HttpDefaults,
68}
69
70/// Aggregate result of a run.
71#[derive(Debug)]
72pub struct RunSummary {
73    /// Per-scenario outcomes, in completion order.
74    pub outcomes: Vec<ScenarioOutcome>,
75    /// Scenarios that passed (warnings allowed).
76    pub passed: usize,
77    /// Scenarios that failed.
78    pub failed: usize,
79    /// Scenarios skipped (cancellation).
80    pub skipped: usize,
81    /// The run was cancelled (Ctrl-C / token) — some scenarios did not run.
82    pub cancelled: bool,
83}
84
85impl RunSummary {
86    /// The run's exit code: system faults dominate, then user faults, then
87    /// test failures (ADR-0009). A cancelled run is never `Success` — the
88    /// suite did not pass; it was interrupted (folds in as a test failure).
89    pub fn exit_code(&self) -> ExitCode {
90        self.exit_code_excluding(&[])
91    }
92
93    /// [`Self::exit_code`], treating the given `(file, name)` scenarios as
94    /// non-gating (`@quarantine`): their *test-failures* no longer count toward
95    /// the exit code, but a `System`/`User` fault still does — quarantine is for
96    /// flaky tests, not broken input or infra.
97    pub fn exit_code_excluding(&self, non_gating: &[(String, String)]) -> ExitCode {
98        let mut worst = if self.cancelled {
99            ExitCode::TestFailure
100        } else {
101            ExitCode::Success
102        };
103        for outcome in &self.outcomes {
104            let quarantined = non_gating.iter().any(|(file, name)| {
105                file.as_str() == outcome.file.as_ref() && name.as_str() == outcome.name.as_ref()
106            });
107            let code = match (&outcome.fault, outcome.status) {
108                (Some(Fault::System(_)), _) => ExitCode::SystemError,
109                (Some(Fault::User(_)), _) => ExitCode::UserError,
110                (None, Status::Failed) if !quarantined => ExitCode::TestFailure,
111                _ => ExitCode::Success,
112            };
113            worst = pick_worse(worst, code);
114        }
115        worst
116    }
117}
118
119/// Prefer system errors over user errors over test failures.
120fn pick_worse(a: ExitCode, b: ExitCode) -> ExitCode {
121    let rank = |c: ExitCode| match c {
122        ExitCode::SystemError => 3,
123        ExitCode::UserError => 2,
124        ExitCode::TestFailure => 1,
125        ExitCode::Success => 0,
126    };
127    if rank(b) > rank(a) { b } else { a }
128}
129
130/// One scenario's outcome.
131#[derive(Debug)]
132pub struct ScenarioOutcome {
133    /// Feature file path.
134    pub file: Arc<str>,
135    /// Scenario name.
136    pub name: Arc<str>,
137    /// 1-based header line.
138    pub line: usize,
139    /// Aggregate status.
140    pub status: Status,
141    /// Step outcomes, in execution order.
142    pub steps: Vec<StepOutcome>,
143    /// Non-test fault, when one occurred.
144    pub fault: Option<Fault>,
145    /// Slug of the emitted artifact, when one exists (drives the CLI's
146    /// `reproduce:` line — the emitter's naming is never re-derived).
147    pub artifact_slug: Option<Arc<str>>,
148}
149
150/// A non-test fault attributed per ADR-0009.
151#[derive(Debug)]
152pub enum Fault {
153    /// The user's input is at fault (runtime resolution, missing secret, …).
154    User(String),
155    /// The environment or proef is at fault (engine infra, abandoned budget).
156    System(String),
157}
158
159enum Msg {
160    BatchBegin {
161        scenario: usize,
162        deadline: Instant,
163    },
164    Done {
165        scenario: usize,
166        outcome: ScenarioOutcome,
167    },
168}
169
170/// A scenario's run-wide identity: `(file, scenario)`, matching the pairing
171/// `identities`/`ScenarioOutcome` already use.
172type ScenarioId = (Arc<str>, Arc<str>);
173
174/// The gate's mutable state, behind one lock (see [`RecordGate`]).
175struct GateState {
176    /// Scenario identities that have been finalized — a worker's late event
177    /// for one of these is dropped rather than reaching `inner`.
178    closed: HashSet<ScenarioId>,
179    /// Set once, after `RunFinished` is written. Once `true`, nothing reaches
180    /// `inner` again, for any identity.
181    run_closed: bool,
182}
183
184/// Wraps the run's sink so a finalized scenario's late events never reach the
185/// record. An abandoned scenario's thread is detached and observes its
186/// cancellation token only at its next batch boundary (ADR-0007), so it can
187/// still try to emit after the sweep recorded its outcome — and after the run
188/// itself was finalized. The record's tail must be the tail.
189///
190/// One gate, two write paths, one lock: [`Self::scenario_sink`] hands each
191/// worker thread a sink pre-bound to its own scenario identity (so it filters
192/// without needing to inspect every event variant for `scenario`/`file`
193/// fields — not all of them carry both); [`Self::finish_scenario`] is called
194/// directly by the dispatcher thread, from both the normal completion path
195/// and `sweep_expired`, to emit a scenario's terminal event and mark it
196/// closed. Every read *and* write of [`GateState`] — including the emit
197/// itself — happens under the same `Mutex`, so "is this identity still open"
198/// and "write the event" are one atomic step rather than a check racing a
199/// concurrent close. [`Self::close_run`] is called only after `RunFinished`
200/// has already been written, per [`Self::emit_run_level`].
201#[derive(Clone)]
202struct RecordGate {
203    inner: EventSink,
204    state: Arc<Mutex<GateState>>,
205}
206
207impl RecordGate {
208    fn new(inner: EventSink) -> Self {
209        Self {
210            inner,
211            state: Arc::new(Mutex::new(GateState {
212                closed: HashSet::new(),
213                run_closed: false,
214            })),
215        }
216    }
217
218    /// A sink bound to one scenario's identity, handed to its worker thread.
219    /// Every event passed through it is dropped once the run is closed, or
220    /// once this scenario has been finalized via [`Self::finish_scenario`] —
221    /// including when that happened on the *dispatcher* thread (the watchdog
222    /// sweep), racing ahead of this scenario's own thread. The state lock is
223    /// held across the emit itself: a check-then-act (read the flag, drop
224    /// the lock, then emit) would leave a window for a worker to be
225    /// preempted right there and still write after the tail — the emit is
226    /// part of the same critical section as the check, not a step after it.
227    fn scenario_sink(&self, file: Arc<str>, scenario: Arc<str>) -> EventSink {
228        let inner = self.inner.clone();
229        let state = Arc::clone(&self.state);
230        let id: ScenarioId = (file, scenario);
231        EventSink::new(move |event| {
232            let guard = state.lock().unwrap_or_else(PoisonError::into_inner);
233            if guard.run_closed || guard.closed.contains(&id) {
234                return;
235            }
236            inner.emit(event);
237            // `guard` drops here, after the emit — not before it.
238        })
239    }
240
241    /// Emit a scenario's terminal `ScenarioFinished` and mark it closed under
242    /// one lock acquisition, so no [`Self::scenario_sink`] call for this
243    /// identity — nor a concurrent [`Self::close_run`] — can land between the
244    /// two: the terminal event and the closing of its identity are atomic.
245    /// Called from both the dispatcher's normal completion path and
246    /// `sweep_expired` — the two places a scenario is ever finalized.
247    fn finish_scenario(&self, event: &Event, file: &Arc<str>, scenario: &Arc<str>) {
248        let mut guard = self.state.lock().unwrap_or_else(PoisonError::into_inner);
249        if !guard.run_closed {
250            self.inner.emit(event);
251        }
252        guard
253            .closed
254            .insert((Arc::clone(file), Arc::clone(scenario)));
255    }
256
257    /// Emit an event that carries no single-scenario identity (`RunStarted`,
258    /// `RunFinished`) — gated only by the run-level close.
259    fn emit_run_level(&self, event: &Event) {
260        let guard = self.state.lock().unwrap_or_else(PoisonError::into_inner);
261        if !guard.run_closed {
262            self.inner.emit(event);
263        }
264    }
265
266    /// Shut the gate for good. Call only after `RunFinished` has already
267    /// been written via [`Self::emit_run_level`] — the tail must be written
268    /// before the gate closes, never the other way around.
269    fn close_run(&self) {
270        self.state
271            .lock()
272            .unwrap_or_else(PoisonError::into_inner)
273            .run_closed = true;
274    }
275}
276
277/// Execute `specs` and return the summary. Emits the full event stream on
278/// `events` (`RunStarted` … `RunFinished` — ADR-0008).
279// One cohesive listing of the dispatch/watchdog loop; splitting hides the order.
280#[allow(clippy::too_many_lines)]
281pub fn run(
282    specs: Vec<ScenarioSpec>,
283    engines: &Arc<Vec<Box<dyn EngineFactory>>>,
284    store: &Arc<Mutex<GlobalStore>>,
285    config: &RunConfig,
286    events: &EventSink,
287    cancel: &CancellationToken,
288) -> RunSummary {
289    // Every emission in this function goes through the gate, never `events`
290    // directly (one mechanism deciding what reaches the record — see
291    // `RecordGate`), so late writes from an abandoned scenario's detached
292    // thread can never slip past it.
293    let gate = RecordGate::new(events.clone());
294    gate.emit_run_level(&Event::RunStarted {
295        schema: EVENT_SCHEMA_VERSION,
296        run_id: Arc::clone(&config.run_id),
297    });
298
299    let (tx, rx) = mpsc::channel::<Msg>();
300    // Identities survive the specs' move into the queue, so an abandoned
301    // scenario is reported as itself, never as a synthetic placeholder.
302    let identities: Vec<(Arc<str>, Arc<str>, usize)> = specs
303        .iter()
304        .map(|spec| (Arc::clone(&spec.file), Arc::clone(&spec.name), spec.line))
305        .collect();
306    let mut queue: std::collections::VecDeque<(usize, ScenarioSpec)> =
307        specs.into_iter().enumerate().collect();
308    let total = queue.len();
309    // Deadline plus the scenario's child cancellation token — retained so
310    // abandonment can cancel the detached thread's work instead of leaving it
311    // appending events forever (record hygiene; engines gain real stop points
312    // as they support cancellation).
313    let mut active: BTreeMap<usize, (Instant, CancellationToken)> = BTreeMap::new();
314    let mut outcomes: Vec<ScenarioOutcome> = Vec::new();
315    let grace = Duration::from_secs(2);
316
317    while outcomes.len() < total {
318        // Fill free slots.
319        while active.len() < config.jobs.max(1) {
320            let Some((index, spec)) = queue.pop_front() else {
321                break;
322            };
323            if cancel.is_cancelled() {
324                gate.finish_scenario(
325                    &Event::ScenarioFinished {
326                        scenario: Arc::clone(&spec.name),
327                        file: Arc::clone(&spec.file),
328                        status: Status::Skipped,
329                        timestamp_ms: None,
330                        worker: None,
331                    },
332                    &spec.file,
333                    &spec.name,
334                );
335                outcomes.push(ScenarioOutcome {
336                    file: spec.file,
337                    name: spec.name,
338                    line: spec.line,
339                    status: Status::Skipped,
340                    steps: Vec::new(),
341                    fault: None,
342                    artifact_slug: None,
343                });
344                continue;
345            }
346            let initial_deadline = Instant::now() + config.default_batch_budget + grace;
347            let child = cancel.child_token();
348            active.insert(index, (initial_deadline, child.clone()));
349            let scenario_events =
350                gate.scenario_sink(Arc::clone(&spec.file), Arc::clone(&spec.name));
351            spawn_scenario(
352                index,
353                spec,
354                Arc::clone(engines),
355                Arc::clone(store),
356                config.clone(),
357                scenario_events,
358                child,
359                tx.clone(),
360            );
361        }
362        if active.is_empty() {
363            continue; // only skipped scenarios remained in the queue
364        }
365
366        // Wait for progress, bounded by the earliest active deadline.
367        let now = Instant::now();
368        let next_deadline = active
369            .values()
370            .map(|(deadline, _)| *deadline)
371            .min()
372            .unwrap_or(now + grace);
373        let wait = next_deadline
374            .saturating_duration_since(now)
375            .max(Duration::from_millis(20));
376        match rx.recv_timeout(wait) {
377            Ok(Msg::BatchBegin { scenario, deadline }) => {
378                if let Some(entry) = active.get_mut(&scenario) {
379                    entry.0 = deadline + grace;
380                }
381            }
382            Ok(Msg::Done { scenario, outcome }) => {
383                if active.remove(&scenario).is_some() {
384                    gate.finish_scenario(
385                        &Event::ScenarioFinished {
386                            scenario: Arc::clone(&outcome.name),
387                            file: Arc::clone(&outcome.file),
388                            status: outcome.status,
389                            timestamp_ms: None,
390                            worker: None,
391                        },
392                        &outcome.file,
393                        &outcome.name,
394                    );
395                    outcomes.push(outcome);
396                }
397                // else: a previously-abandoned thread finished late — ignored.
398            }
399            Err(mpsc::RecvTimeoutError::Timeout) => {}
400            Err(mpsc::RecvTimeoutError::Disconnected) => break,
401        }
402        // Sweep expired deadlines on *every* turn of the loop — a steady
403        // stream of messages from healthy scenarios must not keep a hung
404        // one alive past its budget.
405        sweep_expired(&mut active, &mut outcomes, &gate, &identities);
406    }
407
408    let passed = outcomes
409        .iter()
410        .filter(|o| matches!(o.status, Status::Passed | Status::Warned))
411        .count();
412    let failed = outcomes
413        .iter()
414        .filter(|o| o.status == Status::Failed)
415        .count();
416    let skipped = outcomes
417        .iter()
418        .filter(|o| o.status == Status::Skipped)
419        .count();
420    let cancelled = cancel.is_cancelled();
421    gate.emit_run_level(&Event::RunFinished {
422        passed,
423        failed,
424        skipped,
425        cancelled,
426    });
427    // The tail is written — only now does the gate shut. Any scenario thread
428    // still detached out there (an abandoned one, cooperatively cancelled but
429    // not yet reaped, ADR-0007) can no longer reach the sink at all.
430    gate.close_run();
431    RunSummary {
432        outcomes,
433        passed,
434        failed,
435        skipped,
436        cancelled,
437    }
438}
439
440/// Abandon every scenario whose deadline has passed: cancel its child token,
441/// record a `System` fault, and detach the thread (ADR-0007 — the process
442/// reaps it at exit; the cancelled token is the thread's cooperative signal
443/// to stop). Cooperation is not guaranteed by every engine, so `gate` is the
444/// actual backstop: `finish_scenario` closes this scenario's identity right
445/// here, before the detached thread can notice its token and try to keep
446/// appending to the record.
447fn sweep_expired(
448    active: &mut BTreeMap<usize, (Instant, CancellationToken)>,
449    outcomes: &mut Vec<ScenarioOutcome>,
450    gate: &RecordGate,
451    identities: &[(Arc<str>, Arc<str>, usize)],
452) {
453    let now = Instant::now();
454    let expired: Vec<usize> = active
455        .iter()
456        .filter(|(_, (deadline, _))| *deadline <= now)
457        .map(|(index, _)| *index)
458        .collect();
459    for index in expired {
460        if let Some((_, token)) = active.remove(&index) {
461            token.cancel();
462        }
463        // `active` keys are spec indices and `identities` is built 1:1 from
464        // the same specs — the lookup cannot miss.
465        let (file, name, line) = &identities[index];
466        let outcome = ScenarioOutcome {
467            file: Arc::clone(file),
468            name: Arc::clone(name),
469            line: *line,
470            status: Status::Failed,
471            steps: Vec::new(),
472            fault: Some(Fault::System(
473                "batch budget exceeded — scenario thread abandoned (ADR-0007)".to_owned(),
474            )),
475            artifact_slug: None,
476        };
477        gate.finish_scenario(
478            &Event::ScenarioFinished {
479                scenario: Arc::clone(&outcome.name),
480                file: Arc::clone(&outcome.file),
481                status: Status::Failed,
482                timestamp_ms: None,
483                worker: None,
484            },
485            &outcome.file,
486            &outcome.name,
487        );
488        outcomes.push(outcome);
489    }
490}
491
492#[allow(clippy::too_many_arguments)]
493fn spawn_scenario(
494    index: usize,
495    spec: ScenarioSpec,
496    engines: Arc<Vec<Box<dyn EngineFactory>>>,
497    store: Arc<Mutex<GlobalStore>>,
498    config: RunConfig,
499    events: EventSink,
500    cancel: CancellationToken,
501    tx: mpsc::Sender<Msg>,
502) {
503    std::thread::spawn(move || {
504        let identity = (Arc::clone(&spec.file), Arc::clone(&spec.name), spec.line);
505        let heartbeat_tx = tx.clone();
506        // A panicking engine or prepare closure must never look like a hang:
507        // contain it, report a System fault under the real identity, and let
508        // the dispatcher move on immediately instead of waiting out the budget.
509        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
510            run_scenario(
511                spec,
512                &engines,
513                &store,
514                &config,
515                &events,
516                &cancel,
517                |budget| {
518                    let _ = heartbeat_tx.send(Msg::BatchBegin {
519                        scenario: index,
520                        deadline: Instant::now() + budget,
521                    });
522                },
523            )
524        }));
525        let outcome = result.unwrap_or_else(|panic| {
526            let message = panic
527                .downcast_ref::<&str>()
528                .map(ToString::to_string)
529                .or_else(|| panic.downcast_ref::<String>().cloned())
530                .unwrap_or_else(|| "opaque panic payload".to_owned());
531            ScenarioOutcome {
532                file: identity.0,
533                name: identity.1,
534                line: identity.2,
535                status: Status::Failed,
536                steps: Vec::new(),
537                fault: Some(Fault::System(format!(
538                    "scenario thread panicked: {message}"
539                ))),
540                artifact_slug: None,
541            }
542        });
543        let _ = tx.send(Msg::Done {
544            scenario: index,
545            outcome,
546        });
547    });
548}
549
550// One cohesive listing of the scenario lifecycle; splitting hides the order.
551#[allow(clippy::too_many_lines)]
552fn run_scenario(
553    spec: ScenarioSpec,
554    engines: &[Box<dyn EngineFactory>],
555    store: &Mutex<GlobalStore>,
556    config: &RunConfig,
557    events: &EventSink,
558    cancel: &CancellationToken,
559    heartbeat: impl Fn(Duration),
560) -> ScenarioOutcome {
561    let (file, name, line) = (Arc::clone(&spec.file), Arc::clone(&spec.name), spec.line);
562    let outcome = move |status, steps, fault, artifact_slug| ScenarioOutcome {
563        file: Arc::clone(&file),
564        name: Arc::clone(&name),
565        line,
566        status,
567        steps,
568        fault,
569        artifact_slug,
570    };
571
572    events.emit(&Event::ScenarioStarted {
573        scenario: Arc::clone(&spec.name),
574        file: Arc::clone(&spec.file),
575        timestamp_ms: None,
576        worker: None,
577    });
578
579    // Prepare against a snapshot of the shared globals (lower-time reads).
580    let snapshot = match store.lock() {
581        Ok(guard) => guard.clone(),
582        Err(_) => {
583            return outcome(
584                Status::Failed,
585                Vec::new(),
586                Some(Fault::System("global store lock poisoned".to_owned())),
587                None,
588            );
589        }
590    };
591    let mut world = World::new(snapshot);
592    let prepared = match (spec.prepare)(&world) {
593        Ok(prepared) => prepared,
594        Err(diags) => {
595            let detail = diags
596                .iter()
597                .map(|d| d.message.clone())
598                .collect::<Vec<_>>()
599                .join("; ");
600            return outcome(Status::Failed, Vec::new(), Some(Fault::User(detail)), None);
601        }
602    };
603
604    let mut sessions: Vec<(String, Box<dyn crate::engine::EngineSession>)> = Vec::new();
605    let mut steps: Vec<StepOutcome> = Vec::new();
606    let mut fault: Option<Fault> = None;
607    let mut failed = false;
608
609    let mut interrupted = false;
610    let mut processed = 0usize;
611    for batch in &prepared.batches {
612        if failed {
613            break;
614        }
615        if cancel.is_cancelled() {
616            // Batches remain — the scenario did not run to completion and must
617            // not report `Passed` (a cancelled run would otherwise exit 0).
618            interrupted = true;
619            break;
620        }
621        let engine_id = batch.engine.as_str().to_owned();
622        if !sessions.iter().any(|(id, _)| *id == engine_id) {
623            let Some(factory) = engines.iter().find(|f| f.id() == engine_id) else {
624                fault = Some(Fault::System(format!(
625                    "no engine registered for `{engine_id}`"
626                )));
627                failed = true;
628                break;
629            };
630            let ctx = ScenarioCtx {
631                run_id: Arc::clone(&config.run_id),
632                scenario: Arc::clone(&spec.name),
633                artifact: prepared.artifact.clone(),
634                secrets: Arc::clone(&config.secrets),
635                http: config.http,
636                file_root: spec.file_root.clone(),
637            };
638            match factory.open(&ctx) {
639                Ok(session) => sessions.push((engine_id.clone(), session)),
640                Err(err) => {
641                    fault = Some(Fault::System(format!(
642                        "cannot open engine `{engine_id}`: {err}"
643                    )));
644                    failed = true;
645                    break;
646                }
647            }
648        }
649        let Some((_, session)) = sessions.iter_mut().find(|(id, _)| *id == engine_id) else {
650            break; // unreachable: just ensured above
651        };
652
653        let budget = session
654            .batch_budget(batch)
655            .unwrap_or(config.default_batch_budget);
656        heartbeat(budget);
657        events.emit(&Event::BatchStarted {
658            scenario: Arc::clone(&spec.name),
659            engine: Arc::from(engine_id.as_str()),
660            steps: batch.steps.len(),
661        });
662
663        let result = session.run_batch(batch, &mut world, events, cancel);
664        let all_optional = batch.steps.iter().all(|s| s.optional);
665        for mut step_outcome in result.steps {
666            if step_outcome.status == Status::Failed && all_optional {
667                step_outcome.status = Status::Warned;
668            }
669            steps.push(step_outcome);
670        }
671        if let Some(err) = result.error {
672            if all_optional {
673                // `optional:` — warn and continue (segmentation isolates it).
674                // The batch WAS dispatched and its steps already carry real
675                // outcomes: count it, or the unreached-steps loop below would
676                // re-report it as Skipped on top of them (ADR-0008).
677                processed += 1;
678                continue;
679            }
680            match err.class {
681                crate::error::EngineErrorClass::AssertFailed => {}
682                crate::error::EngineErrorClass::UserInput => {
683                    fault = Some(Fault::User(err.message.clone()));
684                }
685                crate::error::EngineErrorClass::Infra | crate::error::EngineErrorClass::Setup => {
686                    fault = Some(Fault::System(err.message.clone()));
687                }
688            }
689            failed = true;
690        }
691        processed += 1;
692    }
693
694    // Every authored step gets an outcome: batches never dispatched (earlier
695    // failure or cancellation) report their steps as Skipped instead of
696    // silently vanishing from console, record, and JUnit alike.
697    let unreached_reason = if interrupted {
698        "not run (run cancelled)"
699    } else {
700        "not run (an earlier step failed)"
701    };
702    for batch in prepared.batches.iter().skip(processed) {
703        for step in &batch.steps {
704            events.emit(&Event::StepFinished {
705                scenario: Arc::clone(&spec.name),
706                engine: Arc::from(batch.engine.as_str()),
707                step: step.step.clone(),
708                status: Status::Skipped,
709                attempts: 0,
710                duration_ms: 0,
711                captures: Vec::new(),
712                detail: Some(unreached_reason.to_owned()),
713                attempt_details: Vec::new(),
714            });
715            steps.push(StepOutcome {
716                step: step.step.clone(),
717                status: Status::Skipped,
718                attempts: 0,
719                duration: std::time::Duration::ZERO,
720                detail: Some(unreached_reason.to_owned()),
721                attempt_details: Vec::new(),
722                reproduce_hint: None,
723            });
724        }
725    }
726
727    for (engine_id, session) in sessions.iter_mut().rev() {
728        if let Err(err) = session.finish()
729            && fault.is_none()
730        {
731            // Teardown failure on an otherwise-clean scenario is a real infra
732            // signal (never silently swallowed); a scenario that already
733            // failed keeps its primary fault.
734            fault = Some(Fault::System(format!(
735                "engine `{engine_id}` teardown failed: {}",
736                err.message
737            )));
738        }
739    }
740
741    // Merge global promotions back through the store lock (§12) — the write
742    // set only. Writing the whole world back would clobber keys another
743    // scenario promoted after this one took its snapshot (lost update).
744    // Poison recovery is sound here: store inserts are plain map writes with
745    // no cross-key invariant a panicked holder could have torn — dropping the
746    // write set instead would silently lose `saveAs: global` promotions from
747    // a scenario that reports Passed.
748    {
749        let mut guard = store
750            .lock()
751            .unwrap_or_else(std::sync::PoisonError::into_inner);
752        for (key, value) in world.promotions() {
753            guard.insert(key, value.clone());
754        }
755    }
756
757    let status = if failed || steps.iter().any(|s| s.status == Status::Failed) {
758        Status::Failed
759    } else if interrupted {
760        Status::Skipped
761    } else {
762        Status::Passed
763    };
764    let artifact_slug = prepared
765        .artifact
766        .as_ref()
767        .map(|artifact| Arc::clone(&artifact.slug));
768    outcome(status, steps, fault, artifact_slug)
769}