1use 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
28pub struct ScenarioSpec {
31 pub file: Arc<str>,
33 pub name: Arc<str>,
35 pub line: usize,
37 pub file_root: Option<std::path::PathBuf>,
40 pub prepare: PrepareFn,
42}
43
44pub type PrepareFn = Box<dyn FnOnce(&World) -> Result<Prepared, Vec<Diag>> + Send>;
46
47pub struct Prepared {
49 pub batches: Vec<StepBatch>,
51 pub artifact: Option<ArtifactRef>,
53}
54
55#[derive(Clone)]
57pub struct RunConfig {
58 pub run_id: Arc<str>,
60 pub jobs: usize,
62 pub default_batch_budget: Duration,
64 pub secrets: Arc<BTreeMap<String, String>>,
66 pub http: HttpDefaults,
68}
69
70#[derive(Debug)]
72pub struct RunSummary {
73 pub outcomes: Vec<ScenarioOutcome>,
75 pub passed: usize,
77 pub failed: usize,
79 pub skipped: usize,
81 pub cancelled: bool,
83}
84
85impl RunSummary {
86 pub fn exit_code(&self) -> ExitCode {
90 self.exit_code_excluding(&[])
91 }
92
93 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
119fn 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#[derive(Debug)]
132pub struct ScenarioOutcome {
133 pub file: Arc<str>,
135 pub name: Arc<str>,
137 pub line: usize,
139 pub status: Status,
141 pub steps: Vec<StepOutcome>,
143 pub fault: Option<Fault>,
145 pub artifact_slug: Option<Arc<str>>,
148}
149
150#[derive(Debug)]
152pub enum Fault {
153 User(String),
155 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
170type ScenarioId = (Arc<str>, Arc<str>);
173
174struct GateState {
176 closed: HashSet<ScenarioId>,
179 run_closed: bool,
182}
183
184#[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 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 })
239 }
240
241 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 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 fn close_run(&self) {
270 self.state
271 .lock()
272 .unwrap_or_else(PoisonError::into_inner)
273 .run_closed = true;
274 }
275}
276
277#[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 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 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 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 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; }
365
366 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 }
399 Err(mpsc::RecvTimeoutError::Timeout) => {}
400 Err(mpsc::RecvTimeoutError::Disconnected) => break,
401 }
402 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 gate.close_run();
431 RunSummary {
432 outcomes,
433 passed,
434 failed,
435 skipped,
436 cancelled,
437 }
438}
439
440fn 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 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 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#[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 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 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; };
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 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 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 fault = Some(Fault::System(format!(
735 "engine `{engine_id}` teardown failed: {}",
736 err.message
737 )));
738 }
739 }
740
741 {
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}