supercov_engine/

python_evidence.rs

//! Validation and normalization of the Python runtime's evidence records.
//!
//! Each Supercov-hooked interpreter publishes commit-framed JSON records into
//! its own mmap: the process identity, every phase it entered (with the exact
//! test identity that phase stands for), runner outcomes, first-sighting hits,
//! decision vectors and any measurement limitation the runtime detected. The
//! one-byte commit marker is written last, so records completed before a hard
//! kill remain readable while a torn tail stays inert. Rust joins those records
//! into the shared frontend protocol; the runtime never computes a verdict.

use std::{
    collections::{BTreeMap, BTreeSet},
    fs::{self, File},
    path::{Component, Path, PathBuf},
};

use memmap2::{Mmap, MmapOptions};
use serde::Deserialize;
use serde_json::json;
use sha2::{Digest, Sha256};
use supercov_contracts::{
    AttributionPrecision, ExecutionModel, FrontendAttribution, FrontendLimitation,
    FrontendLimitationScope, FrontendRunDeclaration, FrontendRunnerDeclaration,
    LANGUAGE_FRONTEND_PROTOCOL_VERSION, StructuralSource,
};

use crate::{
    coverage_analysis::McdcVector,
    coverage_report::{
        CoverageManifest, CoverageModelDeclaration, CoveragePhase, CoverageReportRequest,
        DecisionMeta, DecisionSnapshot, ExecutionScope, ExitCodeInput, PersistedCoverageModel,
        RawTestResult, RuntimeEvent, RuntimeSnapshot, TestProvenance,
    },
    evidence_archive::EvidenceArchiveEntry,
};

pub const PYTHON_EVIDENCE_VERSION: u32 = 1;
pub const PYTHON_FRONTEND_VERSION: &str = "python-monitoring-v1";
pub const PYTEST_RUNNER: &str = "pytest";
pub const UNITTEST_RUNNER: &str = "unittest";

const TRANSPORT_MAGIC: &[u8; 8] = b"SCVPYTH1";
const TRANSPORT_VERSION: u32 = 1;
const TRANSPORT_HEADER_SIZE: usize = 64;
const TRANSPORT_RECORD_HEADER_SIZE: usize = 16;
const TRANSPORT_MAX_RECORD_SIZE: usize = 4 * 1024 * 1024;

fn default_runner() -> String {
    PYTEST_RUNNER.into()
}

/// Every field the runtime writes is named so `deny_unknown_fields` keeps
/// the record shape frozen, even where Rust does not read the value yet.
#[derive(Debug, Deserialize)]
#[allow(dead_code)]
#[serde(tag = "t", rename_all = "lowercase", deny_unknown_fields)]
enum Record {
    Process {
        v: u32,
        run: String,
        pid: u64,
        worker: String,
        python: String,
        executable: String,
        argv: Vec<String>,
    },
    Worker {
        worker: String,
    },
    Phase {
        ctx: u64,
        at: i64,
        worker: String,
        test: String,
        retry: usize,
        phase: String,
    },
    Outcome {
        worker: String,
        test: String,
        retry: usize,
        phase: String,
        outcome: String,
        xfail: bool,
        #[serde(default = "default_runner")]
        runner: String,
        /// Where the runner says the test is defined. Absent for adapters or
        /// synthesised tests that cannot name a file.
        #[serde(default)]
        file: Option<String>,
    },
    Hit {
        ctx: u64,
        id: String,
    },
    Dec {
        ctx: u64,
        id: String,
        v: String,
        o: u8,
    },
    /// The first assertion of a call phase: what the context recorded before
    /// this record is the assertion's evidence too.
    Assert {
        ctx: u64,
    },
    /// One assertion site a call phase reached, once per site per test.
    /// `unittest` reports the caller's frame and pytest reports the line its
    /// rewriter recorded; both are resolved against the syntax inventory, and
    /// a frame naming no inventoried site witnesses nothing.
    Asite {
        ctx: u64,
        f: String,
        l: usize,
    },
    Limitation {
        id: String,
        reason: String,
        #[serde(default)]
        file: Option<String>,
        #[serde(default)]
        obligation: Option<String>,
    },
    Exit {
        at: i64,
    },
}

#[derive(Debug)]
pub enum PythonEvidenceError {
    Io(String),
    UnsafeEntry(String),
    InvalidRecord {
        file: String,
        line: usize,
        reason: String,
    },
    InvalidTransport {
        file: String,
        reason: String,
    },
    DroppedRecords {
        file: String,
        count: u64,
    },
    RunMismatch {
        expected: String,
        actual: String,
    },
    UnsupportedVersion(u32),
    UnknownContext {
        file: String,
        line: usize,
        context: u64,
    },
    UnknownObligation(String),
    InvalidVector {
        id: String,
        expected: usize,
        actual: usize,
    },
    NoInterpreter,
    NoTests,
    UnsupportedPython(String),
}

impl std::fmt::Display for PythonEvidenceError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Io(reason) => write!(formatter, "could not read Python evidence: {reason}"),
            Self::UnsafeEntry(name) => write!(formatter, "unsafe Python evidence entry: {name}"),
            Self::InvalidRecord { file, line, reason } => {
                write!(formatter, "invalid Python evidence record {file}:{line}: {reason}")
            }
            Self::InvalidTransport { file, reason } => {
                write!(formatter, "invalid Python evidence transport {file}: {reason}")
            }
            Self::DroppedRecords { file, count } => write!(
                formatter,
                "Python evidence transport {file} exhausted its bounded capacity and dropped {count} record(s)"
            ),
            Self::RunMismatch { expected, actual } => write!(
                formatter,
                "Python evidence belongs to run {actual}, expected {expected}"
            ),
            Self::UnsupportedVersion(version) => {
                write!(formatter, "unsupported Python evidence version {version}")
            }
            Self::UnknownContext { file, line, context } => write!(
                formatter,
                "Python evidence {file}:{line} references undeclared context {context}"
            ),
            Self::UnknownObligation(id) => {
                write!(formatter, "Python runtime reported an unknown obligation: {id}")
            }
            Self::InvalidVector {
                id,
                expected,
                actual,
            } => write!(
                formatter,
                "Python decision {id} reported {actual} condition values, expected {expected}"
            ),
            Self::NoInterpreter => formatter.write_str(
                "no Supercov-hooked Python interpreter ran: the test command did not start CPython 3.12+ with Supercov's start-up hook (PYTHONPATH may be ignored by -I/-E/-S, or the runner is not Python)",
            ),
            Self::NoTests => formatter.write_str(
                "the Python run produced no test outcomes; Supercov measures Python through pytest and unittest",
            ),
            Self::UnsupportedPython(version) => write!(
                formatter,
                "Supercov measures CPython 3.12 or newer; the test command ran Python {version}"
            ),
        }
    }
}

impl std::error::Error for PythonEvidenceError {}

fn stable_id(prefix: &str, values: &[&str]) -> String {
    let mut hash = Sha256::new();
    for value in values {
        hash.update(value.as_bytes());
        hash.update([0]);
    }
    let digest = hash.finalize();
    let mut encoded = String::with_capacity(prefix.len() + 25);
    encoded.push_str(prefix);
    encoded.push(':');
    for byte in &digest[..12] {
        use std::fmt::Write as _;
        write!(&mut encoded, "{byte:02x}").expect("string formatting");
    }
    encoded
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
struct Identity {
    worker: String,
    test: String,
    retry: usize,
    phase: String,
}

type ObservedVectors = BTreeSet<(Vec<Option<bool>>, bool)>;
/// (worker, test, retry) -> [(phase, outcome, xfail)]
type OutcomesByAttempt = BTreeMap<(String, String, usize), Vec<(String, String, bool)>>;
/// (worker, test, retry) -> runner that reported the attempt
type RunnersByAttempt = BTreeMap<(String, String, usize), String>;
/// (worker, test, retry) -> the source file the runner named for the test
type TestFilesByAttempt = BTreeMap<(String, String, usize), String>;
/// (worker, test, retry) -> assertion sites the call phase reached, in the
/// order they were first seen, as the runtime reported them: (path, line)
type SitesByAttempt = BTreeMap<(String, String, usize), Vec<(String, usize)>>;

/// The assertion sites Supercov inventoried from source before the run,
/// indexed so a runtime frame can name one exactly.
///
/// Python reports a file and a line for an assertion, while an assertion
/// anchor is a file, line and column. The inventory supplies the missing
/// column and validates the frame: one that names no inventoried site
/// witnesses nothing, so a wrong frame loses a witness rather than inventing
/// one.
pub struct PythonAssertionInventory {
    root: PathBuf,
    /// (project-relative file, line) -> the sites on that line
    columns: BTreeMap<(String, usize), Vec<usize>>,
}

impl PythonAssertionInventory {
    pub fn new(root: &Path, inputs: &crate::assertion_map::Inputs) -> Self {
        let mut columns = BTreeMap::<(String, usize), Vec<usize>>::new();
        for site in &inputs.assertions {
            columns
                .entry((site.at.file.clone(), site.at.line))
                .or_default()
                // Every native manifest reports a zero-based byte column and
                // the report adds one to reach the anchor's own column.
                .push(site.at.column.saturating_sub(1));
        }
        for sites in columns.values_mut() {
            sites.sort_unstable();
            sites.dedup();
        }
        Self {
            root: root.to_path_buf(),
            columns,
        }
    }

    /// An inventory with no sites: every frame names nothing, which is what a
    /// run with no assertion inputs should see.
    pub fn empty() -> Self {
        Self {
            root: PathBuf::new(),
            columns: BTreeMap::new(),
        }
    }

    /// Python reports both forms: a frame's `co_filename` is whatever the
    /// interpreter loaded, absolute or relative. A path outside the project
    /// names nothing here.
    pub fn relative(&self, path: &str) -> Option<String> {
        let candidate = Path::new(path);
        let relative = if candidate.is_absolute() {
            candidate.strip_prefix(&self.root).ok()?
        } else {
            candidate.strip_prefix("./").unwrap_or(candidate)
        };
        let text = relative.to_string_lossy().replace('\\', "/");
        (!text.is_empty() && !text.starts_with("../")).then_some(text)
    }

    /// `file:line:column` when that line holds exactly one inventoried site.
    /// Two assertions on one line cannot be told apart from a line number, so
    /// the frame names neither rather than guessing between them.
    pub fn locate(&self, path: &str, line: usize) -> Option<String> {
        let file = self.relative(path)?;
        match self.columns.get(&(file.clone(), line))?.as_slice() {
            [column] => Some(format!("{file}:{line}:{column}")),
            _ => None,
        }
    }
}

#[derive(Debug, Default)]
struct Observations {
    hits: BTreeSet<String>,
    vectors: BTreeMap<String, ObservedVectors>,
}

#[derive(Debug, Clone)]
struct RuntimeLimitation {
    id: String,
    reason: String,
    file: Option<String>,
    obligation: Option<String>,
}

#[derive(Debug, Default)]
struct Evidence {
    interpreters: usize,
    python_versions: BTreeSet<String>,
    per_identity: BTreeMap<Identity, Observations>,
    background: BTreeMap<String, Observations>,
    outcomes: OutcomesByAttempt,
    runners: RunnersByAttempt,
    test_files: TestFilesByAttempt,
    sites: SitesByAttempt,
    limitations: Vec<RuntimeLimitation>,
}

fn read_evidence_directory(
    directory: &Path,
    run_id: &str,
) -> Result<Evidence, PythonEvidenceError> {
    let mut evidence = Evidence::default();
    let mut files = match fs::read_dir(directory) {
        Ok(entries) => entries
            .collect::<Result<Vec<_>, _>>()
            .map_err(|error| PythonEvidenceError::Io(error.to_string()))?,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
        Err(error) => return Err(PythonEvidenceError::Io(error.to_string())),
    };
    files.sort_by_key(|entry| entry.file_name());
    for entry in files {
        let name = entry
            .file_name()
            .into_string()
            .map_err(|_| PythonEvidenceError::UnsafeEntry("<non-utf8>".into()))?;
        if Path::new(&name)
            .components()
            .any(|component| !matches!(component, Component::Normal(_)))
            || !name.ends_with(".mmap")
        {
            return Err(PythonEvidenceError::UnsafeEntry(name));
        }
        let metadata = fs::symlink_metadata(entry.path())
            .map_err(|error| PythonEvidenceError::Io(error.to_string()))?;
        if !metadata.file_type().is_file() {
            return Err(PythonEvidenceError::UnsafeEntry(name));
        }
        let file =
            File::open(entry.path()).map_err(|error| PythonEvidenceError::Io(error.to_string()))?;
        // The file is immutable from Supercov's perspective after the wrapped
        // interpreter has exited. No mutable alias is created while this map
        // is alive.
        let contents = unsafe { MmapOptions::new().map(&file) }
            .map_err(|error| PythonEvidenceError::Io(error.to_string()))?;
        read_evidence_file(&name, &contents, run_id, &mut evidence)?;
    }
    Ok(evidence)
}

fn transport_u32(bytes: &[u8], offset: usize) -> Option<u32> {
    bytes
        .get(offset..offset + 4)
        .and_then(|value| value.try_into().ok())
        .map(u32::from_le_bytes)
}

fn transport_u64(bytes: &[u8], offset: usize) -> Option<u64> {
    bytes
        .get(offset..offset + 8)
        .and_then(|value| value.try_into().ok())
        .map(u64::from_le_bytes)
}

fn transport_checksum(payload: &[u8]) -> u32 {
    payload.iter().fold(0x811c_9dc5_u32, |value, byte| {
        (value ^ u32::from(*byte)).wrapping_mul(0x0100_0193)
    })
}

fn align_transport(value: usize) -> Option<usize> {
    value.checked_add(7).map(|value| value & !7)
}

fn read_evidence_file(
    name: &str,
    contents: &Mmap,
    run_id: &str,
    evidence: &mut Evidence,
) -> Result<(), PythonEvidenceError> {
    let invalid_transport = |reason: &str| PythonEvidenceError::InvalidTransport {
        file: name.into(),
        reason: reason.into(),
    };
    if contents.len() < TRANSPORT_HEADER_SIZE
        || contents.get(..8) != Some(TRANSPORT_MAGIC.as_slice())
        || transport_u32(contents, 8) != Some(TRANSPORT_VERSION)
        || transport_u32(contents, 12) != Some(TRANSPORT_HEADER_SIZE as u32)
    {
        return Err(invalid_transport("header or version does not match"));
    }
    let declared_capacity =
        transport_u64(contents, 16).ok_or_else(|| invalid_transport("capacity is missing"))?;
    if declared_capacity < TRANSPORT_HEADER_SIZE as u64 || declared_capacity > contents.len() as u64
    {
        return Err(invalid_transport(
            "declared capacity is outside the mapped file",
        ));
    }
    let dropped =
        transport_u64(contents, 24).ok_or_else(|| invalid_transport("drop counter is missing"))?;
    if dropped != 0 {
        return Err(PythonEvidenceError::DroppedRecords {
            file: name.into(),
            count: dropped,
        });
    }
    let transport_pid = transport_u64(contents, 32)
        .filter(|pid| *pid != 0)
        .ok_or_else(|| invalid_transport("process id is missing"))?;
    let mut contexts = BTreeMap::<u64, Identity>::new();
    // What each call phase recorded so far, kept until its first assertion
    // marker moves it to the phase's assertion identity.
    let mut before_assertion = BTreeMap::<u64, Observations>::new();
    let mut process_worker: Option<String> = None;
    let mut cursor = TRANSPORT_HEADER_SIZE;
    let mut record_index = 0;
    while cursor + TRANSPORT_RECORD_HEADER_SIZE <= contents.len() {
        let commit = contents[cursor];
        if commit == 0 {
            // Payload bytes can exist after a killed writer, but an absent
            // commit byte makes that frame and every later zeroed frame inert.
            break;
        }
        record_index += 1;
        let line_number = record_index;
        let invalid = |reason: &str| PythonEvidenceError::InvalidRecord {
            file: name.into(),
            line: line_number,
            reason: reason.into(),
        };
        if commit != 1
            || contents[cursor + 1..cursor + 4] != [0, 0, 0]
            || contents[cursor + 12..cursor + 16] != [0, 0, 0, 0]
        {
            return Err(invalid("commit marker or reserved bytes are invalid"));
        }
        let length = transport_u32(contents, cursor + 4)
            .map(|value| value as usize)
            .ok_or_else(|| invalid("payload length is missing"))?;
        if length == 0 || length > TRANSPORT_MAX_RECORD_SIZE {
            return Err(invalid("payload length is outside the transport bound"));
        }
        let payload_start = cursor + TRANSPORT_RECORD_HEADER_SIZE;
        let payload_end = payload_start
            .checked_add(length)
            .filter(|end| *end <= contents.len())
            .ok_or_else(|| invalid("payload extends past the mapped file"))?;
        let next_cursor = align_transport(payload_end)
            .filter(|end| *end <= contents.len())
            .ok_or_else(|| invalid("aligned frame extends past the mapped file"))?;
        if contents[payload_end..next_cursor]
            .iter()
            .any(|byte| *byte != 0)
        {
            return Err(invalid("frame padding is not zero"));
        }
        let payload = &contents[payload_start..payload_end];
        let expected_checksum = transport_u32(contents, cursor + 8)
            .ok_or_else(|| invalid("payload checksum is missing"))?;
        if transport_checksum(payload) != expected_checksum {
            return Err(invalid("payload checksum does not match"));
        }
        let record: Record = serde_json::from_slice(payload).map_err(|error| {
            PythonEvidenceError::InvalidRecord {
                file: name.into(),
                line: line_number,
                reason: error.to_string(),
            }
        })?;
        match record {
            Record::Process {
                v,
                run,
                pid,
                worker,
                python,
                ..
            } => {
                if v != PYTHON_EVIDENCE_VERSION {
                    return Err(PythonEvidenceError::UnsupportedVersion(v));
                }
                if run != run_id {
                    return Err(PythonEvidenceError::RunMismatch {
                        expected: run_id.into(),
                        actual: run,
                    });
                }
                if pid != transport_pid {
                    return Err(invalid("process record does not match the transport owner"));
                }
                let supported = python
                    .split('.')
                    .take(2)
                    .map(|part| part.parse::<u32>().ok())
                    .collect::<Option<Vec<_>>>()
                    .is_some_and(|parts| parts.len() == 2 && (parts[0], parts[1]) >= (3, 12));
                if !supported {
                    return Err(PythonEvidenceError::UnsupportedPython(python));
                }
                evidence.interpreters += 1;
                evidence.python_versions.insert(python);
                process_worker = Some(worker);
            }
            Record::Worker { worker } => process_worker = Some(worker),
            Record::Phase {
                ctx,
                worker,
                test,
                retry,
                phase,
                ..
            } => {
                if ctx == 0 {
                    return Err(invalid("phase context 0 is reserved for background"));
                }
                if !matches!(phase.as_str(), "setup" | "call" | "teardown") {
                    return Err(invalid("unknown pytest phase"));
                }
                if test.trim().is_empty() || worker.trim().is_empty() {
                    return Err(invalid("phase identity must name a worker and test"));
                }
                if phase == "call" {
                    before_assertion.insert(ctx, Observations::default());
                }
                contexts.insert(
                    ctx,
                    Identity {
                        worker,
                        test,
                        retry,
                        phase,
                    },
                );
            }
            Record::Outcome {
                worker,
                test,
                retry,
                phase,
                outcome,
                xfail,
                runner,
                file,
            } => {
                if !matches!(phase.as_str(), "setup" | "call" | "teardown") {
                    return Err(invalid("unknown test outcome phase"));
                }
                if !matches!(
                    outcome.as_str(),
                    "passed" | "failed" | "skipped" | "rerun" | "error"
                ) {
                    return Err(invalid("unknown test outcome"));
                }
                if !matches!(runner.as_str(), PYTEST_RUNNER | UNITTEST_RUNNER) {
                    return Err(invalid("unknown Python test runner"));
                }
                let key = (worker, test, retry);
                if let Some(previous) = evidence.runners.get(&key)
                    && previous != &runner
                {
                    return Err(invalid("one attempt was reported by two runners"));
                }
                evidence.runners.insert(key.clone(), runner);
                if let Some(file) = file.filter(|path| !path.is_empty()) {
                    evidence.test_files.entry(key.clone()).or_insert(file);
                }
                evidence
                    .outcomes
                    .entry(key)
                    .or_default()
                    .push((phase, outcome, xfail));
            }
            Record::Hit { ctx, id } => {
                if let Some(before) = before_assertion.get_mut(&ctx) {
                    before.hits.insert(id.clone());
                }
                observations(
                    evidence,
                    &contexts,
                    process_worker.as_deref(),
                    ctx,
                    name,
                    line_number,
                )?
                .hits
                .insert(id);
            }
            Record::Dec { ctx, id, v, o } => {
                if v.is_empty() || !v.bytes().all(|digit| matches!(digit, b'0' | b'1' | b'2')) {
                    return Err(invalid("decision vector digits must be 0, 1 or 2"));
                }
                if o > 1 {
                    return Err(invalid("decision outcome must be 0 or 1"));
                }
                let values = v
                    .bytes()
                    .map(|digit| match digit {
                        b'0' => None,
                        b'1' => Some(false),
                        _ => Some(true),
                    })
                    .collect::<Vec<_>>();
                if let Some(before) = before_assertion.get_mut(&ctx) {
                    before
                        .vectors
                        .entry(id.clone())
                        .or_default()
                        .insert((values.clone(), o == 1));
                }
                observations(
                    evidence,
                    &contexts,
                    process_worker.as_deref(),
                    ctx,
                    name,
                    line_number,
                )?
                .vectors
                .entry(id)
                .or_default()
                .insert((values, o == 1));
            }
            Record::Assert { ctx } => {
                // Only the first marker of a call phase moves anything; a
                // later one, or one outside a call phase, is inert.
                if let Some(before) = before_assertion.remove(&ctx) {
                    let identity =
                        contexts
                            .get(&ctx)
                            .ok_or(PythonEvidenceError::UnknownContext {
                                file: name.into(),
                                line: line_number,
                                context: ctx,
                            })?;
                    let asserted = evidence
                        .per_identity
                        .entry(Identity {
                            phase: "assertion".into(),
                            ..identity.clone()
                        })
                        .or_default();
                    asserted.hits.extend(before.hits);
                    for (id, vectors) in before.vectors {
                        asserted.vectors.entry(id).or_default().extend(vectors);
                    }
                }
            }
            Record::Asite { ctx, f, l } => {
                if f.is_empty() || l == 0 {
                    return Err(invalid("assertion site needs a file and a line"));
                }
                let identity = contexts
                    .get(&ctx)
                    .ok_or(PythonEvidenceError::UnknownContext {
                        file: name.into(),
                        line: line_number,
                        context: ctx,
                    })?;
                // Only the call phase witnesses a test's assertions; setup and
                // teardown assertions belong to no single site under test.
                if identity.phase == "call" {
                    let key = (
                        identity.worker.clone(),
                        identity.test.clone(),
                        identity.retry,
                    );
                    let sites = evidence.sites.entry(key).or_default();
                    let site = (f, l);
                    if !sites.contains(&site) {
                        sites.push(site);
                    }
                }
            }
            Record::Limitation {
                id,
                reason,
                file,
                obligation,
            } => evidence.limitations.push(RuntimeLimitation {
                id,
                reason,
                file,
                obligation,
            }),
            Record::Exit { .. } => {}
        }
        cursor = next_cursor;
    }
    Ok(())
}

fn observations<'a>(
    evidence: &'a mut Evidence,
    contexts: &BTreeMap<u64, Identity>,
    process_worker: Option<&str>,
    context: u64,
    file: &str,
    line: usize,
) -> Result<&'a mut Observations, PythonEvidenceError> {
    if context == 0 {
        return Ok(evidence
            .background
            .entry(process_worker.unwrap_or("main").to_owned())
            .or_default());
    }
    let identity = contexts
        .get(&context)
        .ok_or(PythonEvidenceError::UnknownContext {
            file: file.into(),
            line,
            context,
        })?;
    Ok(evidence.per_identity.entry(identity.clone()).or_default())
}

pub fn python_coverage_model() -> CoverageModelDeclaration {
    CoverageModelDeclaration {
        language: "python".into(),
        variant: "python-owned-monitoring".into(),
        name: "python-sys-monitoring-v1".into(),
        completeness_meaning: "Every statement, function, decision vector, loop, short-circuit, match and exception-flow obligation Supercov derived from the source was observed through CPython's monitoring events with exact test identity; the declared runtime limitations remain separate.".into(),
        measured: vec![
            "executable statements proven by CPython LINE events on their header lines, or INSTRUCTION events when they share a line".into(),
            "function and lambda entry".into(),
            "boolean decision vectors with masking MC/DC from conditional-jump events".into(),
            "for-loop and comprehension zero-versus-entered iteration".into(),
            "logical and/or short-circuit alternatives".into(),
            "match case selection and guards".into(),
            "try completion, handler selection and exception propagation".into(),
            "pytest and unittest worker, test, retry and setup/call/teardown phase identity".into(),
            "evidence a test recorded before its first assertion, linked to that assertion when the test passes".into(),
        ],
        not_measured: vec![
            "zero-iteration executions of a loop after it has run and exited 16 times within one test phase on CPython 3.14".into(),
            "causal linkage to individual actions, or to any assertion after a test's first".into(),
            "code compiled from strings at runtime".into(),
            "causal test context for raw _thread or native-extension-created threads".into(),
            "child coverage outside subprocess.Popen and multiprocessing adapters".into(),
            "all input values, semantic partitions, paths, or concurrency interleavings".into(),
            "mutation score or assertion fault-detection strength".into(),
        ],
    }
}

fn phase_id(run: &str, identity: &Identity) -> String {
    stable_id(
        "python-phase",
        &[
            run,
            &identity.worker,
            &identity.test,
            &identity.retry.to_string(),
            &identity.phase,
        ],
    )
}

fn scope(run: &str, worker: &str, test: &str, retry: usize) -> ExecutionScope {
    ExecutionScope {
        version: 1,
        run_id: run.into(),
        worker_id: worker.into(),
        test_id: test.into(),
        test_key: stable_id("python-test", &[worker, test]),
        retry,
        attempt_id: stable_id("python-attempt", &[run, worker, test, &retry.to_string()]),
    }
}

struct ManifestIndex<'a> {
    points: BTreeSet<&'a str>,
    alternatives: BTreeSet<&'a str>,
    decisions: BTreeMap<&'a str, &'a DecisionMeta>,
    lines: BTreeMap<&'a str, (String, usize)>,
}

impl<'a> ManifestIndex<'a> {
    fn new(manifest: &'a CoverageManifest) -> Self {
        let mut lines = BTreeMap::new();
        for point in &manifest.points {
            lines.insert(point.id.as_str(), (point.file.clone(), point.line));
        }
        for decision in &manifest.decisions {
            lines.insert(decision.id.as_str(), (decision.file.clone(), decision.line));
        }
        for branch in &manifest.branches {
            lines.insert(branch.id.as_str(), (branch.file.clone(), branch.line));
        }
        Self {
            points: manifest
                .points
                .iter()
                .map(|point| point.id.as_str())
                .collect(),
            alternatives: manifest
                .branches
                .iter()
                .flat_map(|branch| branch.alternatives.iter().map(|alt| alt.id.as_str()))
                .collect(),
            decisions: manifest
                .decisions
                .iter()
                .map(|decision| (decision.id.as_str(), decision))
                .collect(),
            lines,
        }
    }
}

fn snapshot(
    index: &ManifestIndex<'_>,
    observations: &Observations,
    phase: &str,
) -> Result<RuntimeSnapshot, PythonEvidenceError> {
    let mut hits = BTreeSet::new();
    for id in &observations.hits {
        if !index.points.contains(id.as_str()) && !index.alternatives.contains(id.as_str()) {
            return Err(PythonEvidenceError::UnknownObligation(id.clone()));
        }
        hits.insert(id.clone());
    }
    let mut decisions = Vec::new();
    let mut events = Vec::new();
    let mut clock = 1;
    for id in &hits {
        events.push(RuntimeEvent {
            event_type: "hit".into(),
            id: id.clone(),
            vector: None,
            timestamp_ms: clock,
            phase_id: Some(phase.into()),
            statement_id: None,
            environment: "python".into(),
        });
        clock += 1;
    }
    for (id, vectors) in &observations.vectors {
        let Some(meta) = index.decisions.get(id.as_str()) else {
            return Err(PythonEvidenceError::UnknownObligation(id.clone()));
        };
        let mut observed = Vec::new();
        for (values, outcome) in vectors {
            if values.len() != meta.conditions.len() {
                return Err(PythonEvidenceError::InvalidVector {
                    id: id.clone(),
                    expected: meta.conditions.len(),
                    actual: values.len(),
                });
            }
            let vector = McdcVector {
                values: values.clone(),
                outcome: *outcome,
            };
            events.push(RuntimeEvent {
                event_type: "decision".into(),
                id: id.clone(),
                vector: Some(vector.clone()),
                timestamp_ms: clock,
                phase_id: Some(phase.into()),
                statement_id: None,
                environment: "python".into(),
            });
            clock += 1;
            observed.push(vector);
        }
        decisions.push(DecisionSnapshot {
            meta: (*meta).clone(),
            vectors: observed,
        });
    }
    Ok(RuntimeSnapshot {
        decisions,
        hits: hits.into_iter().collect(),
        events,
        logicals: Vec::new(),
    })
}

fn attempt_status(outcomes: &[(String, String, bool)]) -> String {
    if outcomes
        .iter()
        .any(|(_, outcome, _)| matches!(outcome.as_str(), "failed" | "rerun" | "error"))
    {
        "failed"
    } else if outcomes.iter().any(|(_, outcome, _)| outcome == "skipped") {
        "skipped"
    } else {
        "passed"
    }
    .into()
}

#[derive(Debug, Clone, PartialEq)]
pub struct PythonFrontendRun {
    pub declaration: FrontendRunDeclaration,
    pub request: CoverageReportRequest,
    pub tests: usize,
    pub interpreters: usize,
    pub python_versions: Vec<String>,
}

impl PythonFrontendRun {
    pub fn archive_entries(&self) -> Result<Vec<EvidenceArchiveEntry>, serde_json::Error> {
        let model = PersistedCoverageModel::from_declaration(
            self.request
                .coverage_model
                .as_ref()
                .expect("Python frontend always declares a coverage model"),
        )
        .expect("Python coverage model is contract-valid");
        let mut entries = vec![
            EvidenceArchiveEntry {
                path: "coverage-model.json".into(),
                contents: serde_json::to_vec(&model)?,
            },
            EvidenceArchiveEntry {
                path: "frontend.json".into(),
                contents: serde_json::to_vec(&self.declaration)?,
            },
            EvidenceArchiveEntry {
                path: "manifest.json".into(),
                contents: serde_json::to_vec(&self.request.manifest)?,
            },
        ];
        for (index, result) in self.request.raw_results.iter().enumerate() {
            entries.push(EvidenceArchiveEntry {
                path: format!("results/{index:08}/mcdc.json"),
                contents: serde_json::to_vec(result)?,
            });
        }
        Ok(entries)
    }
}

/// Join the runtime's evidence directory with the ahead-of-run manifest into
/// a protocol-conformant frontend run.
pub fn build_python_frontend_run(
    manifest: &CoverageManifest,
    evidence_directory: &Path,
    run_id: &str,
    generated_at: &str,
    test_exit_code: i32,
    assertions: &PythonAssertionInventory,
) -> Result<PythonFrontendRun, PythonEvidenceError> {
    let evidence = read_evidence_directory(evidence_directory, run_id)?;
    if evidence.interpreters == 0 {
        return Err(PythonEvidenceError::NoInterpreter);
    }
    if evidence.outcomes.is_empty() {
        return Err(PythonEvidenceError::NoTests);
    }
    let Evidence {
        interpreters,
        python_versions,
        per_identity,
        background,
        outcomes,
        runners,
        test_files,
        sites,
        limitations,
    } = evidence;
    let mut manifest = manifest.clone();
    let index = ManifestIndex::new(&manifest);

    let mut raw_results = Vec::new();
    let mut observed_runners = BTreeSet::new();
    let mut identities_by_attempt =
        BTreeMap::<(String, String, usize), Vec<(&Identity, &Observations)>>::new();
    for (identity, observations) in &per_identity {
        identities_by_attempt
            .entry((
                identity.worker.clone(),
                identity.test.clone(),
                identity.retry,
            ))
            .or_default()
            .push((identity, observations));
    }
    for ((worker, test, retry), mut outcomes) in outcomes {
        let runner = runners
            .get(&(worker.clone(), test.clone(), retry))
            .cloned()
            .unwrap_or_else(default_runner);
        let attempt_identities = identities_by_attempt
            .remove(&(worker.clone(), test.clone(), retry))
            .unwrap_or_default();
        observed_runners.insert(runner.clone());
        outcomes.sort_by_key(|(phase, _, _)| match phase.as_str() {
            "setup" => 0,
            "call" => 1,
            _ => 2,
        });
        let mut phases = Vec::new();
        let mut runtime = Vec::new();
        let mut observed_phases = BTreeSet::new();
        for (position, (phase_name, outcome, xfail)) in outcomes.iter().enumerate() {
            observed_phases.insert(phase_name.clone());
            let identity = Identity {
                worker: worker.clone(),
                test: test.clone(),
                retry,
                phase: phase_name.clone(),
            };
            let id = phase_id(run_id, &identity);
            phases.push(CoveragePhase {
                id: id.clone(),
                kind: match phase_name.as_str() {
                    "call" => "test",
                    value => value,
                }
                .into(),
                operation: format!("{runner} {phase_name}"),
                source: Some(test.clone()),
                caused_by_phase_id: None,
                started_at_ms: position as i64 * 2 + 1,
                ended_at_ms: Some(position as i64 * 2 + 2),
                status: Some(match outcome.as_str() {
                    "rerun" | "error" => "failed".into(),
                    value => value.into(),
                }),
                error: None,
            });
            if let Some((_, observations)) = attempt_identities
                .iter()
                .find(|(candidate, _)| candidate.phase == phase_name.as_str())
            {
                runtime.push(snapshot(&index, observations, &id)?);
            }
            if phase_name != "call" {
                continue;
            }
            // What the test recorded before its first assertion is that
            // assertion's evidence, linked when the phase passed outright:
            // a failed, skipped or expected-to-fail phase witnessed nothing.
            if let Some((identity, observations)) = attempt_identities
                .iter()
                .find(|(candidate, _)| candidate.phase == "assertion")
            {
                observed_phases.insert("assertion".to_owned());
                let id = phase_id(run_id, identity);
                phases.push(CoveragePhase {
                    id: id.clone(),
                    kind: "assertion".into(),
                    operation: format!("{runner} assertion"),
                    source: Some(test.clone()),
                    caused_by_phase_id: None,
                    started_at_ms: position as i64 * 2 + 1,
                    ended_at_ms: Some(position as i64 * 2 + 2),
                    status: Some(
                        if outcome == "passed" && !*xfail {
                            "passed"
                        } else {
                            "failed"
                        }
                        .into(),
                    ),
                    error: None,
                });
                runtime.push(snapshot(&index, observations, &id)?);
                // One phase per assertion site the call phase reached, so an
                // assertion map can tell the sites apart. The per-test phase
                // above keeps carrying the pre-assertion evidence; these are
                // witnesses only, and a site the inventory does not know is
                // skipped rather than guessed at.
                let attempt = (worker.clone(), test.clone(), retry);
                for (path, line) in sites.get(&attempt).into_iter().flatten() {
                    let Some(location) = assertions.locate(path, *line) else {
                        continue;
                    };
                    phases.push(CoveragePhase {
                        id: stable_id("python-assertion", &[run_id, &id, &location]),
                        kind: "assertion".into(),
                        operation: format!("{runner} assertion at {location}"),
                        source: Some(location),
                        caused_by_phase_id: Some(id.clone()),
                        started_at_ms: position as i64 * 2 + 1,
                        ended_at_ms: Some(position as i64 * 2 + 2),
                        status: Some(
                            if outcome == "passed" && !*xfail {
                                "passed"
                            } else {
                                "failed"
                            }
                            .into(),
                        ),
                        error: None,
                    });
                }
            }
        }
        // A phase the runtime entered but pytest never reported (the worker
        // died inside it) is a failed phase with its evidence kept.
        for (identity, observations) in attempt_identities {
            if !observed_phases.contains(&identity.phase) {
                let id = phase_id(run_id, identity);
                phases.push(CoveragePhase {
                    id: id.clone(),
                    kind: match identity.phase.as_str() {
                        "call" => "test",
                        value => value,
                    }
                    .into(),
                    operation: format!("{runner} {}", identity.phase),
                    source: Some(test.clone()),
                    caused_by_phase_id: None,
                    started_at_ms: phases.len() as i64 * 2 + 1,
                    ended_at_ms: None,
                    status: Some("failed".into()),
                    error: Some("the phase started but the runner reported no outcome".into()),
                });
                runtime.push(snapshot(&index, observations, &id)?);
            }
        }
        let status = if phases.iter().any(|phase| phase.error.is_some()) {
            "failed".into()
        } else {
            attempt_status(&outcomes)
        };
        raw_results.push(RawTestResult {
            test_id: Some(test.clone()),
            scope: Some(scope(run_id, &worker, &test, retry)),
            test: test.clone(),
            // What the runner named, as the project names it. A dotted
            // module path or a pytest node id is not a path, so the old
            // derivation stays only as a fallback.
            test_file: test_files
                .get(&(worker.clone(), test.clone(), retry))
                .and_then(|path| assertions.relative(path))
                .or_else(|| test.split("::").next().map(str::to_owned)),
            title: test.rsplit("::").next().map(str::to_owned),
            retry: Some(retry),
            status: Some(status),
            expected_status: Some(
                if outcomes.iter().any(|(_, _, xfail)| *xfail) {
                    "failed"
                } else {
                    "passed"
                }
                .into(),
            ),
            flaky: false,
            provenance: TestProvenance {
                runner: runner.clone(),
                kind: "unit".into(),
                project: None,
                source: PYTHON_FRONTEND_VERSION.into(),
            },
            role: "test".into(),
            attribution: crate::coverage_report::ATTRIBUTION_EXACT.into(),
            phases,
            runtime,
            browser: Vec::new(),
            server: Vec::new(),
        });
    }
    // Phases with observations whose test never produced any outcome at all
    // (for example a worker killed during its first phase).
    let default_observed = observed_runners
        .iter()
        .next()
        .cloned()
        .unwrap_or_else(default_runner);
    for ((worker, test, retry), identities) in identities_by_attempt {
        let runner = default_observed.clone();
        let mut phases = Vec::new();
        let mut runtime = Vec::new();
        for (position, (identity, observations)) in identities.iter().enumerate() {
            let id = phase_id(run_id, identity);
            phases.push(CoveragePhase {
                id: id.clone(),
                kind: match identity.phase.as_str() {
                    "call" => "test",
                    value => value,
                }
                .into(),
                operation: format!("{runner} {}", identity.phase),
                source: Some(test.clone()),
                caused_by_phase_id: None,
                started_at_ms: position as i64 * 2 + 1,
                ended_at_ms: None,
                status: Some("failed".into()),
                error: Some("the phase started but the runner reported no outcome".into()),
            });
            runtime.push(snapshot(&index, observations, &id)?);
        }
        raw_results.push(RawTestResult {
            test_id: Some(test.clone()),
            scope: Some(scope(run_id, &worker, &test, retry)),
            test: test.clone(),
            // What the runner named, as the project names it. A dotted
            // module path or a pytest node id is not a path, so the old
            // derivation stays only as a fallback.
            test_file: test_files
                .get(&(worker.clone(), test.clone(), retry))
                .and_then(|path| assertions.relative(path))
                .or_else(|| test.split("::").next().map(str::to_owned)),
            title: test.rsplit("::").next().map(str::to_owned),
            retry: Some(retry),
            status: Some("failed".into()),
            expected_status: Some("passed".into()),
            flaky: false,
            provenance: TestProvenance {
                runner: runner.clone(),
                kind: "unit".into(),
                project: None,
                source: PYTHON_FRONTEND_VERSION.into(),
            },
            role: "test".into(),
            attribution: crate::coverage_report::ATTRIBUTION_EXACT.into(),
            phases,
            runtime,
            browser: Vec::new(),
            server: Vec::new(),
        });
    }
    for (worker, observations) in &background {
        if observations.hits.is_empty() && observations.vectors.is_empty() {
            continue;
        }
        let test = format!("__supercov_background__:{worker}");
        let identity = Identity {
            worker: worker.clone(),
            test: test.clone(),
            retry: 0,
            phase: "background".into(),
        };
        let phase = phase_id(run_id, &identity);
        raw_results.push(RawTestResult {
            test_id: Some(test.clone()),
            scope: Some(scope(run_id, worker, &test, 0)),
            test: "Python import, collection and background execution".into(),
            test_file: None,
            title: None,
            retry: Some(0),
            status: Some("unknown".into()),
            expected_status: None,
            flaky: false,
            provenance: TestProvenance {
                runner: default_observed.clone(),
                kind: "unit".into(),
                project: None,
                source: PYTHON_FRONTEND_VERSION.into(),
            },
            role: "background".into(),
            attribution: crate::coverage_report::ATTRIBUTION_EXACT.into(),
            phases: vec![CoveragePhase {
                id: phase.clone(),
                kind: "background".into(),
                operation: "Python import and collection background".into(),
                source: None,
                caused_by_phase_id: None,
                started_at_ms: 0,
                ended_at_ms: Some(0),
                status: Some("passed".into()),
                error: None,
            }],
            runtime: vec![snapshot(&index, observations, &phase)?],
            browser: Vec::new(),
            server: Vec::new(),
        });
    }

    // Runtime-detected limitations: obligations the runtime could not map
    // become unmeasured, and every limitation ID joins the manifest so the
    // declaration and manifest agree.
    let mut limitation_ids = manifest
        .limitations
        .iter()
        .filter_map(|item| item.get("id").and_then(serde_json::Value::as_str))
        .map(str::to_owned)
        .collect::<BTreeSet<_>>();
    let mut unmeasured = manifest.unmeasured.iter().cloned().collect::<BTreeSet<_>>();
    let mut new_limitations = Vec::new();
    for limitation in &limitations {
        if let Some(obligation) = &limitation.obligation {
            if !index.lines.contains_key(obligation.as_str()) {
                return Err(PythonEvidenceError::UnknownObligation(obligation.clone()));
            }
            unmeasured.insert(obligation.clone());
        } else if let Some(file) = &limitation.file {
            // A code-object mapping failure or missing debug ranges prevents
            // every obligation in that source file from being observed. Mark
            // the whole file unmeasured instead of presenting its denominator
            // as ordinary uncovered code.
            unmeasured.extend(
                index
                    .lines
                    .iter()
                    .filter(|(_, (obligation_file, _))| obligation_file == file)
                    .map(|(id, _)| (*id).to_owned()),
            );
        }
        if limitation_ids.insert(limitation.id.clone()) {
            let (file, line) = limitation
                .obligation
                .as_deref()
                .and_then(|id| index.lines.get(id).cloned())
                .unwrap_or_else(|| {
                    (
                        limitation.file.clone().unwrap_or_else(|| {
                            manifest
                                .points
                                .first()
                                .map_or(".".into(), |point| point.file.clone())
                        }),
                        1,
                    )
                });
            new_limitations.push(json!({
                "id": limitation.id,
                "kind": "semantic-safety",
                "file": file,
                "line": line,
                "column": 0,
                "source": "",
                "reason": limitation.reason
            }));
        }
    }
    manifest.limitations.extend(new_limitations);
    manifest.unmeasured = unmeasured.into_iter().collect();
    let structural_limitations = limitation_ids.into_iter().collect::<Vec<_>>();

    // Retries are separate raw results so their coverage remains attempt
    // exact, but the public lifecycle diagnostic reports logical tests rather
    // than inflating the count when a flaky test is rerun.
    let tests = raw_results
        .iter()
        .filter(|raw| raw.role == "test")
        .map(|raw| raw.test.as_str())
        .collect::<BTreeSet<_>>()
        .len();
    Ok(PythonFrontendRun {
        declaration: FrontendRunDeclaration {
            protocol_version: LANGUAGE_FRONTEND_PROTOCOL_VERSION,
            frontend_id: "python".into(),
            frontend_version: PYTHON_FRONTEND_VERSION.into(),
            language: "python".into(),
            structural_source: StructuralSource::OwnedProbes,
            runners: observed_runners
                .iter()
                .map(|runner| FrontendRunnerDeclaration {
                    runner: runner.clone(),
                    execution_model: if runner == UNITTEST_RUNNER {
                        ExecutionModel::SerialInProcess
                    } else {
                        ExecutionModel::ParallelContextPropagated
                    },
                    attribution: FrontendAttribution {
                        run: AttributionPrecision::Exact,
                        worker: AttributionPrecision::Exact,
                        test: AttributionPrecision::Exact,
                        retry: AttributionPrecision::Exact,
                        phase: AttributionPrecision::Exact,
                        action: AttributionPrecision::Unavailable,
                        assertion: AttributionPrecision::Exact,
                    },
                    limitations: vec![FrontendLimitation {
                        id: "python-action-linkage".into(),
                        scopes: vec![FrontendLimitationScope::Action],
                        reason: format!("{runner} exposes no general action lifecycle"),
                    }],
                })
                .collect(),
            structural_limitations,
        },
        request: CoverageReportRequest {
            run_id: run_id.into(),
            manifest,
            raw_results,
            generated_at: generated_at.into(),
            coverage_model: Some(python_coverage_model()),
            integrity: None,
            test_exit_code: ExitCodeInput::Present(Some(test_exit_code)),
        },
        tests,
        interpreters,
        python_versions: python_versions.into_iter().collect(),
    })
}

#[cfg(test)]
mod tests {
    use std::time::{SystemTime, UNIX_EPOCH};

    use super::*;
    use crate::{
        frontend_protocol::validate_frontend_report_request,
        python_instrumenter::build_python_obligations,
    };

    fn temporary(name: &str) -> std::path::PathBuf {
        let nonce = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "supercov-python-evidence-{}-{nonce}-{name}",
            std::process::id()
        ));
        fs::create_dir_all(&path).unwrap();
        path
    }

    fn write_transport(path: &Path, records: &[serde_json::Value], dropped: u64) {
        let payloads = records
            .iter()
            .map(|record| serde_json::to_vec(record).unwrap())
            .collect::<Vec<_>>();
        let capacity = payloads
            .iter()
            .fold(TRANSPORT_HEADER_SIZE, |cursor, payload| {
                align_transport(cursor + TRANSPORT_RECORD_HEADER_SIZE + payload.len()).unwrap()
            })
            + 64;
        let mut bytes = vec![0_u8; capacity];
        bytes[..8].copy_from_slice(TRANSPORT_MAGIC);
        bytes[8..12].copy_from_slice(&TRANSPORT_VERSION.to_le_bytes());
        bytes[12..16].copy_from_slice(&(TRANSPORT_HEADER_SIZE as u32).to_le_bytes());
        bytes[16..24].copy_from_slice(&(capacity as u64).to_le_bytes());
        bytes[24..32].copy_from_slice(&dropped.to_le_bytes());
        bytes[32..40].copy_from_slice(&1_u64.to_le_bytes());
        let mut cursor = TRANSPORT_HEADER_SIZE;
        for payload in payloads {
            let payload_start = cursor + TRANSPORT_RECORD_HEADER_SIZE;
            let payload_end = payload_start + payload.len();
            bytes[payload_start..payload_end].copy_from_slice(&payload);
            bytes[cursor + 4..cursor + 8].copy_from_slice(&(payload.len() as u32).to_le_bytes());
            bytes[cursor + 8..cursor + 12]
                .copy_from_slice(&transport_checksum(&payload).to_le_bytes());
            bytes[cursor] = 1;
            cursor = align_transport(payload_end).unwrap();
        }
        fs::write(path, bytes).unwrap();
    }

    // A path is only absolute in the platform's own spelling: "/project" is a
    // relative path on Windows, where an absolute one needs a drive. The
    // runtimes report whatever the interpreter loaded, so these fixtures have
    // to speak the host's dialect too.
    fn under(first: &str, rest: &str) -> String {
        let mut path = PathBuf::from(if cfg!(windows) {
            format!("C:\\{first}")
        } else {
            format!("/{first}")
        });
        for part in rest.split('/').filter(|part| !part.is_empty()) {
            path.push(part);
        }
        path.to_string_lossy().into_owned()
    }

    fn inventory_of(root: &str, sites: &[(&str, usize, usize)]) -> PythonAssertionInventory {
        use crate::assertion_map::{Anchor, Files, Inputs, InventorySite};
        PythonAssertionInventory::new(
            Path::new(root),
            &Inputs {
                schema_version: 1,
                language: "python".into(),
                context_digest: "context".into(),
                files: Files::new(),
                assertions: sites
                    .iter()
                    .map(|(file, line, column)| InventorySite {
                        at: Anchor {
                            file: (*file).into(),
                            line: *line,
                            column: *column,
                            text: "assert f(1) == 1".into(),
                        },
                        operation: "assert".into(),
                    })
                    .collect(),
                limitations: vec![],
            },
        )
    }

    fn assertion_sources(run: &PythonFrontendRun) -> Vec<String> {
        run.request.raw_results[0]
            .phases
            .iter()
            .filter(|phase| phase.kind == "assertion")
            .filter_map(|phase| phase.source.clone())
            .collect()
    }

    fn run_with_sites(
        name: &str,
        sites: &[serde_json::Value],
        outcome_file: Option<&str>,
        inventory: &PythonAssertionInventory,
    ) -> PythonFrontendRun {
        let source = "def f(a):\n    return a\n";
        let obligations = build_python_obligations("m.py", source).unwrap();
        let mut outcome = json!({"t":"outcome","worker":"main","test":"tests/test_m.py::test_a","retry":0,"phase":"call","outcome":"passed","xfail":false});
        if let Some(file) = outcome_file {
            outcome["file"] = json!(file);
        }
        let mut lines = vec![
            json!({"t":"process","v":1,"run":"run-1","pid":1,"worker":"main","python":"3.14.4","executable":"python","argv":["pytest"]}),
            json!({"t":"phase","ctx":1,"at":5,"worker":"main","test":"tests/test_m.py::test_a","retry":0,"phase":"call"}),
            json!({"t":"assert","ctx":1}),
        ];
        lines.extend(sites.iter().cloned());
        lines.push(outcome);
        lines.push(json!({"t":"exit","at":9}));
        let directory = temporary(name);
        write_transport(&directory.join("main.1.mmap"), &lines, 0);
        let run = build_python_frontend_run(
            &obligations.manifest,
            &directory,
            "run-1",
            "now",
            0,
            inventory,
        )
        .unwrap();
        validate_frontend_report_request(&run.declaration, &run.request).unwrap();
        fs::remove_dir_all(directory).unwrap();
        run
    }

    #[test]
    fn an_assertion_site_becomes_a_located_phase_when_the_inventory_names_one() {
        // Python reports a file and a line for an assertion, so a line is a
        // witness only when the inventory holds exactly one site on it. The
        // column reported is zero-based, which is what every native manifest
        // reports and what the assertion report adds one to.
        let inventory = inventory_of(&under("project", ""), &[("tests/test_m.py", 6, 5)]);
        let run = run_with_sites(
            "py-asite-located",
            &[json!({"t":"asite","ctx":1,"f":under("project", "tests/test_m.py"),"l":6})],
            None,
            &inventory,
        );
        assert!(
            assertion_sources(&run).contains(&"tests/test_m.py:6:4".to_string()),
            "expected a located assertion phase, got {:?}",
            assertion_sources(&run)
        );
    }

    #[test]
    fn an_ambiguous_or_foreign_assertion_site_witnesses_nothing() {
        // Two sites on one line cannot be told apart from a line number, and a
        // frame outside the project names nothing. Both lose the witness
        // rather than guessing one.
        let ambiguous = inventory_of(
            &under("project", ""),
            &[("tests/test_m.py", 6, 5), ("tests/test_m.py", 6, 30)],
        );
        let run = run_with_sites(
            "py-asite-ambiguous",
            &[json!({"t":"asite","ctx":1,"f":under("project", "tests/test_m.py"),"l":6})],
            None,
            &ambiguous,
        );
        assert_eq!(
            assertion_sources(&run),
            vec!["tests/test_m.py::test_a".to_string()],
            "only the per-test assertion phase should remain"
        );

        let known = inventory_of(&under("project", ""), &[("tests/test_m.py", 6, 5)]);
        let outside = run_with_sites(
            "py-asite-outside",
            &[json!({"t":"asite","ctx":1,"f":under("elsewhere", "tests/test_m.py"),"l":6})],
            None,
            &known,
        );
        assert_eq!(
            assertion_sources(&outside),
            vec!["tests/test_m.py::test_a".to_string()]
        );
    }

    #[test]
    fn the_runner_names_the_test_file_in_either_path_form() {
        // pytest reports the file from the report's location and unittest from
        // the test's module; either may be absolute or relative. Both name the
        // same project file, and a runner that names none falls back to the
        // node id.
        let inventory = inventory_of(&under("project", ""), &[("tests/test_m.py", 6, 5)]);
        for reported in [
            under("project", "tests/test_m.py"),
            "tests/test_m.py".to_owned(),
            "./tests/test_m.py".to_owned(),
        ] {
            let run = run_with_sites("py-asite-file", &[], Some(reported.as_str()), &inventory);
            assert_eq!(
                run.request.raw_results[0].test_file.as_deref(),
                Some("tests/test_m.py"),
                "{reported} should resolve to the project path"
            );
        }
        let without = run_with_sites("py-asite-nofile", &[], None, &inventory);
        assert_eq!(
            without.request.raw_results[0].test_file.as_deref(),
            Some("tests/test_m.py"),
            "a pytest node id already begins with the file"
        );
    }

    #[test]
    fn evidence_before_the_first_assertion_links_to_it_when_the_test_passes() {
        // The runtime's marker says everything the call phase recorded so far
        // ran before an assertion. That evidence carries an assertion phase
        // that passed with the test; what ran after the marker, and all of a
        // test that failed, stays execution only. A second marker is inert.
        let source = "def f(a):\n    return a\n\n\ndef g(b):\n    return b\n";
        let obligations = build_python_obligations("m.py", source).unwrap();
        let before = &obligations.plan.statements[0].id;
        let after = &obligations.plan.statements[1].id;
        let run_with = |name: &str, outcome: &str| {
            let lines = [
                json!({"t":"process","v":1,"run":"run-1","pid":1,"worker":"main","python":"3.14.4","executable":"python","argv":["pytest"]}),
                json!({"t":"phase","ctx":1,"at":5,"worker":"main","test":"tests/test_m.py::test_a","retry":0,"phase":"call"}),
                json!({"t":"hit","ctx":1,"id":before}),
                json!({"t":"assert","ctx":1}),
                json!({"t":"assert","ctx":1}),
                json!({"t":"hit","ctx":1,"id":after}),
                json!({"t":"outcome","worker":"main","test":"tests/test_m.py::test_a","retry":0,"phase":"call","outcome":outcome,"xfail":false}),
                json!({"t":"exit","at":9}),
            ];
            let directory = temporary(name);
            write_transport(&directory.join("main.1.mmap"), &lines, 0);
            let run = build_python_frontend_run(
                &obligations.manifest,
                &directory,
                "run-1",
                "now",
                0,
                &PythonAssertionInventory::empty(),
            )
            .unwrap();
            validate_frontend_report_request(&run.declaration, &run.request).unwrap();
            fs::remove_dir_all(directory).unwrap();
            run
        };
        let events_of = |result: &RawTestResult, phase: &str| -> BTreeSet<String> {
            result
                .runtime
                .iter()
                .flat_map(|snapshot| snapshot.events.iter())
                .filter(|event| event.phase_id.as_deref() == Some(phase))
                .map(|event| event.id.clone())
                .collect()
        };

        let run = run_with("asserted-passed", "passed");
        let passed = &run.request.raw_results[0];
        assert_eq!(passed.test, "tests/test_m.py::test_a");
        let assertion = passed
            .phases
            .iter()
            .find(|phase| phase.kind == "assertion")
            .expect("the asserting test carries an assertion phase");
        assert_eq!(assertion.status.as_deref(), Some("passed"));
        assert_eq!(
            events_of(passed, &assertion.id),
            BTreeSet::from([before.clone()]),
            "only what ran before the marker is the assertion's evidence"
        );
        let test_phase = passed
            .phases
            .iter()
            .find(|phase| phase.kind == "test")
            .unwrap();
        assert_eq!(
            events_of(passed, &test_phase.id),
            BTreeSet::from([before.clone(), after.clone()]),
            "the test phase keeps everything it ran"
        );
        assert_eq!(
            run.declaration.runners[0].attribution.assertion,
            AttributionPrecision::Exact
        );

        let run = run_with("asserted-failed", "failed");
        let failed = &run.request.raw_results[0];
        let assertion = failed
            .phases
            .iter()
            .find(|phase| phase.kind == "assertion")
            .unwrap();
        assert_eq!(
            assertion.status.as_deref(),
            Some("failed"),
            "a failed test's assertion witnessed nothing"
        );
    }

    #[test]
    fn joins_phases_outcomes_hits_and_vectors_into_exact_results() {
        let source = "def f(a, b):\n    if a and b:\n        return 1\n    return 0\n";
        let obligations = build_python_obligations("m.py", source).unwrap();
        let decision = &obligations.plan.decisions[0];
        let statement = &obligations.plan.statements[0];
        let directory = temporary("join");
        let lines = [
            json!({"t":"process","v":1,"run":"run-1","pid":1,"worker":"main","python":"3.14.4","executable":"python","argv":["pytest"]}),
            json!({"t":"hit","ctx":0,"id":statement.id}),
            json!({"t":"phase","ctx":1,"at":5,"worker":"main","test":"tests/test_m.py::test_a","retry":0,"phase":"call"}),
            json!({"t":"dec","ctx":1,"id":decision.id,"v":"22","o":1}),
            json!({"t":"hit","ctx":1,"id":decision.outcome_true}),
            json!({"t":"outcome","worker":"main","test":"tests/test_m.py::test_a","retry":0,"phase":"setup","outcome":"passed","xfail":false}),
            json!({"t":"outcome","worker":"main","test":"tests/test_m.py::test_a","retry":0,"phase":"call","outcome":"passed","xfail":false}),
            json!({"t":"outcome","worker":"main","test":"tests/test_m.py::test_a","retry":0,"phase":"teardown","outcome":"passed","xfail":false}),
            json!({"t":"limitation","id":"python-decision-partially-mapped","reason":"folded","obligation":decision.id}),
            json!({"t":"exit","at":9}),
        ];
        write_transport(&directory.join("main.1.mmap"), &lines, 0);
        let run = build_python_frontend_run(
            &obligations.manifest,
            &directory,
            "run-1",
            "now",
            0,
            &PythonAssertionInventory::empty(),
        )
        .unwrap();
        validate_frontend_report_request(&run.declaration, &run.request).unwrap();
        assert_eq!(run.tests, 1);
        assert_eq!(run.request.raw_results.len(), 2);
        let test = &run.request.raw_results[0];
        assert_eq!(test.status.as_deref(), Some("passed"));
        assert_eq!(test.phases.len(), 3);
        assert_eq!(test.runtime.len(), 1);
        assert_eq!(test.runtime[0].decisions.len(), 1);
        assert!(
            test.runtime[0]
                .events
                .iter()
                .all(|event| event.phase_id.is_some())
        );
        let background = &run.request.raw_results[1];
        assert_eq!(background.role, "background");
        assert!(run.request.manifest.unmeasured.contains(&decision.id));
        assert!(
            run.declaration
                .structural_limitations
                .contains(&"python-decision-partially-mapped".to_owned())
        );
        fs::remove_dir_all(directory).unwrap();
    }

    #[test]
    fn fails_closed_without_an_interpreter_or_tests() {
        let obligations = build_python_obligations("m.py", "x = 1\n").unwrap();
        let directory = temporary("empty");
        assert!(matches!(
            build_python_frontend_run(
                &obligations.manifest,
                &directory,
                "run-1",
                "now",
                0,
                &PythonAssertionInventory::empty()
            ),
            Err(PythonEvidenceError::NoInterpreter)
        ));
        let path = directory.join("main.1.mmap");
        write_transport(
            &path,
            &[
                json!({"t":"process","v":1,"run":"run-1","pid":1,"worker":"main","python":"3.14.4","executable":"p","argv":[]}),
            ],
            0,
        );
        assert!(matches!(
            build_python_frontend_run(
                &obligations.manifest,
                &directory,
                "run-1",
                "now",
                0,
                &PythonAssertionInventory::empty()
            ),
            Err(PythonEvidenceError::NoTests)
        ));
        // A killed writer may have copied part of its next payload without
        // publishing the commit byte. The reader must stop at that frame,
        // even when the tail would be invalid JSON if treated as committed.
        let mut torn = fs::read(&path).unwrap();
        let process_length = transport_u32(&torn, TRANSPORT_HEADER_SIZE + 4).unwrap() as usize;
        let torn_cursor =
            align_transport(TRANSPORT_HEADER_SIZE + TRANSPORT_RECORD_HEADER_SIZE + process_length)
                .unwrap();
        torn[torn_cursor + 4..torn_cursor + 8].copy_from_slice(&5_u32.to_le_bytes());
        torn[torn_cursor + TRANSPORT_RECORD_HEADER_SIZE
            ..torn_cursor + TRANSPORT_RECORD_HEADER_SIZE + 5]
            .copy_from_slice(b"{nope");
        fs::write(&path, torn).unwrap();
        assert!(matches!(
            build_python_frontend_run(
                &obligations.manifest,
                &directory,
                "run-1",
                "now",
                0,
                &PythonAssertionInventory::empty()
            ),
            Err(PythonEvidenceError::NoTests)
        ));
        write_transport(
            &path,
            &[
                json!({"t":"process","v":1,"run":"run-1","pid":1,"worker":"main","python":"3.11.9","executable":"p","argv":[]}),
            ],
            0,
        );
        assert!(matches!(
            build_python_frontend_run(
                &obligations.manifest,
                &directory,
                "run-1",
                "now",
                0,
                &PythonAssertionInventory::empty()
            ),
            Err(PythonEvidenceError::UnsupportedPython(_))
        ));
        write_transport(
            &path,
            &[
                json!({"t":"process","v":1,"run":"run-1","pid":1,"worker":"main","python":"3.14.4","executable":"p","argv":[]}),
            ],
            2,
        );
        assert!(matches!(
            build_python_frontend_run(
                &obligations.manifest,
                &directory,
                "run-1",
                "now",
                0,
                &PythonAssertionInventory::empty()
            ),
            Err(PythonEvidenceError::DroppedRecords { count: 2, .. })
        ));
        fs::remove_dir_all(directory).unwrap();
    }
}