proef-core 0.9.0

Engine-agnostic core of proef: parsing, binding, lowering, IR, emit, dispatch, World, events, errors
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
//! The run orchestrator (TECH-SPEC §1, §12; ADR-0007): scenario-per-OS-thread,
//! `--jobs`-bounded, cooperative cancellation at batch boundaries, and a
//! heartbeat watchdog that **abandons** over-budget scenario threads (recording
//! a `System` failure and detaching — the process reaps them at exit).
//!
//! Scenarios *prepare* lazily at dispatch time: `${global:key}` reads happen at
//! lower time of the scenario (ADR-0005), so lowering + emission run inside the
//! scenario thread against a snapshot of the shared global store; `saveAs:
//! global` promotions merge back through the store lock when the scenario ends.
//!
//! Clock note: the pipeline stays clock-free (core purity); the orchestrator's
//! monotonic-clock use is confined to budget enforcement and never enters
//! events (durations are engine-measured).

use std::collections::{BTreeMap, HashSet};
use std::sync::mpsc;
use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, Instant};

use crate::cancel::CancellationToken;
use crate::diag::Diag;
use crate::engine::{ArtifactRef, EngineFactory, HttpDefaults, ScenarioCtx};
use crate::error::ExitCode;
use crate::event::{EVENT_SCHEMA_VERSION, Event, EventSink};
use crate::step::{Status, StepBatch, StepOutcome};
use crate::world::{GlobalStore, World};

/// Everything a scenario needs at dispatch time. Built by the CLI edge from
/// owned/`Arc`ed data so scenario threads are `'static` (abandonable).
pub struct ScenarioSpec {
    /// Feature file path.
    pub file: Arc<str>,
    /// Scenario name (post-expansion).
    pub name: Arc<str>,
    /// 1-based header line.
    pub line: usize,
    /// Root for file bodies referenced by the scenario's entries (the feature
    /// file's directory — hurl `context_dir` confinement, TECH-SPEC §13).
    pub file_root: Option<std::path::PathBuf>,
    /// Lower + emit against the live World snapshot (pure; runs in-thread).
    pub prepare: PrepareFn,
}

/// The dispatch-time preparation: World snapshot in, batches + artifact out.
pub type PrepareFn = Box<dyn FnOnce(&World) -> Result<Prepared, Vec<Diag>> + Send>;

/// A scenario ready to execute.
pub struct Prepared {
    /// Engine batches in authored order.
    pub batches: Vec<StepBatch>,
    /// The emitted artifact (the hurl engine's executed input, ADR-0010).
    pub artifact: Option<ArtifactRef>,
}

/// Run-level configuration.
#[derive(Clone)]
pub struct RunConfig {
    /// Injected run identifier.
    pub run_id: Arc<str>,
    /// Parallel scenario workers.
    pub jobs: usize,
    /// Watchdog budget for a batch whose engine cannot estimate one.
    pub default_batch_budget: Duration,
    /// Secret name → value (engines inject via their redacting mechanisms).
    pub secrets: Arc<BTreeMap<String, String>>,
    /// Batch-level HTTP defaults.
    pub http: HttpDefaults,
}

/// Aggregate result of a run.
#[derive(Debug)]
pub struct RunSummary {
    /// Per-scenario outcomes, in completion order.
    pub outcomes: Vec<ScenarioOutcome>,
    /// Scenarios that passed (warnings allowed).
    pub passed: usize,
    /// Scenarios that failed.
    pub failed: usize,
    /// Scenarios skipped (cancellation).
    pub skipped: usize,
    /// The run was cancelled (Ctrl-C / token) — some scenarios did not run.
    pub cancelled: bool,
}

impl RunSummary {
    /// The run's exit code: system faults dominate, then user faults, then
    /// test failures (ADR-0009). A cancelled run is never `Success` — the
    /// suite did not pass; it was interrupted (folds in as a test failure).
    pub fn exit_code(&self) -> ExitCode {
        self.exit_code_excluding(&[])
    }

    /// [`Self::exit_code`], treating the given `(file, name)` scenarios as
    /// non-gating (`@quarantine`): their *test-failures* no longer count toward
    /// the exit code, but a `System`/`User` fault still does — quarantine is for
    /// flaky tests, not broken input or infra.
    pub fn exit_code_excluding(&self, non_gating: &[(String, String)]) -> ExitCode {
        let mut worst = if self.cancelled {
            ExitCode::TestFailure
        } else {
            ExitCode::Success
        };
        for outcome in &self.outcomes {
            let quarantined = non_gating.iter().any(|(file, name)| {
                file.as_str() == outcome.file.as_ref() && name.as_str() == outcome.name.as_ref()
            });
            let code = match (&outcome.fault, outcome.status) {
                (Some(Fault::System(_)), _) => ExitCode::SystemError,
                (Some(Fault::User(_)), _) => ExitCode::UserError,
                (None, Status::Failed) if !quarantined => ExitCode::TestFailure,
                _ => ExitCode::Success,
            };
            worst = pick_worse(worst, code);
        }
        worst
    }
}

/// Prefer system errors over user errors over test failures.
fn pick_worse(a: ExitCode, b: ExitCode) -> ExitCode {
    let rank = |c: ExitCode| match c {
        ExitCode::SystemError => 3,
        ExitCode::UserError => 2,
        ExitCode::TestFailure => 1,
        ExitCode::Success => 0,
    };
    if rank(b) > rank(a) { b } else { a }
}

/// One scenario's outcome.
#[derive(Debug)]
pub struct ScenarioOutcome {
    /// Feature file path.
    pub file: Arc<str>,
    /// Scenario name.
    pub name: Arc<str>,
    /// 1-based header line.
    pub line: usize,
    /// Aggregate status.
    pub status: Status,
    /// Step outcomes, in execution order.
    pub steps: Vec<StepOutcome>,
    /// Non-test fault, when one occurred.
    pub fault: Option<Fault>,
    /// Slug of the emitted artifact, when one exists (drives the CLI's
    /// `reproduce:` line — the emitter's naming is never re-derived).
    pub artifact_slug: Option<Arc<str>>,
}

/// A non-test fault attributed per ADR-0009.
#[derive(Debug)]
pub enum Fault {
    /// The user's input is at fault (runtime resolution, missing secret, …).
    User(String),
    /// The environment or proef is at fault (engine infra, abandoned budget).
    System(String),
}

enum Msg {
    BatchBegin {
        scenario: usize,
        deadline: Instant,
    },
    Done {
        scenario: usize,
        outcome: ScenarioOutcome,
    },
}

/// A scenario's run-wide identity: `(file, scenario)`, matching the pairing
/// `identities`/`ScenarioOutcome` already use.
type ScenarioId = (Arc<str>, Arc<str>);

/// The gate's mutable state, behind one lock (see [`RecordGate`]).
struct GateState {
    /// Scenario identities that have been finalized — a worker's late event
    /// for one of these is dropped rather than reaching `inner`.
    closed: HashSet<ScenarioId>,
    /// Set once, after `RunFinished` is written. Once `true`, nothing reaches
    /// `inner` again, for any identity.
    run_closed: bool,
}

/// Wraps the run's sink so a finalized scenario's late events never reach the
/// record. An abandoned scenario's thread is detached and observes its
/// cancellation token only at its next batch boundary (ADR-0007), so it can
/// still try to emit after the sweep recorded its outcome — and after the run
/// itself was finalized. The record's tail must be the tail.
///
/// One gate, two write paths, one lock: [`Self::scenario_sink`] hands each
/// worker thread a sink pre-bound to its own scenario identity (so it filters
/// without needing to inspect every event variant for `scenario`/`file`
/// fields — not all of them carry both); [`Self::finish_scenario`] is called
/// directly by the dispatcher thread, from both the normal completion path
/// and `sweep_expired`, to emit a scenario's terminal event and mark it
/// closed. Every read *and* write of [`GateState`] — including the emit
/// itself — happens under the same `Mutex`, so "is this identity still open"
/// and "write the event" are one atomic step rather than a check racing a
/// concurrent close. [`Self::close_run`] is called only after `RunFinished`
/// has already been written, per [`Self::emit_run_level`].
#[derive(Clone)]
struct RecordGate {
    inner: EventSink,
    state: Arc<Mutex<GateState>>,
}

impl RecordGate {
    fn new(inner: EventSink) -> Self {
        Self {
            inner,
            state: Arc::new(Mutex::new(GateState {
                closed: HashSet::new(),
                run_closed: false,
            })),
        }
    }

    /// A sink bound to one scenario's identity, handed to its worker thread.
    /// Every event passed through it is dropped once the run is closed, or
    /// once this scenario has been finalized via [`Self::finish_scenario`] —
    /// including when that happened on the *dispatcher* thread (the watchdog
    /// sweep), racing ahead of this scenario's own thread. The state lock is
    /// held across the emit itself: a check-then-act (read the flag, drop
    /// the lock, then emit) would leave a window for a worker to be
    /// preempted right there and still write after the tail — the emit is
    /// part of the same critical section as the check, not a step after it.
    fn scenario_sink(&self, file: Arc<str>, scenario: Arc<str>) -> EventSink {
        let inner = self.inner.clone();
        let state = Arc::clone(&self.state);
        let id: ScenarioId = (file, scenario);
        EventSink::new(move |event| {
            let guard = state.lock().unwrap_or_else(PoisonError::into_inner);
            if guard.run_closed || guard.closed.contains(&id) {
                return;
            }
            inner.emit(event);
            // `guard` drops here, after the emit — not before it.
        })
    }

    /// Emit a scenario's terminal `ScenarioFinished` and mark it closed under
    /// one lock acquisition, so no [`Self::scenario_sink`] call for this
    /// identity — nor a concurrent [`Self::close_run`] — can land between the
    /// two: the terminal event and the closing of its identity are atomic.
    /// Called from both the dispatcher's normal completion path and
    /// `sweep_expired` — the two places a scenario is ever finalized.
    fn finish_scenario(&self, event: &Event, file: &Arc<str>, scenario: &Arc<str>) {
        let mut guard = self.state.lock().unwrap_or_else(PoisonError::into_inner);
        if !guard.run_closed {
            self.inner.emit(event);
        }
        guard
            .closed
            .insert((Arc::clone(file), Arc::clone(scenario)));
    }

    /// Emit an event that carries no single-scenario identity (`RunStarted`,
    /// `RunFinished`) — gated only by the run-level close.
    fn emit_run_level(&self, event: &Event) {
        let guard = self.state.lock().unwrap_or_else(PoisonError::into_inner);
        if !guard.run_closed {
            self.inner.emit(event);
        }
    }

    /// Shut the gate for good. Call only after `RunFinished` has already
    /// been written via [`Self::emit_run_level`] — the tail must be written
    /// before the gate closes, never the other way around.
    fn close_run(&self) {
        self.state
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .run_closed = true;
    }
}

/// Execute `specs` and return the summary. Emits the full event stream on
/// `events` (`RunStarted` … `RunFinished` — ADR-0008).
// One cohesive listing of the dispatch/watchdog loop; splitting hides the order.
#[allow(clippy::too_many_lines)]
pub fn run(
    specs: Vec<ScenarioSpec>,
    engines: &Arc<Vec<Box<dyn EngineFactory>>>,
    store: &Arc<Mutex<GlobalStore>>,
    config: &RunConfig,
    events: &EventSink,
    cancel: &CancellationToken,
) -> RunSummary {
    // Every emission in this function goes through the gate, never `events`
    // directly (one mechanism deciding what reaches the record — see
    // `RecordGate`), so late writes from an abandoned scenario's detached
    // thread can never slip past it.
    let gate = RecordGate::new(events.clone());
    gate.emit_run_level(&Event::RunStarted {
        schema: EVENT_SCHEMA_VERSION,
        run_id: Arc::clone(&config.run_id),
    });

    let (tx, rx) = mpsc::channel::<Msg>();
    // Identities survive the specs' move into the queue, so an abandoned
    // scenario is reported as itself, never as a synthetic placeholder.
    let identities: Vec<(Arc<str>, Arc<str>, usize)> = specs
        .iter()
        .map(|spec| (Arc::clone(&spec.file), Arc::clone(&spec.name), spec.line))
        .collect();
    let mut queue: std::collections::VecDeque<(usize, ScenarioSpec)> =
        specs.into_iter().enumerate().collect();
    let total = queue.len();
    // Deadline plus the scenario's child cancellation token — retained so
    // abandonment can cancel the detached thread's work instead of leaving it
    // appending events forever (record hygiene; engines gain real stop points
    // as they support cancellation).
    let mut active: BTreeMap<usize, (Instant, CancellationToken)> = BTreeMap::new();
    let mut outcomes: Vec<ScenarioOutcome> = Vec::new();
    let grace = Duration::from_secs(2);

    while outcomes.len() < total {
        // Fill free slots.
        while active.len() < config.jobs.max(1) {
            let Some((index, spec)) = queue.pop_front() else {
                break;
            };
            if cancel.is_cancelled() {
                gate.finish_scenario(
                    &Event::ScenarioFinished {
                        scenario: Arc::clone(&spec.name),
                        file: Arc::clone(&spec.file),
                        status: Status::Skipped,
                        timestamp_ms: None,
                        worker: None,
                        phase: None,
                    },
                    &spec.file,
                    &spec.name,
                );
                outcomes.push(ScenarioOutcome {
                    file: spec.file,
                    name: spec.name,
                    line: spec.line,
                    status: Status::Skipped,
                    steps: Vec::new(),
                    fault: None,
                    artifact_slug: None,
                });
                continue;
            }
            let initial_deadline = Instant::now() + config.default_batch_budget + grace;
            let child = cancel.child_token();
            active.insert(index, (initial_deadline, child.clone()));
            let scenario_events =
                gate.scenario_sink(Arc::clone(&spec.file), Arc::clone(&spec.name));
            spawn_scenario(
                index,
                spec,
                Arc::clone(engines),
                Arc::clone(store),
                config.clone(),
                scenario_events,
                child,
                tx.clone(),
            );
        }
        if active.is_empty() {
            continue; // only skipped scenarios remained in the queue
        }

        // Wait for progress, bounded by the earliest active deadline.
        let now = Instant::now();
        let next_deadline = active
            .values()
            .map(|(deadline, _)| *deadline)
            .min()
            .unwrap_or(now + grace);
        let wait = next_deadline
            .saturating_duration_since(now)
            .max(Duration::from_millis(20));
        match rx.recv_timeout(wait) {
            Ok(Msg::BatchBegin { scenario, deadline }) => {
                if let Some(entry) = active.get_mut(&scenario) {
                    entry.0 = deadline + grace;
                }
            }
            Ok(Msg::Done { scenario, outcome }) => {
                if active.remove(&scenario).is_some() {
                    gate.finish_scenario(
                        &Event::ScenarioFinished {
                            scenario: Arc::clone(&outcome.name),
                            file: Arc::clone(&outcome.file),
                            status: outcome.status,
                            timestamp_ms: None,
                            worker: None,
                            phase: None,
                        },
                        &outcome.file,
                        &outcome.name,
                    );
                    outcomes.push(outcome);
                }
                // else: a previously-abandoned thread finished late — ignored.
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {}
            Err(mpsc::RecvTimeoutError::Disconnected) => break,
        }
        // Sweep expired deadlines on *every* turn of the loop — a steady
        // stream of messages from healthy scenarios must not keep a hung
        // one alive past its budget.
        sweep_expired(&mut active, &mut outcomes, &gate, &identities);
    }

    let passed = outcomes
        .iter()
        .filter(|o| matches!(o.status, Status::Passed | Status::Warned))
        .count();
    let failed = outcomes
        .iter()
        .filter(|o| o.status == Status::Failed)
        .count();
    let skipped = outcomes
        .iter()
        .filter(|o| o.status == Status::Skipped)
        .count();
    let cancelled = cancel.is_cancelled();
    gate.emit_run_level(&Event::RunFinished {
        passed,
        failed,
        skipped,
        cancelled,
    });
    // The tail is written — only now does the gate shut. Any scenario thread
    // still detached out there (an abandoned one, cooperatively cancelled but
    // not yet reaped, ADR-0007) can no longer reach the sink at all.
    gate.close_run();
    RunSummary {
        outcomes,
        passed,
        failed,
        skipped,
        cancelled,
    }
}

/// Abandon every scenario whose deadline has passed: cancel its child token,
/// record a `System` fault, and detach the thread (ADR-0007 — the process
/// reaps it at exit; the cancelled token is the thread's cooperative signal
/// to stop). Cooperation is not guaranteed by every engine, so `gate` is the
/// actual backstop: `finish_scenario` closes this scenario's identity right
/// here, before the detached thread can notice its token and try to keep
/// appending to the record.
fn sweep_expired(
    active: &mut BTreeMap<usize, (Instant, CancellationToken)>,
    outcomes: &mut Vec<ScenarioOutcome>,
    gate: &RecordGate,
    identities: &[(Arc<str>, Arc<str>, usize)],
) {
    let now = Instant::now();
    let expired: Vec<usize> = active
        .iter()
        .filter(|(_, (deadline, _))| *deadline <= now)
        .map(|(index, _)| *index)
        .collect();
    for index in expired {
        if let Some((_, token)) = active.remove(&index) {
            token.cancel();
        }
        // `active` keys are spec indices and `identities` is built 1:1 from
        // the same specs — the lookup cannot miss.
        let (file, name, line) = &identities[index];
        let outcome = ScenarioOutcome {
            file: Arc::clone(file),
            name: Arc::clone(name),
            line: *line,
            status: Status::Failed,
            steps: Vec::new(),
            fault: Some(Fault::System(
                "batch budget exceeded — scenario thread abandoned (ADR-0007)".to_owned(),
            )),
            artifact_slug: None,
        };
        gate.finish_scenario(
            &Event::ScenarioFinished {
                scenario: Arc::clone(&outcome.name),
                file: Arc::clone(&outcome.file),
                status: Status::Failed,
                timestamp_ms: None,
                worker: None,
                phase: None,
            },
            &outcome.file,
            &outcome.name,
        );
        outcomes.push(outcome);
    }
}

#[allow(clippy::too_many_arguments)]
fn spawn_scenario(
    index: usize,
    spec: ScenarioSpec,
    engines: Arc<Vec<Box<dyn EngineFactory>>>,
    store: Arc<Mutex<GlobalStore>>,
    config: RunConfig,
    events: EventSink,
    cancel: CancellationToken,
    tx: mpsc::Sender<Msg>,
) {
    std::thread::spawn(move || {
        let identity = (Arc::clone(&spec.file), Arc::clone(&spec.name), spec.line);
        let heartbeat_tx = tx.clone();
        // A panicking engine or prepare closure must never look like a hang:
        // contain it, report a System fault under the real identity, and let
        // the dispatcher move on immediately instead of waiting out the budget.
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            run_scenario(
                spec,
                &engines,
                &store,
                &config,
                &events,
                &cancel,
                |budget| {
                    let _ = heartbeat_tx.send(Msg::BatchBegin {
                        scenario: index,
                        deadline: Instant::now() + budget,
                    });
                },
            )
        }));
        let outcome = result.unwrap_or_else(|panic| {
            let message = panic
                .downcast_ref::<&str>()
                .map(ToString::to_string)
                .or_else(|| panic.downcast_ref::<String>().cloned())
                .unwrap_or_else(|| "opaque panic payload".to_owned());
            ScenarioOutcome {
                file: identity.0,
                name: identity.1,
                line: identity.2,
                status: Status::Failed,
                steps: Vec::new(),
                fault: Some(Fault::System(format!(
                    "scenario thread panicked: {message}"
                ))),
                artifact_slug: None,
            }
        });
        let _ = tx.send(Msg::Done {
            scenario: index,
            outcome,
        });
    });
}

// One cohesive listing of the scenario lifecycle; splitting hides the order.
#[allow(clippy::too_many_lines)]
fn run_scenario(
    spec: ScenarioSpec,
    engines: &[Box<dyn EngineFactory>],
    store: &Mutex<GlobalStore>,
    config: &RunConfig,
    events: &EventSink,
    cancel: &CancellationToken,
    heartbeat: impl Fn(Duration),
) -> ScenarioOutcome {
    let (file, name, line) = (Arc::clone(&spec.file), Arc::clone(&spec.name), spec.line);
    let outcome = move |status, steps, fault, artifact_slug| ScenarioOutcome {
        file: Arc::clone(&file),
        name: Arc::clone(&name),
        line,
        status,
        steps,
        fault,
        artifact_slug,
    };

    events.emit(&Event::ScenarioStarted {
        scenario: Arc::clone(&spec.name),
        file: Arc::clone(&spec.file),
        timestamp_ms: None,
        worker: None,
        phase: None,
    });

    // Prepare against a snapshot of the shared globals (lower-time reads).
    let snapshot = match store.lock() {
        Ok(guard) => guard.clone(),
        Err(_) => {
            return outcome(
                Status::Failed,
                Vec::new(),
                Some(Fault::System("global store lock poisoned".to_owned())),
                None,
            );
        }
    };
    let mut world = World::new(snapshot);
    let prepared = match (spec.prepare)(&world) {
        Ok(prepared) => prepared,
        Err(diags) => {
            let detail = diags
                .iter()
                .map(|d| d.message.clone())
                .collect::<Vec<_>>()
                .join("; ");
            return outcome(Status::Failed, Vec::new(), Some(Fault::User(detail)), None);
        }
    };

    let mut sessions: Vec<(String, Box<dyn crate::engine::EngineSession>)> = Vec::new();
    let mut steps: Vec<StepOutcome> = Vec::new();
    let mut fault: Option<Fault> = None;
    let mut failed = false;

    let mut interrupted = false;
    let mut processed = 0usize;
    for batch in &prepared.batches {
        if failed {
            break;
        }
        if cancel.is_cancelled() {
            // Batches remain — the scenario did not run to completion and must
            // not report `Passed` (a cancelled run would otherwise exit 0).
            interrupted = true;
            break;
        }
        let engine_id = batch.engine.as_str().to_owned();
        if !sessions.iter().any(|(id, _)| *id == engine_id) {
            let Some(factory) = engines.iter().find(|f| f.id() == engine_id) else {
                fault = Some(Fault::System(format!(
                    "no engine registered for `{engine_id}`"
                )));
                failed = true;
                break;
            };
            let ctx = ScenarioCtx {
                run_id: Arc::clone(&config.run_id),
                scenario: Arc::clone(&spec.name),
                artifact: prepared.artifact.clone(),
                secrets: Arc::clone(&config.secrets),
                http: config.http,
                file_root: spec.file_root.clone(),
            };
            match factory.open(&ctx) {
                Ok(session) => sessions.push((engine_id.clone(), session)),
                Err(err) => {
                    fault = Some(Fault::System(format!(
                        "cannot open engine `{engine_id}`: {err}"
                    )));
                    failed = true;
                    break;
                }
            }
        }
        let Some((_, session)) = sessions.iter_mut().find(|(id, _)| *id == engine_id) else {
            break; // unreachable: just ensured above
        };

        let budget = session
            .batch_budget(batch)
            .unwrap_or(config.default_batch_budget);
        heartbeat(budget);
        events.emit(&Event::BatchStarted {
            scenario: Arc::clone(&spec.name),
            engine: Arc::from(engine_id.as_str()),
            steps: batch.steps.len(),
        });

        let result = session.run_batch(batch, &mut world, events, cancel);
        let all_optional = batch.steps.iter().all(|s| s.optional);
        for mut step_outcome in result.steps {
            if step_outcome.status == Status::Failed && all_optional {
                step_outcome.status = Status::Warned;
            }
            steps.push(step_outcome);
        }
        if let Some(err) = result.error {
            if all_optional {
                // `optional:` — warn and continue (segmentation isolates it).
                // The batch WAS dispatched and its steps already carry real
                // outcomes: count it, or the unreached-steps loop below would
                // re-report it as Skipped on top of them (ADR-0008).
                processed += 1;
                continue;
            }
            match err.class {
                crate::error::EngineErrorClass::AssertFailed => {}
                crate::error::EngineErrorClass::UserInput => {
                    fault = Some(Fault::User(err.message.clone()));
                }
                crate::error::EngineErrorClass::Infra | crate::error::EngineErrorClass::Setup => {
                    fault = Some(Fault::System(err.message.clone()));
                }
            }
            failed = true;
        }
        processed += 1;
    }

    // Every authored step gets an outcome: batches never dispatched (earlier
    // failure or cancellation) report their steps as Skipped instead of
    // silently vanishing from console, record, and JUnit alike.
    let unreached_reason = if interrupted {
        "not run (run cancelled)"
    } else {
        "not run (an earlier step failed)"
    };
    for batch in prepared.batches.iter().skip(processed) {
        for step in &batch.steps {
            events.emit(&Event::StepFinished {
                scenario: Arc::clone(&spec.name),
                engine: Arc::from(batch.engine.as_str()),
                step: step.step.clone(),
                status: Status::Skipped,
                attempts: 0,
                duration_ms: 0,
                captures: Vec::new(),
                detail: Some(unreached_reason.to_owned()),
                attempt_details: Vec::new(),
            });
            steps.push(StepOutcome {
                step: step.step.clone(),
                status: Status::Skipped,
                attempts: 0,
                duration: std::time::Duration::ZERO,
                detail: Some(unreached_reason.to_owned()),
                attempt_details: Vec::new(),
                reproduce_hint: None,
            });
        }
    }

    for (engine_id, session) in sessions.iter_mut().rev() {
        if let Err(err) = session.finish()
            && fault.is_none()
        {
            // Teardown failure on an otherwise-clean scenario is a real infra
            // signal (never silently swallowed); a scenario that already
            // failed keeps its primary fault.
            fault = Some(Fault::System(format!(
                "engine `{engine_id}` teardown failed: {}",
                err.message
            )));
        }
    }

    // Merge global promotions back through the store lock (§12) — the write
    // set only. Writing the whole world back would clobber keys another
    // scenario promoted after this one took its snapshot (lost update).
    // Poison recovery is sound here: store inserts are plain map writes with
    // no cross-key invariant a panicked holder could have torn — dropping the
    // write set instead would silently lose `saveAs: global` promotions from
    // a scenario that reports Passed.
    {
        let mut guard = store
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        for (key, value) in world.promotions() {
            guard.insert(key, value.clone());
        }
    }

    let status = if failed || steps.iter().any(|s| s.status == Status::Failed) {
        Status::Failed
    } else if interrupted {
        Status::Skipped
    } else {
        Status::Passed
    };
    let artifact_slug = prepared
        .artifact
        .as_ref()
        .map(|artifact| Arc::clone(&artifact.slug));
    outcome(status, steps, fault, artifact_slug)
}