hegeltest 0.50.3

//! Shared per-test-case execution lifecycle.
//!
//! Provides the panic hook, the `run_test_case` wrapper that catches a single
//! test body's panic and converts it into a [`TestCaseResult`], and the
//! `drive` function that starts a libhegel run, pulls each test case the
//! engine schedules, and surfaces the run-level result.
//!
//! The engine lives behind libhegel's C ABI; `drive` owns everything around
//! it — installing the panic hook, wrapping each test body with
//! `catch_unwind` plus `mark_complete`, the antithesis integration, printing
//! each reported failure from its captured material, and the closing
//! re-raise of the failing test's own panic.

use std::backtrace::{Backtrace, BacktraceStatus};
use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::panic::{self, AssertUnwindSafe, catch_unwind};
use std::sync::{Arc, Once};

use crate::backend::{Failure, TestCaseResult};
use crate::control::{
    AssumeFailed, InternalError, InvalidArgument, LoopDone, StopTest, currently_in_test_context,
    hegel_internal_error, with_test_context,
};
use crate::ffi::{CTestCase, RunHandle, SettingsHandle};
use crate::runner::{Settings, TestLocation, Verbosity};
use crate::test_case::{RunOutput, TestCase};

static PANIC_HOOK_INIT: Once = Once::new();

thread_local! {
    /// `(thread_name, thread_id, location, backtrace)` captured by the panic
    /// hook for the most recent panic raised inside a test context. The
    /// hook overwrites this on every panic; callers consume it with
    /// [`take_panic_info`] right after `catch_unwind` returns.
    static LAST_PANIC_INFO: RefCell<Option<(String, String, String, Backtrace)>> =
        const { RefCell::new(None) };

    /// Whether the panic hook should pay to capture a backtrace for the next
    /// panic on this thread. Set by [`run_test_case`] to `should_emit` —
    /// i.e. only when the resulting diagnostic will actually be shown.
    /// Capturing (and, under
    /// `RUST_BACKTRACE`, symbolizing) a backtrace for every discarded shrink
    /// probe is the dominant cost of failing-heavy property runs, and is far
    /// worse on Windows.
    static CAPTURE_BACKTRACE: Cell<bool> = const { Cell::new(false) };
}

/// The `(thread_name, thread_id, location, backtrace)` tuple the panic hook
/// captures.
pub(crate) type PanicInfo = (String, String, String, Backtrace);

pub(crate) fn take_panic_info() -> Option<PanicInfo> {
    LAST_PANIC_INFO.with(|info| info.borrow_mut().take())
}

/// Install `info` into this thread's panic-info slot, as if the panic hook
/// had captured it here. Used by `stateful::Machine::run_concurrent` to re-install a
/// worker thread's capture on the main thread before `resume_unwind`ing the
/// ferried payload — the re-raise skips the panic hook, so without this the
/// lifecycle would fall back to [`unknown_panic_info`] and every concurrent
/// failure would share the origin `"Panic at <unknown>"`.
pub(crate) fn install_panic_info(info: PanicInfo) {
    LAST_PANIC_INFO.with(|slot| *slot.borrow_mut() = Some(info));
}

/// Whether the panic hook captures backtraces on this thread right now.
/// `run_test_case` decides this per case; `stateful::Machine::run_concurrent` reads
/// it on the main thread to mirror the setting onto its worker threads.
pub(crate) fn backtrace_capture_enabled() -> bool {
    CAPTURE_BACKTRACE.get()
}

/// Set whether the panic hook captures backtraces on this thread. Called by
/// `stateful::Machine::run_concurrent` on each worker thread (see
/// [`backtrace_capture_enabled`]).
pub(crate) fn set_backtrace_capture(enabled: bool) {
    CAPTURE_BACKTRACE.set(enabled);
}

/// Install the cross-backend panic hook on first call.
///
/// Idempotent across all backends: the hook captures the location for any
/// panic raised inside a test context (so [`run_test_case`] can read it after
/// `catch_unwind`), and forwards everything else to the previous hook
/// unchanged. A backtrace is captured only when [`CAPTURE_BACKTRACE`] is
/// set. Control-flow unwinds (a rejected assumption, out-of-data, ...)
/// never reach any hook at all — they are raised via
/// [`crate::control::raise_control`]'s `resume_unwind`, which skips hooks
/// by construction.
pub(crate) fn init_panic_hook() {
    PANIC_HOOK_INIT.call_once(|| {
        let prev_hook = panic::take_hook();
        panic::set_hook(Box::new(move |info| {
            if !currently_in_test_context() {
                prev_hook(info);
                return;
            }

            let thread = std::thread::current();
            let thread_name = thread.name().unwrap_or("<unnamed>").to_string();
            let thread_id = format!("{:?}", thread.id())
                .trim_start_matches("ThreadId(")
                .trim_end_matches(')')
                .to_string();
            let location = info
                .location()
                .map(|loc| format!("{}:{}:{}", loc.file(), loc.line(), loc.column()))
                .unwrap_or_else(|| "<unknown>".to_string());
            let backtrace = if CAPTURE_BACKTRACE.get() {
                Backtrace::capture()
            } else {
                Backtrace::disabled()
            };

            LAST_PANIC_INFO
                .with(|l| *l.borrow_mut() = Some((thread_name, thread_id, location, backtrace)));
        }));
    });
}

/// Format a backtrace, optionally filtering to the "short" format that the
/// default Rust panic handler uses.
fn format_backtrace(bt: &Backtrace, full: bool) -> String {
    let backtrace_str = format!("{}", bt);
    if full {
        return backtrace_str;
    }
    filter_short_backtrace(&backtrace_str)
}

/// Trim a `Backtrace`-as-string down to the "short" view (between
/// `__rust_end_short_backtrace` and `__rust_begin_short_backtrace`) and
/// renumber the surviving frames. Split out from [`format_backtrace`] so
/// the string-shape branches (frames with a `digit:` prefix, frames with
/// a digit but no colon, frames that don't start with a digit) can be
/// covered directly.
fn filter_short_backtrace(backtrace_str: &str) -> String {
    let lines: Vec<&str> = backtrace_str.lines().collect();
    let mut start_idx = 0;
    let mut end_idx = lines.len();

    for (i, line) in lines.iter().enumerate() {
        if line.contains("__rust_end_short_backtrace") {
            for (j, next_line) in lines.iter().enumerate().skip(i + 1) {
                if next_line
                    .trim_start()
                    .chars()
                    .next()
                    .is_some_and(|c| c.is_ascii_digit())
                {
                    start_idx = j;
                    break;
                }
            }
        }
        if line.contains("__rust_begin_short_backtrace") {
            for (j, prev_line) in lines
                .iter()
                .enumerate()
                .take(i + 1)
                .collect::<Vec<_>>()
                .into_iter()
                .rev()
            {
                if prev_line
                    .trim_start()
                    .chars()
                    .next()
                    .is_some_and(|c| c.is_ascii_digit())
                {
                    end_idx = j;
                    break;
                }
            }
            break;
        }
    }

    if start_idx > end_idx {
        start_idx = 0;
        end_idx = lines.len();
    }
    let filtered: Vec<&str> = lines[start_idx..end_idx].to_vec();
    let mut new_frame_num = 0usize;
    let mut result = Vec::new();
    for line in filtered {
        let trimmed = line.trim_start();
        if trimmed.chars().next().is_some_and(|c| c.is_ascii_digit()) {
            if let Some(colon_pos) = trimmed.find(':') {
                let rest = &trimmed[colon_pos..];
                result.push(format!("{:>4}{}", new_frame_num, rest));
                new_frame_num += 1;
            } else {
                result.push(line.to_string());
            }
        } else {
            result.push(line.to_string());
        }
    }
    result.join("\n")
}

/// Placeholder thread/location/backtrace tuple used when the panic hook
/// captured nothing for a caught panic. This *is* reached in production:
/// a panic payload ferried from another thread and re-raised on the test
/// thread with `resume_unwind` skips the hook, so the lifecycle finds
/// nothing here. (A plain `join().unwrap()` instead panics afresh
/// through the hook, pinning the origin to the join site and degrading
/// the message; a never-joined thread's panic produces no failure at
/// all.) One consequence is that every ferried failure shares the origin
/// `"Panic at <unknown>"`, merging distinct threaded bugs into one
/// counterexample; fixing that needs cross-thread capture, which is
/// deferred until there is structured concurrency support to hang it on.
pub(crate) fn unknown_panic_info() -> (String, String, String, Backtrace) {
    (
        "<unknown>".to_string(),
        "?".to_string(),
        "<unknown>".to_string(),
        Backtrace::disabled(),
    )
}

/// Extract a string message from a panic payload. Pre-N1 this was duplicated
/// in `src/native/runner.rs:11`; that copy now re-exports this one.
///
/// `pub` so that the libhegel C bindings (`libhegel/`) can use it from their
/// own `catch_unwind` wrapper around `run_native`. Not part of the
/// supported public surface — `#[doc(hidden)]`.
#[doc(hidden)]
pub fn panic_message(payload: &Box<dyn std::any::Any + Send>) -> String {
    if let Some(s) = payload.downcast_ref::<&str>() {
        s.to_string()
    } else if let Some(s) = payload.downcast_ref::<String>() {
        s.clone()
    } else {
        "Unknown panic".to_string()
    }
}

/// Run the user's test body once against the supplied libhegel test case,
/// catching any panic and translating it to a [`TestCaseResult`].
///
/// The body's `TestCase` shares `c_tc` through an `Arc`, and the outcome is
/// reported on the same shared handle after the body finishes — so no
/// per-case libhegel clone is needed (a real clone is created only when the
/// body itself calls `tc.clone()`), while a `TestCase` that escapes the body
/// (moved to a thread that is never joined) keeps the handle alive rather
/// than dangling; its later draws fail cleanly because the case has finished.
///
/// Reports the outcome back to the engine via [`report_outcome`] (which calls
/// `hegel_mark_complete`): that is the channel for per-test-case results, and
/// the engine reads it back over the C ABI.
/// On the `Interesting` path the panic site is captured as a
/// `file:line:col` string and stored on the [`Failure`] so per-origin
/// shrinking can key on it, and the rendered diagnostic block (panic
/// location, message, backtrace) is built here, at the moment the panic
/// is caught — on a non-quiet final replay it is returned to the caller to
/// print (right after the live draw/note lines, which is what keeps each
/// failure one block), for a non-final case in verbose mode it goes to
/// `output`, the run's resolved destination, and for a stamped non-final
/// case it is returned to the caller for the deferred failure report (and,
/// in verbose mode, printed to `output` as well, like any other non-final
/// case's).
///
/// The capture decision is the engine's, read once at case start
/// ([`CTestCase::should_capture`]): the engine stamps the executions a
/// failure report can be built from — confirmation batches, database-reuse
/// replays, the report-time final replay, and generation cases under
/// nondeterministic handling, whose failing origins may be reported
/// unconfirmed — so those carry their diagnostics, while ordinary
/// exploration cases skip backtrace capture. A stamped replay executes the
/// whole body, so its capture holds the case's full trace — for a
/// concurrent-machine case, draws and notes made before
/// `stateful::Machine::run_concurrent` included.
///
/// Also returns the caught panic payload for an `Interesting` result, so a
/// final replay's caller can re-raise the test's *own* panic as the run's
/// closing unwind instead of synthesizing one.
pub(crate) fn run_test_case(
    c_tc: CTestCase,
    test_fn: &mut dyn FnMut(TestCase),
    is_final: bool,
    verbosity: Verbosity,
    output: &RunOutput,
    case_sink: Option<crate::test_case::OutputSink>,
) -> (
    TestCaseResult,
    Option<Box<dyn std::any::Any + Send>>,
    Option<String>,
) {
    let verbose = matches!(verbosity, Verbosity::Verbose | Verbosity::Debug);
    let quiet = verbosity == Verbosity::Quiet;
    let stamped = !is_final && c_tc.should_capture();
    let should_emit = ((is_final || stamped) && !quiet) || verbose;
    CAPTURE_BACKTRACE.with(|c| c.set(should_emit));
    // Drop any capture left over from a previous test case on this thread
    // (e.g. a body that caught its own panic and then passed): a later panic
    // that skips the hook via resume_unwind must not inherit it.
    take_panic_info();

    let c_tc = Arc::new(c_tc);
    let tc = TestCase::new(
        Arc::clone(&c_tc),
        should_emit,
        case_sink.or_else(|| output.sink().cloned()),
    );
    let reporter = tc.reporter();
    let result = with_test_context(|| catch_unwind(AssertUnwindSafe(|| test_fn(tc))));
    reporter.emit_rendered_output();

    let (tc_result, payload, diagnostic) = match result {
        Ok(()) => (TestCaseResult::Valid, None, None),
        Err(e) if e.downcast_ref::<AssumeFailed>().is_some() => {
            (TestCaseResult::Invalid, None, None)
        }
        Err(e) if e.downcast_ref::<StopTest>().is_some() => (TestCaseResult::Overrun, None, None),
        Err(e) if e.downcast_ref::<LoopDone>().is_some() => (TestCaseResult::Valid, None, None),
        Err(e) => {
            let e = match e.downcast::<InvalidArgument>() {
                Ok(invalid) => panic!("{}", invalid.0),
                Err(e) => e,
            };
            let e = match e.downcast::<InternalError>() {
                Ok(internal) => panic!("{}", internal.0),
                Err(e) => e,
            };
            let (thread_name, thread_id, location, backtrace) =
                take_panic_info().unwrap_or_else(unknown_panic_info);

            let captured = if (is_final || stamped) && !quiet {
                let msg = panic_message(&e);
                let diagnostic =
                    render_diagnostic(&thread_name, &thread_id, &location, &msg, &backtrace);
                if verbose && stamped {
                    output.block(&diagnostic);
                }
                Some(diagnostic)
            } else {
                if verbose {
                    let msg = panic_message(&e);
                    let diagnostic =
                        render_diagnostic(&thread_name, &thread_id, &location, &msg, &backtrace);
                    for line in diagnostic.trim_end_matches('\n').split('\n') {
                        output.line(line);
                    }
                }
                None
            };
            let failure = TestCaseResult::Interesting(Failure {
                origin: format!("Panic at {}", location),
            });
            (failure, Some(e), captured)
        }
    };

    if verbose {
        emit_verbose_stop_reason(&tc_result, output);
    }

    report_outcome(&c_tc, &tc_result);

    (tc_result, payload, diagnostic)
}

/// Report a test case's outcome to the engine over the C ABI.
///
/// Only the status — and, for an interesting (failing) case, the bug origin —
/// crosses the boundary; the panic message and reproduce blob are recovered
/// from the run result afterward, not pushed through here. This reports on
/// the handle the body drew on (shared with it through the `Arc`);
/// `hegel_mark_complete` waits for any in-flight operation on the handle, so
/// a still-running leaked thread cannot make this report fail.
fn report_outcome(handle: &CTestCase, result: &TestCaseResult) {
    use crate::ffi::sys::hegel_status_t as Status;
    let (status, origin) = match result {
        TestCaseResult::Valid => (Status::HEGEL_STATUS_VALID, None),
        TestCaseResult::Invalid => (Status::HEGEL_STATUS_INVALID, None),
        TestCaseResult::Overrun => (Status::HEGEL_STATUS_OVERRUN, None),
        TestCaseResult::Interesting(failure) => (
            Status::HEGEL_STATUS_INTERESTING,
            Some(failure.origin.as_str()),
        ),
    };
    handle
        .mark_complete(status, origin)
        .unwrap_or_else(|rc| crate::test_case::raise_for_rc(rc));
}

/// Print a per-test-case line describing why this test case stopped.
fn emit_verbose_stop_reason(result: &TestCaseResult, output: &RunOutput) {
    match result {
        TestCaseResult::Invalid => {
            output.line("Test case stopped: failed assumption");
        }
        TestCaseResult::Overrun => {
            output.line("Test case stopped: out of data");
        }
        TestCaseResult::Valid | TestCaseResult::Interesting(_) => {}
    }
}

/// Render a failure's diagnostic block. Mirrors the default Rust
/// panic-handler output so each failure in the report looks like a
/// stand-alone test failure.
fn render_diagnostic(
    thread_name: &str,
    thread_id: &str,
    location: &str,
    msg: &str,
    backtrace: &Backtrace,
) -> String {
    let mut out = String::new();
    out.push_str(&format!(
        "thread '{}' ({}) panicked at {}:\n",
        thread_name, thread_id, location
    ));
    out.push_str(msg);
    out.push('\n');
    // nocov start
    if backtrace.status() == BacktraceStatus::Captured {
        let is_full = std::env::var("RUST_BACKTRACE")
            .map(|v| v == "full")
            .unwrap_or(false);
        let formatted = format_backtrace(backtrace, is_full);
        out.push_str(&format!("stack backtrace:\n{}\n", formatted));
        if !is_full {
            out.push_str(
                "note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.\n",
            );
        }
    }
    // nocov end
    out
}

/// The copy-pasteable reproduce failure text to append after a failure's diagnostic,
/// or `None` when nothing should be printed.
///
/// `Some` only when [`Settings::print_blob`](crate::Settings::print_blob) is
/// enabled *and* the failure carries a reproduce blob. A blobless failure —
/// an unconfirmed nondeterministic failure reported caveat-only, a blob
/// replay (the caller already holds the blob) — prints nothing.
fn reproducer_line(settings: &Settings, reproduce_blob: Option<&str>) -> Option<String> {
    if !settings.print_blob {
        return None;
    }
    let blob = reproduce_blob?;
    Some(format!(
        "\nTo reproduce this failure, add the attribute below \
         #[hegel::test]:\n    #[hegel::reproduce_failure(\"{blob}\")]"
    ))
}

/// One origin's failure report material. Captures are ranked
/// ([`capture_rank`]): a rendered diagnostic over draw/note lines over a
/// bare capture, and a new interesting case replaces the stored capture
/// only at rank >= stored ([`store_capture`]). An unstamped shrink or
/// measurement probe's bare capture thus cannot clobber a stamped
/// diagnostic or draw lines, while at equal rank the newer wins: final
/// replay over confirmation over discovery. The panic payload travels with its capture, so the re-raised
/// panic always matches the printed diagnostic. When the final replay is
/// dry, the printed lines are the freshest stamped *failing* execution
/// (usually confirmation-time, pre-shrink values) while the blob and
/// choice sequence carry the shrunk incumbent. Printing is deferred
/// to the run verdict (verbose runs stream the lines live too), because a
/// frontend-interesting report can lose silently to an engine-side family
/// conclusion, and because a captured bug may not be reported at all —
/// the engine's confirmation may reject it, and the run may pass.
struct CapturedReport {
    /// The buffered draw/note lines of the case (empty under
    /// [`Verbosity::Quiet`], where nothing would be printed), already in
    /// report order: the case's document renders with each round's output
    /// grouped worker by worker.
    lines: Vec<String>,
    /// The rendered panic diagnostic (thread, location, message, backtrace);
    /// `None` under quiet and for unstamped, non-final cases.
    diagnostic: Option<String>,
    /// The caught panic payload, re-raised as the run's closing unwind.
    payload: Box<dyn std::any::Any + Send>,
}

fn capture_rank(diagnostic: &Option<String>, lines: &[String]) -> u8 {
    match (diagnostic.is_some(), !lines.is_empty()) {
        (true, _) => 2,
        (false, true) => 1,
        (false, false) => 0,
    }
}

/// Store `report` as `origin`'s capture unless a higher-ranked capture
/// already stands (see [`CapturedReport`]).
fn store_capture(
    captured: &mut HashMap<String, CapturedReport>,
    origin: String,
    report: CapturedReport,
) {
    match captured.entry(origin) {
        std::collections::hash_map::Entry::Vacant(entry) => {
            entry.insert(report);
        }
        std::collections::hash_map::Entry::Occupied(mut entry) => {
            let stored = entry.get();
            if capture_rank(&report.diagnostic, &report.lines)
                >= capture_rank(&stored.diagnostic, &stored.lines)
            {
                entry.insert(report);
            }
        }
    }
}

/// Drive a libhegel run to completion against the user's test function.
///
/// Installs the cross-backend panic hook, starts the engine through the C ABI
/// (`hegel_run_start`), and pulls each test case the engine schedules
/// (`hegel_next_test_case`), wrapping every one in [`run_test_case`]. The
/// engine owns the whole exploration — generation, shrinking, and the final
/// replay of every failure it is about to report — so every pumped case is
/// non-final, and the report material is captured as the cases run: each
/// interesting case's buffered output, diagnostic, and panic payload is
/// kept per origin, the highest-ranked capture winning, newest at equal
/// rank (see [`CapturedReport`]).
///
/// Once the loop drains, a failed verdict prints each distinct failure as
/// one grouped block — the captured lines and diagnostic, the caveat when
/// the run handled nondeterminism, and the reproducer line when the
/// failure carries a blob — and ends by re-raising the failing test's own
/// panic (or, for several distinct bugs, a panic carrying the count). A
/// run-level error — a failed health check, a nondeterminism abort under
/// `error` strictness, an engine panic — surfaces with the engine's own
/// message instead of the `Property test failed:` framing.
pub(crate) fn drive<F>(
    test_fn: F,
    settings: &Settings,
    database_key: Option<&str>,
    test_location: Option<&TestLocation>,
) where
    F: FnMut(TestCase),
{
    init_panic_hook();
    let mut test_fn = test_fn;
    let output = RunOutput::resolve();

    let c_settings = SettingsHandle::build(settings, database_key, test_location);
    let run = match RunHandle::start(&c_settings, output.sink()) {
        Ok(run) => run,
        Err(message) => panic!("{message}"), // nocov
    };

    drive_run(run, &mut test_fn, settings, &output, None);
}

/// The shared pump-and-report loop behind [`drive`] and
/// [`drive_blob_replay`]: pull every test case off `run`, capture report
/// material per origin, then act on the run's verdict. A passing run
/// returns — unless `stale_message` is set (a blob replay), where a pass
/// means the blob did not reproduce and panics with that message.
///
/// `#[inline(always)]`: without it the test body's call stops being
/// inlined and panic backtraces lose the user closure's source location
/// (pinned by `test_output`'s backtrace tests).
#[inline(always)]
fn drive_run<F: FnMut(TestCase)>(
    run: RunHandle,
    test_fn: &mut F,
    settings: &Settings,
    output: &RunOutput,
    stale_message: Option<&str>,
) {
    let verbosity = settings.verbosity;
    let quiet = verbosity == Verbosity::Quiet;
    let verbose = matches!(verbosity, Verbosity::Verbose | Verbosity::Debug);
    let mut captured: HashMap<String, CapturedReport> = HashMap::new();
    while let Some(c_tc) = run.next_test_case() {
        let buffer: Arc<std::sync::Mutex<Vec<String>>> = Arc::default();
        let live: Option<RunOutput> = if verbose { Some(output.clone()) } else { None };
        let case_sink: Option<crate::test_case::OutputSink> = if quiet {
            None
        } else {
            let buffer = Arc::clone(&buffer);
            Some(Arc::new(move |line: &str| {
                if let Some(live) = &live {
                    live.line(line);
                }
                buffer
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .push(line.to_string());
            }))
        };
        let (tc_result, payload, diagnostic) =
            run_test_case(c_tc, test_fn, false, verbosity, output, case_sink);
        if let TestCaseResult::Interesting(failure) = &tc_result {
            let records = std::mem::take(&mut *buffer.lock().unwrap_or_else(|e| e.into_inner()));
            store_capture(
                &mut captured,
                failure.origin.clone(),
                CapturedReport {
                    lines: records,
                    diagnostic,
                    payload: payload.expect("an interesting case carries the caught panic payload"),
                },
            );
        }
    }

    let result = run.result();
    use crate::ffi::sys::hegel_run_status_t as RunStatus;
    match result.status() {
        RunStatus::HEGEL_RUN_STATUS_PASSED => {
            if let Some(message) = stale_message {
                panic!("{message}");
            }
        }
        RunStatus::HEGEL_RUN_STATUS_ERROR => {
            let message = result
                .error()
                .unwrap_or_else(|| "the run failed with an unknown error".to_string());
            panic!("{message}");
        }
        RunStatus::HEGEL_RUN_STATUS_FAILED => {
            let count = result.failure_count();
            let multiple = count > 1;
            if multiple && !quiet {
                output.line(&format!(
                    "Property-based test failed with {count} distinct failures."
                ));
            }
            let mut last_payload: Option<Box<dyn std::any::Any + Send>> = None;
            for index in 0..count {
                if multiple && !quiet {
                    output.line("");
                }
                let failure = result.failure(index);
                let report = captured
                    .remove(&failure.origin)
                    .unwrap_or_else(|| hegel_internal_error!("failure {index} has no capture"));
                for line in &report.lines {
                    output.line(line);
                }
                if let Some(diagnostic) = report.diagnostic {
                    output.block(&diagnostic);
                }
                if !quiet {
                    if let Some(caveat) = &failure.caveat {
                        output.line(&format!("note: {caveat}"));
                    }
                }
                if let Some(line) = reproducer_line(settings, failure.reproduce_blob.as_deref()) {
                    output.block(&format!("{line}\n"));
                }
                last_payload = Some(report.payload);
            }

            if multiple {
                std::panic::resume_unwind(Box::new(format!(
                    "Property-based test failed with {count} distinct failures."
                )));
            } else {
                std::panic::resume_unwind(
                    last_payload
                        .unwrap_or_else(|| hegel_internal_error!("a failed run has no failures")),
                );
            }
        }
    }
}

/// Replay a base64 failure blob through the C ABI (`hegel_run_start_blob`),
/// bypassing generation and shrinking.
///
/// The engine owns the replay: a deterministic blob replays its choices up
/// to four times, stopping at the first failure; a nondeterministic blob
/// replays its stored timelines until one fails, like database reuse. Decoding
/// failures (corrupt or incompatible blobs) panic with the engine's
/// diagnostic. A blob that decodes but does not reproduce panics naming
/// both hypotheses: the failure is fixed, or a nondeterministic one did not
/// recur within the replay budget. A reproduced failure re-raises the
/// test's own panic; a replayed example has no fresh blob to print.
pub(crate) fn drive_blob_replay<F>(
    test_fn: F,
    settings: &Settings,
    database_key: Option<&str>,
    blob: &str,
    test_location: Option<&TestLocation>,
) where
    F: FnMut(TestCase),
{
    init_panic_hook();
    let mut test_fn = test_fn;
    let output = RunOutput::resolve();
    let c_settings = SettingsHandle::build(settings, database_key, test_location);
    let run = RunHandle::start_blob(&c_settings, blob, output.sink());
    drive_run(
        run,
        &mut test_fn,
        settings,
        &output,
        Some(
            "reproduce_failure: the supplied failure blob did not reproduce a \
             failure. The failure may have been fixed — or, for a nondeterministic \
             blob, it may not have recurred within the replay budget (a bug failing \
             10% of the time escapes it about 5% of the time). Re-run to retry, or \
             delete the attribute once the failure is fixed.",
        ),
    );
}

#[cfg(test)]
#[path = "../tests/embedded/run_lifecycle_tests.rs"]
mod tests;