benchmatrix

benchmatrix

Public pytest-benchmark matrix and JSON results API.

BenchmarkLifecycleHook module-attribute

BenchmarkLifecycleHook: TypeAlias = Callable[
    [BenchmarkHookContext], None
]

Synchronous setup or cleanup hook for one benchmark invocation.

BenchmarkResultValidator module-attribute

BenchmarkResultValidator: TypeAlias = Callable[
    [BenchmarkHookContext, object], None
]

Synchronous correctness hook for one benchmark invocation result.

TargetFunction module-attribute

TargetFunction: TypeAlias = Callable[..., object]

Synchronous callable measured by pytest-benchmark through benchmatrix.

Target functions must perform the work being measured before returning. Async functions are rejected. Lazy return values are not forced by the harness.

BenchmarkPairedRunGroup dataclass

Manifest-backed paired AB/BA benchmark collection.

Complete pairs are atomic adjacent block attempts: an orphan success from a block whose other command failed is retained in records and runs but excluded from complete_pairs and statistical inference.

Source code in src/benchmatrix/bench_collection.py

@dataclass(frozen=True, slots=True)
class BenchmarkPairedRunGroup:
    """Manifest-backed paired AB/BA benchmark collection.

    Complete pairs are atomic adjacent block attempts: an orphan success from
    a block whose other command failed is retained in ``records`` and ``runs``
    but excluded from ``complete_pairs`` and statistical inference.
    """

    runs: tuple[BenchmarkRun, ...]
    records: tuple[BenchmarkPairedRunRecord, ...]
    baseline_command: tuple[str, ...]
    candidate_command: tuple[str, ...]
    created_at: str
    baseline_cwd: Path
    candidate_cwd: Path
    baseline_commit: str | None
    candidate_commit: str | None
    baseline_environment_fingerprint: str | None
    candidate_environment_fingerprint: str | None
    expected_cells: tuple[BenchmarkCell, ...]
    requested_pairs: int
    random_seed: int
    manifest_path: Path
    automatic_pairs: bool = False

    def __post_init__(self) -> None:
        """Normalize containers and validate paired collection invariants."""
        runs = tuple(self.runs)
        records = tuple(self.records)
        baseline_command = tuple(self.baseline_command)
        candidate_command = tuple(self.candidate_command)
        expected_cells = tuple(self.expected_cells)
        for name, command in (("baseline_command", baseline_command), ("candidate_command", candidate_command)):
            if not command or any(not isinstance(argument, str) or not argument for argument in command):
                raise ValueError(f"BenchmarkPairedRunGroup.{name} must contain non-empty strings.")
        _validate_timestamp(self.created_at, field_name="BenchmarkPairedRunGroup.created_at")
        if isinstance(self.requested_pairs, bool) or not isinstance(self.requested_pairs, int):
            raise TypeError("BenchmarkPairedRunGroup.requested_pairs must be an integer.")
        if self.requested_pairs <= 0:
            raise ValueError("BenchmarkPairedRunGroup.requested_pairs must be positive.")
        if isinstance(self.random_seed, bool) or not isinstance(self.random_seed, int):
            raise TypeError("BenchmarkPairedRunGroup.random_seed must be an integer.")
        if self.random_seed < 0:
            raise ValueError("BenchmarkPairedRunGroup.random_seed must be non-negative.")
        if not isinstance(self.automatic_pairs, bool):
            raise TypeError("BenchmarkPairedRunGroup.automatic_pairs must be a boolean.")
        if tuple(record.index for record in records) != tuple(range(1, len(records) + 1)):
            raise ValueError("BenchmarkPairedRunGroup record indexes must be contiguous and one-based.")
        if len(runs) != sum(record.status == "succeeded" for record in records):
            raise ValueError("BenchmarkPairedRunGroup runs must align with successful records.")
        if len(set(expected_cells)) != len(expected_cells):
            raise ValueError("BenchmarkPairedRunGroup.expected_cells must not contain duplicates.")
        for cell in expected_cells:
            _validate_cell(cell)
        if runs and not expected_cells:
            raise ValueError("BenchmarkPairedRunGroup with successful runs requires expected cells.")
        _validate_optional_anchor(self.baseline_commit, field="baseline_commit")
        _validate_optional_anchor(self.candidate_commit, field="candidate_commit")
        _validate_optional_fingerprint(
            self.baseline_environment_fingerprint,
            field="BenchmarkPairedRunGroup.baseline_environment_fingerprint",
        )
        _validate_optional_fingerprint(
            self.candidate_environment_fingerprint,
            field="BenchmarkPairedRunGroup.candidate_environment_fingerprint",
        )
        if not runs and (
            expected_cells
            or self.baseline_commit is not None
            or self.candidate_commit is not None
            or self.baseline_environment_fingerprint is not None
            or self.candidate_environment_fingerprint is not None
        ):
            raise ValueError("BenchmarkPairedRunGroup without successful runs cannot define anchors or cells.")

        if self.automatic_pairs:
            automatic_target = (
                _automatic_pair_target(len(expected_cells)) if expected_cells else _PROVISIONAL_AUTOMATIC_PAIRS
            )
            if self.requested_pairs != automatic_target:
                raise ValueError(
                    "BenchmarkPairedRunGroup automatic requested_pairs does not match the learned matrix supercycle."
                )

        schedule = make_paired_ab_ba_schedule(
            self.requested_pairs,
            random_seed=self.random_seed,
            cell_count=len(expected_cells) or None,
        )
        schedule_by_pair = {entry.pair_index: entry for entry in schedule}
        blocks: dict[tuple[int, int], set[PairedVariant]] = {}
        latest_attempt_by_pair: dict[int, int] = {}
        successful_records = tuple(record for record in records if record.status == "succeeded")
        first_seen_pairs: list[int] = []
        seen_pairs: set[int] = set()
        matrix_anchor_seen = False
        for record in records:
            if record.pair_index > self.requested_pairs:
                raise ValueError("BenchmarkPairedRunGroup record pair_index exceeds requested_pairs.")
            if record.pair_index not in seen_pairs:
                if record.pair_index != len(seen_pairs) + 1:
                    raise ValueError("BenchmarkPairedRunGroup first-seen pair indexes must form a one-based prefix.")
                first_seen_pairs.append(record.pair_index)
                seen_pairs.add(record.pair_index)
            if record.pair_index > 1 and not matrix_anchor_seen:
                raise ValueError("BenchmarkPairedRunGroup cannot attempt a later pair before a matrix anchor.")
            scheduled = schedule_by_pair[record.pair_index]
            if record.pair_order != scheduled.pair_order or record.cell_order_index != scheduled.cell_order_index:
                raise ValueError("BenchmarkPairedRunGroup record differs from the deterministic schedule.")
            previous_attempt = latest_attempt_by_pair.get(record.pair_index, record.block_attempt)
            if record.block_attempt < previous_attempt:
                raise ValueError("BenchmarkPairedRunGroup block attempts must be chronological per pair.")
            latest_attempt_by_pair[record.pair_index] = record.block_attempt
            variants = blocks.setdefault((record.pair_index, record.block_attempt), set())
            if record.variant in variants:
                raise ValueError("BenchmarkPairedRunGroup block attempts cannot repeat a variant.")
            variants.add(record.variant)
            matrix_anchor_seen = matrix_anchor_seen or record.status == "succeeded"
        if first_seen_pairs != list(range(1, len(first_seen_pairs) + 1)):
            raise ValueError("BenchmarkPairedRunGroup first-seen pair indexes must form a one-based prefix.")
        if not expected_cells and any(record.pair_index != 1 for record in records):
            raise ValueError("BenchmarkPairedRunGroup cannot contain later pairs before learning the matrix.")
        records_by_block: dict[tuple[int, int], list[BenchmarkPairedRunRecord]] = {}
        for record in records:
            records_by_block.setdefault((record.pair_index, record.block_attempt), []).append(record)
        for (pair_index, _block_attempt), block_records in records_by_block.items():
            scheduled_variants = schedule_by_pair[pair_index].variants
            observed_variants = tuple(record.variant for record in block_records)
            if observed_variants != scheduled_variants[: len(observed_variants)]:
                raise ValueError("BenchmarkPairedRunGroup block records are not in scheduled AB/BA order.")
            if len(block_records) == 2 and block_records[1].index != block_records[0].index + 1:
                raise ValueError("BenchmarkPairedRunGroup block members must be adjacent records.")
            if len(block_records) == 1 and block_records[0].index < len(records):
                following = records[block_records[0].index]
                if following.pair_index != pair_index or following.block_attempt != block_records[0].block_attempt + 1:
                    raise ValueError(
                        "BenchmarkPairedRunGroup a partial block must be followed by a retry of the same pair."
                    )
        for pair_index in range(1, self.requested_pairs + 1):
            attempts = sorted(attempt for pair, attempt in blocks if pair == pair_index)
            if attempts and attempts != list(range(1, attempts[-1] + 1)):
                raise ValueError("BenchmarkPairedRunGroup block attempts must be contiguous per pair.")

        successful_by_block: dict[tuple[int, int], set[PairedVariant]] = {}
        for record in successful_records:
            successful_by_block.setdefault((record.pair_index, record.block_attempt), set()).add(record.variant)
        complete_by_pair: dict[int, int] = {}
        for (pair_index, block_attempt), variants in successful_by_block.items():
            if variants == {"baseline", "candidate"}:
                if pair_index in complete_by_pair:
                    raise ValueError("BenchmarkPairedRunGroup cannot contain two complete blocks for one pair.")
                complete_by_pair[pair_index] = block_attempt
        for pair_index, complete_attempt in complete_by_pair.items():
            if any(pair == pair_index and attempt > complete_attempt for pair, attempt in blocks):
                raise ValueError("BenchmarkPairedRunGroup cannot retry a pair after a complete block.")

        scheduled_orders = {
            pair_index: balanced_cell_order(
                expected_cells,
                order_index=schedule_by_pair[pair_index].cell_order_index,
                random_seed=self.random_seed,
            )
            for pair_index in range(1, self.requested_pairs + 1)
        }
        for run, record in zip(runs, successful_records, strict=True):
            if not isinstance(run, BenchmarkRun):
                raise TypeError("BenchmarkPairedRunGroup.runs must contain BenchmarkRun values.")
            if _run_cells(run) != expected_cells:
                raise ValueError("BenchmarkPairedRunGroup run matrix does not match expected_cells.")
            if _run_cell_order(run) != scheduled_orders[record.pair_index]:
                raise ValueError("BenchmarkPairedRunGroup run does not match its scheduled cell order.")

        _validate_paired_variant_anchors(
            successful_records,
            baseline_commit=self.baseline_commit,
            candidate_commit=self.candidate_commit,
            baseline_fingerprint=self.baseline_environment_fingerprint,
            candidate_fingerprint=self.candidate_environment_fingerprint,
        )

        object.__setattr__(self, "runs", runs)
        object.__setattr__(self, "records", records)
        object.__setattr__(self, "baseline_command", baseline_command)
        object.__setattr__(self, "candidate_command", candidate_command)
        object.__setattr__(self, "baseline_cwd", Path(self.baseline_cwd))
        object.__setattr__(self, "candidate_cwd", Path(self.candidate_cwd))
        object.__setattr__(self, "expected_cells", expected_cells)
        object.__setattr__(self, "manifest_path", Path(self.manifest_path))

    @property
    def successful_count(self) -> int:
        """Return successful commands, including orphan successes."""
        return len(self.runs)

    @property
    def attempted_count(self) -> int:
        """Return the number of completed command attempts."""
        return len(self.records)

    @property
    def failed_count(self) -> int:
        """Return the number of failed command attempts."""
        return sum(record.status == "failed" for record in self.records)

    @property
    def complete_pairs(self) -> tuple[BenchmarkRunPair, ...]:
        """Return complete atomic blocks in deterministic target-pair order."""
        run_by_record = {
            (record.pair_index, record.block_attempt, record.variant): (record, run)
            for record, run in zip(
                (record for record in self.records if record.status == "succeeded"),
                self.runs,
                strict=True,
            )
        }
        pairs: list[BenchmarkRunPair] = []
        for pair_index in range(1, self.requested_pairs + 1):
            attempts = sorted({block_attempt for pair, block_attempt, _variant in run_by_record if pair == pair_index})
            for block_attempt in attempts:
                baseline_item = run_by_record.get((pair_index, block_attempt, "baseline"))
                candidate_item = run_by_record.get((pair_index, block_attempt, "candidate"))
                if baseline_item is None or candidate_item is None:
                    continue
                baseline_record, baseline = baseline_item
                candidate_record, candidate = candidate_item
                pairs.append(
                    BenchmarkRunPair(
                        pair_index=pair_index,
                        block_attempt=block_attempt,
                        pair_order=baseline_record.pair_order,
                        cell_order=balanced_cell_order(
                            self.expected_cells,
                            order_index=baseline_record.cell_order_index,
                            random_seed=self.random_seed,
                        ),
                        baseline=baseline,
                        candidate=candidate,
                        baseline_record=baseline_record,
                        candidate_record=candidate_record,
                    )
                )
                break
        return tuple(pairs)

    @property
    def baseline_runs(self) -> tuple[BenchmarkRun, ...]:
        """Return baseline members of complete pairs in pair order."""
        return tuple(pair.baseline for pair in self.complete_pairs)

    @property
    def candidate_runs(self) -> tuple[BenchmarkRun, ...]:
        """Return candidate members of complete pairs in pair order."""
        return tuple(pair.candidate for pair in self.complete_pairs)

    @property
    def complete_pair_count(self) -> int:
        """Return the number of complete atomic collection blocks."""
        return len(self.complete_pairs)

    @property
    def orphan_success_count(self) -> int:
        """Return successes excluded because their block is incomplete."""
        return self.successful_count - 2 * self.complete_pair_count

    @property
    def is_complete(self) -> bool:
        """Return whether every requested target pair has a complete block."""
        return self.complete_pair_count == self.requested_pairs

    @property
    def order_supercycle_length(self) -> int | None:
        """Return the joint AB/BA-by-row cycle, once the matrix is known."""
        if not self.expected_cells:
            return None
        return balanced_order_supercycle_length(len(self.expected_cells))

    @property
    def is_jointly_balanced(self) -> bool:
        """Return whether the fixed target contains whole joint supercycles."""
        supercycle = self.order_supercycle_length
        return supercycle is not None and self.requested_pairs % supercycle == 0

    @property
    def remaining_pair_count(self) -> int:
        """Return the number of target pairs still lacking a complete block."""
        return self.requested_pairs - self.complete_pair_count

    @property
    def incomplete_pair_indexes(self) -> tuple[int, ...]:
        """Return target-pair indexes without a complete atomic block."""
        complete = {pair.pair_index for pair in self.complete_pairs}
        return tuple(index for index in range(1, self.requested_pairs + 1) if index not in complete)

    def compare(
        self,
        *,
        compatibility_policy: RunCompatibilityPolicy | None = None,
        regression_policy: RegressionPolicy | None = None,
        evidence_policy: EvidencePolicy | None = None,
        inference_policy: InferencePolicy | None = None,
        precision_policy: PrecisionPolicy | None = None,
    ) -> BenchmarkRunComparison:
        """Compare members after every requested atomic block is complete.

        Raises:
            BenchmarkCollectionError: If the fixed paired design is incomplete.
        """
        if not self.is_complete:
            raise BenchmarkCollectionError(
                "Paired collection is incomplete; finish or retry every requested block before inference."
            )
        supercycle = self.order_supercycle_length
        if supercycle is None or self.requested_pairs % supercycle != 0:
            raise BenchmarkCollectionError(
                "Paired collection target is not a whole AB/BA-by-balanced-row supercycle; "
                "collect a jointly balanced fixed design before inference."
            )
        return compare_paired_benchmark_run_groups(
            self.baseline_runs,
            self.candidate_runs,
            pair_strata=tuple(pair.pair_order for pair in self.complete_pairs),
            compatibility_policy=compatibility_policy,
            regression_policy=regression_policy,
            evidence_policy=evidence_policy,
            inference_policy=inference_policy,
            precision_policy=precision_policy,
            precision_pair_count_multiple=supercycle,
        )

successful_count property

successful_count: int

Return successful commands, including orphan successes.

attempted_count property

attempted_count: int

Return the number of completed command attempts.

failed_count property

failed_count: int

Return the number of failed command attempts.

complete_pairs property

complete_pairs: tuple[BenchmarkRunPair, ...]

Return complete atomic blocks in deterministic target-pair order.

baseline_runs property

baseline_runs: tuple[BenchmarkRun, ...]

Return baseline members of complete pairs in pair order.

candidate_runs property

candidate_runs: tuple[BenchmarkRun, ...]

Return candidate members of complete pairs in pair order.

complete_pair_count property

complete_pair_count: int

Return the number of complete atomic collection blocks.

orphan_success_count property

orphan_success_count: int

Return successes excluded because their block is incomplete.

is_complete property

is_complete: bool

Return whether every requested target pair has a complete block.

order_supercycle_length property

order_supercycle_length: int | None

Return the joint AB/BA-by-row cycle, once the matrix is known.

is_jointly_balanced property

is_jointly_balanced: bool

Return whether the fixed target contains whole joint supercycles.

remaining_pair_count property

remaining_pair_count: int

Return the number of target pairs still lacking a complete block.

incomplete_pair_indexes property

incomplete_pair_indexes: tuple[int, ...]

Return target-pair indexes without a complete atomic block.

__post_init__

__post_init__() -> None

Normalize containers and validate paired collection invariants.

Source code in src/benchmatrix/bench_collection.py

def __post_init__(self) -> None:
    """Normalize containers and validate paired collection invariants."""
    runs = tuple(self.runs)
    records = tuple(self.records)
    baseline_command = tuple(self.baseline_command)
    candidate_command = tuple(self.candidate_command)
    expected_cells = tuple(self.expected_cells)
    for name, command in (("baseline_command", baseline_command), ("candidate_command", candidate_command)):
        if not command or any(not isinstance(argument, str) or not argument for argument in command):
            raise ValueError(f"BenchmarkPairedRunGroup.{name} must contain non-empty strings.")
    _validate_timestamp(self.created_at, field_name="BenchmarkPairedRunGroup.created_at")
    if isinstance(self.requested_pairs, bool) or not isinstance(self.requested_pairs, int):
        raise TypeError("BenchmarkPairedRunGroup.requested_pairs must be an integer.")
    if self.requested_pairs <= 0:
        raise ValueError("BenchmarkPairedRunGroup.requested_pairs must be positive.")
    if isinstance(self.random_seed, bool) or not isinstance(self.random_seed, int):
        raise TypeError("BenchmarkPairedRunGroup.random_seed must be an integer.")
    if self.random_seed < 0:
        raise ValueError("BenchmarkPairedRunGroup.random_seed must be non-negative.")
    if not isinstance(self.automatic_pairs, bool):
        raise TypeError("BenchmarkPairedRunGroup.automatic_pairs must be a boolean.")
    if tuple(record.index for record in records) != tuple(range(1, len(records) + 1)):
        raise ValueError("BenchmarkPairedRunGroup record indexes must be contiguous and one-based.")
    if len(runs) != sum(record.status == "succeeded" for record in records):
        raise ValueError("BenchmarkPairedRunGroup runs must align with successful records.")
    if len(set(expected_cells)) != len(expected_cells):
        raise ValueError("BenchmarkPairedRunGroup.expected_cells must not contain duplicates.")
    for cell in expected_cells:
        _validate_cell(cell)
    if runs and not expected_cells:
        raise ValueError("BenchmarkPairedRunGroup with successful runs requires expected cells.")
    _validate_optional_anchor(self.baseline_commit, field="baseline_commit")
    _validate_optional_anchor(self.candidate_commit, field="candidate_commit")
    _validate_optional_fingerprint(
        self.baseline_environment_fingerprint,
        field="BenchmarkPairedRunGroup.baseline_environment_fingerprint",
    )
    _validate_optional_fingerprint(
        self.candidate_environment_fingerprint,
        field="BenchmarkPairedRunGroup.candidate_environment_fingerprint",
    )
    if not runs and (
        expected_cells
        or self.baseline_commit is not None
        or self.candidate_commit is not None
        or self.baseline_environment_fingerprint is not None
        or self.candidate_environment_fingerprint is not None
    ):
        raise ValueError("BenchmarkPairedRunGroup without successful runs cannot define anchors or cells.")

    if self.automatic_pairs:
        automatic_target = (
            _automatic_pair_target(len(expected_cells)) if expected_cells else _PROVISIONAL_AUTOMATIC_PAIRS
        )
        if self.requested_pairs != automatic_target:
            raise ValueError(
                "BenchmarkPairedRunGroup automatic requested_pairs does not match the learned matrix supercycle."
            )

    schedule = make_paired_ab_ba_schedule(
        self.requested_pairs,
        random_seed=self.random_seed,
        cell_count=len(expected_cells) or None,
    )
    schedule_by_pair = {entry.pair_index: entry for entry in schedule}
    blocks: dict[tuple[int, int], set[PairedVariant]] = {}
    latest_attempt_by_pair: dict[int, int] = {}
    successful_records = tuple(record for record in records if record.status == "succeeded")
    first_seen_pairs: list[int] = []
    seen_pairs: set[int] = set()
    matrix_anchor_seen = False
    for record in records:
        if record.pair_index > self.requested_pairs:
            raise ValueError("BenchmarkPairedRunGroup record pair_index exceeds requested_pairs.")
        if record.pair_index not in seen_pairs:
            if record.pair_index != len(seen_pairs) + 1:
                raise ValueError("BenchmarkPairedRunGroup first-seen pair indexes must form a one-based prefix.")
            first_seen_pairs.append(record.pair_index)
            seen_pairs.add(record.pair_index)
        if record.pair_index > 1 and not matrix_anchor_seen:
            raise ValueError("BenchmarkPairedRunGroup cannot attempt a later pair before a matrix anchor.")
        scheduled = schedule_by_pair[record.pair_index]
        if record.pair_order != scheduled.pair_order or record.cell_order_index != scheduled.cell_order_index:
            raise ValueError("BenchmarkPairedRunGroup record differs from the deterministic schedule.")
        previous_attempt = latest_attempt_by_pair.get(record.pair_index, record.block_attempt)
        if record.block_attempt < previous_attempt:
            raise ValueError("BenchmarkPairedRunGroup block attempts must be chronological per pair.")
        latest_attempt_by_pair[record.pair_index] = record.block_attempt
        variants = blocks.setdefault((record.pair_index, record.block_attempt), set())
        if record.variant in variants:
            raise ValueError("BenchmarkPairedRunGroup block attempts cannot repeat a variant.")
        variants.add(record.variant)
        matrix_anchor_seen = matrix_anchor_seen or record.status == "succeeded"
    if first_seen_pairs != list(range(1, len(first_seen_pairs) + 1)):
        raise ValueError("BenchmarkPairedRunGroup first-seen pair indexes must form a one-based prefix.")
    if not expected_cells and any(record.pair_index != 1 for record in records):
        raise ValueError("BenchmarkPairedRunGroup cannot contain later pairs before learning the matrix.")
    records_by_block: dict[tuple[int, int], list[BenchmarkPairedRunRecord]] = {}
    for record in records:
        records_by_block.setdefault((record.pair_index, record.block_attempt), []).append(record)
    for (pair_index, _block_attempt), block_records in records_by_block.items():
        scheduled_variants = schedule_by_pair[pair_index].variants
        observed_variants = tuple(record.variant for record in block_records)
        if observed_variants != scheduled_variants[: len(observed_variants)]:
            raise ValueError("BenchmarkPairedRunGroup block records are not in scheduled AB/BA order.")
        if len(block_records) == 2 and block_records[1].index != block_records[0].index + 1:
            raise ValueError("BenchmarkPairedRunGroup block members must be adjacent records.")
        if len(block_records) == 1 and block_records[0].index < len(records):
            following = records[block_records[0].index]
            if following.pair_index != pair_index or following.block_attempt != block_records[0].block_attempt + 1:
                raise ValueError(
                    "BenchmarkPairedRunGroup a partial block must be followed by a retry of the same pair."
                )
    for pair_index in range(1, self.requested_pairs + 1):
        attempts = sorted(attempt for pair, attempt in blocks if pair == pair_index)
        if attempts and attempts != list(range(1, attempts[-1] + 1)):
            raise ValueError("BenchmarkPairedRunGroup block attempts must be contiguous per pair.")

    successful_by_block: dict[tuple[int, int], set[PairedVariant]] = {}
    for record in successful_records:
        successful_by_block.setdefault((record.pair_index, record.block_attempt), set()).add(record.variant)
    complete_by_pair: dict[int, int] = {}
    for (pair_index, block_attempt), variants in successful_by_block.items():
        if variants == {"baseline", "candidate"}:
            if pair_index in complete_by_pair:
                raise ValueError("BenchmarkPairedRunGroup cannot contain two complete blocks for one pair.")
            complete_by_pair[pair_index] = block_attempt
    for pair_index, complete_attempt in complete_by_pair.items():
        if any(pair == pair_index and attempt > complete_attempt for pair, attempt in blocks):
            raise ValueError("BenchmarkPairedRunGroup cannot retry a pair after a complete block.")

    scheduled_orders = {
        pair_index: balanced_cell_order(
            expected_cells,
            order_index=schedule_by_pair[pair_index].cell_order_index,
            random_seed=self.random_seed,
        )
        for pair_index in range(1, self.requested_pairs + 1)
    }
    for run, record in zip(runs, successful_records, strict=True):
        if not isinstance(run, BenchmarkRun):
            raise TypeError("BenchmarkPairedRunGroup.runs must contain BenchmarkRun values.")
        if _run_cells(run) != expected_cells:
            raise ValueError("BenchmarkPairedRunGroup run matrix does not match expected_cells.")
        if _run_cell_order(run) != scheduled_orders[record.pair_index]:
            raise ValueError("BenchmarkPairedRunGroup run does not match its scheduled cell order.")

    _validate_paired_variant_anchors(
        successful_records,
        baseline_commit=self.baseline_commit,
        candidate_commit=self.candidate_commit,
        baseline_fingerprint=self.baseline_environment_fingerprint,
        candidate_fingerprint=self.candidate_environment_fingerprint,
    )

    object.__setattr__(self, "runs", runs)
    object.__setattr__(self, "records", records)
    object.__setattr__(self, "baseline_command", baseline_command)
    object.__setattr__(self, "candidate_command", candidate_command)
    object.__setattr__(self, "baseline_cwd", Path(self.baseline_cwd))
    object.__setattr__(self, "candidate_cwd", Path(self.candidate_cwd))
    object.__setattr__(self, "expected_cells", expected_cells)
    object.__setattr__(self, "manifest_path", Path(self.manifest_path))

compare

compare(
    *,
    compatibility_policy: RunCompatibilityPolicy
    | None = None,
    regression_policy: RegressionPolicy | None = None,
    evidence_policy: EvidencePolicy | None = None,
    inference_policy: InferencePolicy | None = None,
    precision_policy: PrecisionPolicy | None = None,
) -> BenchmarkRunComparison

Compare members after every requested atomic block is complete.

Raises:

Type	Description
BenchmarkCollectionError	If the fixed paired design is incomplete.

Source code in src/benchmatrix/bench_collection.py

def compare(
    self,
    *,
    compatibility_policy: RunCompatibilityPolicy | None = None,
    regression_policy: RegressionPolicy | None = None,
    evidence_policy: EvidencePolicy | None = None,
    inference_policy: InferencePolicy | None = None,
    precision_policy: PrecisionPolicy | None = None,
) -> BenchmarkRunComparison:
    """Compare members after every requested atomic block is complete.

    Raises:
        BenchmarkCollectionError: If the fixed paired design is incomplete.
    """
    if not self.is_complete:
        raise BenchmarkCollectionError(
            "Paired collection is incomplete; finish or retry every requested block before inference."
        )
    supercycle = self.order_supercycle_length
    if supercycle is None or self.requested_pairs % supercycle != 0:
        raise BenchmarkCollectionError(
            "Paired collection target is not a whole AB/BA-by-balanced-row supercycle; "
            "collect a jointly balanced fixed design before inference."
        )
    return compare_paired_benchmark_run_groups(
        self.baseline_runs,
        self.candidate_runs,
        pair_strata=tuple(pair.pair_order for pair in self.complete_pairs),
        compatibility_policy=compatibility_policy,
        regression_policy=regression_policy,
        evidence_policy=evidence_policy,
        inference_policy=inference_policy,
        precision_policy=precision_policy,
        precision_pair_count_multiple=supercycle,
    )

BenchmarkPairedRunRecord dataclass

One command attempt within a scheduled paired collection block.

pair_index identifies the target pair. block_attempt identifies an adjacent two-command attempt at that pair; a block contributes inference evidence only when both variants succeed in the same block attempt.

Attributes:

Name	Type	Description
index	int	One-based command-attempt index across the collection.
pair_index	int	One-based target-pair index.
block_attempt	int	One-based atomic-block attempt for that target pair.
variant	PairedVariant	Baseline or candidate member.
pair_order	PairedOrder	Scheduled AB or BA orientation.
order_position	int	One for the first command in the block, otherwise two.
cell_order_index	int	Balanced cell-order row shared by the pair.
status	CollectionRunStatus	Whether this command produced accepted benchmark evidence.
path	Path	Benchmark JSON path for the command attempt.
returncode	int | None	Child-process return code, when the command started.
started_at	str	UTC ISO 8601 timestamp for the command attempt.
duration_seconds	float	Child command and validation duration.
error	str | None	Failure reason for an unsuccessful command attempt.
warnings	tuple[str, ...]	Non-blocking environment diagnostics.
commit	str | None	Source commit reported by pytest-benchmark, when present.
environment_fingerprint	str | None	SHA-256 environment fingerprint, when valid.

Source code in src/benchmatrix/bench_collection.py

@dataclass(frozen=True, slots=True)
class BenchmarkPairedRunRecord:
    """One command attempt within a scheduled paired collection block.

    ``pair_index`` identifies the target pair. ``block_attempt`` identifies an
    adjacent two-command attempt at that pair; a block contributes inference
    evidence only when both variants succeed in the same block attempt.

    Attributes:
        index: One-based command-attempt index across the collection.
        pair_index: One-based target-pair index.
        block_attempt: One-based atomic-block attempt for that target pair.
        variant: Baseline or candidate member.
        pair_order: Scheduled AB or BA orientation.
        order_position: One for the first command in the block, otherwise two.
        cell_order_index: Balanced cell-order row shared by the pair.
        status: Whether this command produced accepted benchmark evidence.
        path: Benchmark JSON path for the command attempt.
        returncode: Child-process return code, when the command started.
        started_at: UTC ISO 8601 timestamp for the command attempt.
        duration_seconds: Child command and validation duration.
        error: Failure reason for an unsuccessful command attempt.
        warnings: Non-blocking environment diagnostics.
        commit: Source commit reported by pytest-benchmark, when present.
        environment_fingerprint: SHA-256 environment fingerprint, when valid.
    """

    index: int
    pair_index: int
    block_attempt: int
    variant: PairedVariant
    pair_order: PairedOrder
    order_position: int
    cell_order_index: int
    status: CollectionRunStatus
    path: Path
    returncode: int | None
    started_at: str
    duration_seconds: float
    error: str | None = None
    warnings: tuple[str, ...] = ()
    commit: str | None = None
    environment_fingerprint: str | None = None

    def __post_init__(self) -> None:
        """Normalize and validate one paired command record."""
        for name, value in (
            ("index", self.index),
            ("pair_index", self.pair_index),
            ("block_attempt", self.block_attempt),
            ("cell_order_index", self.cell_order_index),
        ):
            if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
                raise ValueError(f"BenchmarkPairedRunRecord.{name} must be a positive integer.")
        if self.variant not in {"baseline", "candidate"}:
            raise ValueError(f"Unsupported paired collection variant: {self.variant!r}.")
        if self.pair_order not in {"AB", "BA"}:
            raise ValueError(f"Unsupported paired collection order: {self.pair_order!r}.")
        if self.order_position != _variant_order_position(self.variant, self.pair_order):
            raise ValueError("BenchmarkPairedRunRecord.order_position is inconsistent with variant and pair_order.")
        if self.status not in {"succeeded", "failed"}:
            raise ValueError(f"Unsupported benchmark collection status: {self.status!r}.")
        if self.returncode is not None and (isinstance(self.returncode, bool) or not isinstance(self.returncode, int)):
            raise TypeError("BenchmarkPairedRunRecord.returncode must be an integer or None.")
        _validate_timestamp(self.started_at, field_name="BenchmarkPairedRunRecord.started_at")
        if (
            isinstance(self.duration_seconds, bool)
            or not isinstance(self.duration_seconds, int | float)
            or self.duration_seconds < 0.0
        ):
            raise ValueError("BenchmarkPairedRunRecord.duration_seconds must be a non-negative number.")

        warnings = tuple(self.warnings)
        if any(not isinstance(warning, str) or not warning for warning in warnings):
            raise ValueError("BenchmarkPairedRunRecord.warnings must contain non-empty strings.")
        if self.commit is not None and (not isinstance(self.commit, str) or not self.commit):
            raise ValueError("BenchmarkPairedRunRecord.commit must be a non-empty string or None.")
        if self.environment_fingerprint is not None:
            _validate_fingerprint(
                self.environment_fingerprint,
                field_name="BenchmarkPairedRunRecord.environment_fingerprint",
            )
        if self.status == "succeeded":
            if self.returncode != 0 or self.error is not None:
                raise ValueError("Successful paired records require returncode 0 and no error.")
            if self.environment_fingerprint is None:
                raise ValueError("Successful paired records require an environment fingerprint.")
        elif not isinstance(self.error, str) or not self.error:
            raise ValueError("Failed paired records require a non-empty error.")

        object.__setattr__(self, "path", Path(self.path))
        object.__setattr__(self, "duration_seconds", float(self.duration_seconds))
        object.__setattr__(self, "warnings", warnings)

__post_init__

__post_init__() -> None

Normalize and validate one paired command record.

Source code in src/benchmatrix/bench_collection.py

def __post_init__(self) -> None:
    """Normalize and validate one paired command record."""
    for name, value in (
        ("index", self.index),
        ("pair_index", self.pair_index),
        ("block_attempt", self.block_attempt),
        ("cell_order_index", self.cell_order_index),
    ):
        if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
            raise ValueError(f"BenchmarkPairedRunRecord.{name} must be a positive integer.")
    if self.variant not in {"baseline", "candidate"}:
        raise ValueError(f"Unsupported paired collection variant: {self.variant!r}.")
    if self.pair_order not in {"AB", "BA"}:
        raise ValueError(f"Unsupported paired collection order: {self.pair_order!r}.")
    if self.order_position != _variant_order_position(self.variant, self.pair_order):
        raise ValueError("BenchmarkPairedRunRecord.order_position is inconsistent with variant and pair_order.")
    if self.status not in {"succeeded", "failed"}:
        raise ValueError(f"Unsupported benchmark collection status: {self.status!r}.")
    if self.returncode is not None and (isinstance(self.returncode, bool) or not isinstance(self.returncode, int)):
        raise TypeError("BenchmarkPairedRunRecord.returncode must be an integer or None.")
    _validate_timestamp(self.started_at, field_name="BenchmarkPairedRunRecord.started_at")
    if (
        isinstance(self.duration_seconds, bool)
        or not isinstance(self.duration_seconds, int | float)
        or self.duration_seconds < 0.0
    ):
        raise ValueError("BenchmarkPairedRunRecord.duration_seconds must be a non-negative number.")

    warnings = tuple(self.warnings)
    if any(not isinstance(warning, str) or not warning for warning in warnings):
        raise ValueError("BenchmarkPairedRunRecord.warnings must contain non-empty strings.")
    if self.commit is not None and (not isinstance(self.commit, str) or not self.commit):
        raise ValueError("BenchmarkPairedRunRecord.commit must be a non-empty string or None.")
    if self.environment_fingerprint is not None:
        _validate_fingerprint(
            self.environment_fingerprint,
            field_name="BenchmarkPairedRunRecord.environment_fingerprint",
        )
    if self.status == "succeeded":
        if self.returncode != 0 or self.error is not None:
            raise ValueError("Successful paired records require returncode 0 and no error.")
        if self.environment_fingerprint is None:
            raise ValueError("Successful paired records require an environment fingerprint.")
    elif not isinstance(self.error, str) or not self.error:
        raise ValueError("Failed paired records require a non-empty error.")

    object.__setattr__(self, "path", Path(self.path))
    object.__setattr__(self, "duration_seconds", float(self.duration_seconds))
    object.__setattr__(self, "warnings", warnings)

BenchmarkPairSchedule dataclass

One deterministic baseline/candidate collection block.

Attributes:

Name	Type	Description
pair_index	int	One-based target-pair index.
pair_order	PairedOrder	AB for baseline first or BA for candidate first.
cell_order_index	int	One-based balanced matrix-order row used by both variants in the block.

Source code in src/benchmatrix/bench_collection.py

@dataclass(frozen=True, slots=True)
class BenchmarkPairSchedule:
    """One deterministic baseline/candidate collection block.

    Attributes:
        pair_index: One-based target-pair index.
        pair_order: ``AB`` for baseline first or ``BA`` for candidate first.
        cell_order_index: One-based balanced matrix-order row used by both
            variants in the block.
    """

    pair_index: int
    pair_order: PairedOrder
    cell_order_index: int

    def __post_init__(self) -> None:
        """Validate a scheduled pair."""
        if isinstance(self.pair_index, bool) or not isinstance(self.pair_index, int) or self.pair_index <= 0:
            raise ValueError("BenchmarkPairSchedule.pair_index must be a positive integer.")
        if self.pair_order not in {"AB", "BA"}:
            raise ValueError(f"Unsupported paired collection order: {self.pair_order!r}.")
        if (
            isinstance(self.cell_order_index, bool)
            or not isinstance(self.cell_order_index, int)
            or self.cell_order_index <= 0
        ):
            raise ValueError("BenchmarkPairSchedule.cell_order_index must be a positive integer.")

    @property
    def variants(self) -> tuple[PairedVariant, PairedVariant]:
        """Return variants in their scheduled execution order."""
        return ("baseline", "candidate") if self.pair_order == "AB" else ("candidate", "baseline")

variants property

variants: tuple[PairedVariant, PairedVariant]

Return variants in their scheduled execution order.

__post_init__

__post_init__() -> None

Validate a scheduled pair.

Source code in src/benchmatrix/bench_collection.py

def __post_init__(self) -> None:
    """Validate a scheduled pair."""
    if isinstance(self.pair_index, bool) or not isinstance(self.pair_index, int) or self.pair_index <= 0:
        raise ValueError("BenchmarkPairSchedule.pair_index must be a positive integer.")
    if self.pair_order not in {"AB", "BA"}:
        raise ValueError(f"Unsupported paired collection order: {self.pair_order!r}.")
    if (
        isinstance(self.cell_order_index, bool)
        or not isinstance(self.cell_order_index, int)
        or self.cell_order_index <= 0
    ):
        raise ValueError("BenchmarkPairSchedule.cell_order_index must be a positive integer.")

BenchmarkRunGroup dataclass

A manifest-backed collection of repeated benchmark attempts.

Only successful records appear in runs and can contribute evidence. Failed attempts remain available in records for lifecycle diagnostics.

Attributes:

Name	Type	Description
runs	tuple[BenchmarkRun, ...]	Successfully parsed benchmark runs in attempt order.
records	tuple[BenchmarkRunRecord, ...]	All attempted collection records.
command	tuple[str, ...]	Original pytest command before output-path injection.
created_at	str	UTC ISO 8601 collection timestamp.
cwd	Path	Working directory inherited by the child commands.
commit	str | None	Commit reported by the first successful run, when present.
environment_fingerprint	str | None	Environment fingerprint from the first successful run.
expected_cells	tuple[BenchmarkCell, ...]	Matrix cells established by the first successful run.
requested_runs	int	Number of successful runs requested.
manifest_path	Path	Source manifest path.

Source code in src/benchmatrix/bench_collection.py

@dataclass(frozen=True, slots=True)
class BenchmarkRunGroup:
    """A manifest-backed collection of repeated benchmark attempts.

    Only successful records appear in ``runs`` and can contribute evidence.
    Failed attempts remain available in ``records`` for lifecycle diagnostics.

    Attributes:
        runs: Successfully parsed benchmark runs in attempt order.
        records: All attempted collection records.
        command: Original pytest command before output-path injection.
        created_at: UTC ISO 8601 collection timestamp.
        cwd: Working directory inherited by the child commands.
        commit: Commit reported by the first successful run, when present.
        environment_fingerprint: Environment fingerprint from the first
            successful run.
        expected_cells: Matrix cells established by the first successful run.
        requested_runs: Number of successful runs requested.
        manifest_path: Source manifest path.
    """

    runs: tuple[BenchmarkRun, ...]
    records: tuple[BenchmarkRunRecord, ...]
    command: tuple[str, ...]
    created_at: str
    cwd: Path
    commit: str | None
    environment_fingerprint: str | None
    expected_cells: tuple[BenchmarkCell, ...]
    requested_runs: int
    manifest_path: Path

    def __post_init__(self) -> None:
        """Normalize containers and validate collection invariants."""
        runs = tuple(self.runs)
        records = tuple(self.records)
        command = tuple(self.command)
        expected_cells = tuple(self.expected_cells)

        if not command or any(not isinstance(argument, str) or not argument for argument in command):
            raise ValueError("BenchmarkRunGroup.command must contain non-empty strings.")
        _validate_timestamp(self.created_at, field_name="BenchmarkRunGroup.created_at")
        if self.commit is not None and (not isinstance(self.commit, str) or not self.commit):
            raise ValueError("BenchmarkRunGroup.commit must be a non-empty string or None.")
        if self.environment_fingerprint is not None:
            _validate_fingerprint(
                self.environment_fingerprint,
                field_name="BenchmarkRunGroup.environment_fingerprint",
            )
        if isinstance(self.requested_runs, bool) or not isinstance(self.requested_runs, int):
            raise TypeError("BenchmarkRunGroup.requested_runs must be an integer.")
        if self.requested_runs <= 0:
            raise ValueError("BenchmarkRunGroup.requested_runs must be positive.")
        if tuple(record.index for record in records) != tuple(range(1, len(records) + 1)):
            raise ValueError("BenchmarkRunGroup record indexes must be contiguous and one-based.")
        if len(runs) != sum(record.status == "succeeded" for record in records):
            raise ValueError("BenchmarkRunGroup runs must align with successful records.")
        if len(runs) > self.requested_runs:
            raise ValueError("BenchmarkRunGroup has more successful runs than requested.")
        if len(set(expected_cells)) != len(expected_cells):
            raise ValueError("BenchmarkRunGroup.expected_cells must not contain duplicates.")
        for cell in expected_cells:
            _validate_cell(cell)
        if runs and not expected_cells:
            raise ValueError("BenchmarkRunGroup with successful runs requires expected cells.")
        if not runs and (self.commit is not None or self.environment_fingerprint is not None or expected_cells):
            raise ValueError("BenchmarkRunGroup without successful runs cannot define an anchor.")
        for run in runs:
            if not isinstance(run, BenchmarkRun):
                raise TypeError("BenchmarkRunGroup.runs must contain BenchmarkRun values.")
            if _run_cells(run) != expected_cells:
                raise ValueError("BenchmarkRunGroup run matrix does not match expected_cells.")

        object.__setattr__(self, "runs", runs)
        object.__setattr__(self, "records", records)
        object.__setattr__(self, "command", command)
        object.__setattr__(self, "cwd", Path(self.cwd))
        object.__setattr__(self, "expected_cells", expected_cells)
        object.__setattr__(self, "manifest_path", Path(self.manifest_path))

    @property
    def successful_count(self) -> int:
        """Return the number of accepted benchmark runs."""
        return len(self.runs)

    @property
    def failed_count(self) -> int:
        """Return the number of failed attempts."""
        return sum(record.status == "failed" for record in self.records)

    @property
    def attempted_count(self) -> int:
        """Return the number of completed attempts."""
        return len(self.records)

    @property
    def is_complete(self) -> bool:
        """Return whether the requested successful-run target was reached."""
        return self.successful_count == self.requested_runs

    @property
    def pending_count(self) -> int:
        """Return initial collection slots that have not been attempted."""
        return max(0, self.requested_runs - self.attempted_count)

    @property
    def retry_count(self) -> int:
        """Return attempts appended after the initial collection slots."""
        return max(0, self.attempted_count - self.requested_runs)

    @property
    def remaining_count(self) -> int:
        """Return additional successful runs needed for completeness."""
        return self.requested_runs - self.successful_count

    @property
    def failed_records(self) -> tuple[BenchmarkRunRecord, ...]:
        """Return failed attempts in collection order."""
        return tuple(record for record in self.records if record.status == "failed")

    def compare_to(
        self,
        candidate: BenchmarkRunGroup,
        *,
        compatibility_policy: RunCompatibilityPolicy | None = None,
        regression_policy: RegressionPolicy | None = None,
        evidence_policy: EvidencePolicy | None = None,
        inference_policy: InferencePolicy | None = None,
        precision_policy: PrecisionPolicy | None = None,
    ) -> BenchmarkRunComparison:
        """Compare this repeated baseline collection with a candidate.

        Args:
            candidate: Repeated candidate collection.
            compatibility_policy: Environment checks to apply.
            regression_policy: Thresholds used to classify cell changes.
            evidence_policy: Minimum repeated-run evidence to require.
            inference_policy: Statistical inference and multiplicity controls.
            precision_policy: Optional precision-planning policy. Independent
                groups require its planning mode to remain disabled.

        Returns:
            A matrix-aware repeated-run comparison.

        Raises:
            ValueError: If either collection has no successful runs.
        """
        return compare_benchmark_run_groups(
            self.runs,
            candidate.runs,
            compatibility_policy=compatibility_policy,
            regression_policy=regression_policy,
            evidence_policy=evidence_policy,
            inference_policy=inference_policy,
            precision_policy=precision_policy,
        )

successful_count property

successful_count: int

Return the number of accepted benchmark runs.

failed_count property

failed_count: int

Return the number of failed attempts.

attempted_count property

attempted_count: int

Return the number of completed attempts.

is_complete property

is_complete: bool

Return whether the requested successful-run target was reached.

pending_count property

pending_count: int

Return initial collection slots that have not been attempted.

retry_count property

retry_count: int

Return attempts appended after the initial collection slots.

remaining_count property

remaining_count: int

Return additional successful runs needed for completeness.

failed_records property

failed_records: tuple[BenchmarkRunRecord, ...]

Return failed attempts in collection order.

__post_init__

__post_init__() -> None

Normalize containers and validate collection invariants.

Source code in src/benchmatrix/bench_collection.py

def __post_init__(self) -> None:
    """Normalize containers and validate collection invariants."""
    runs = tuple(self.runs)
    records = tuple(self.records)
    command = tuple(self.command)
    expected_cells = tuple(self.expected_cells)

    if not command or any(not isinstance(argument, str) or not argument for argument in command):
        raise ValueError("BenchmarkRunGroup.command must contain non-empty strings.")
    _validate_timestamp(self.created_at, field_name="BenchmarkRunGroup.created_at")
    if self.commit is not None and (not isinstance(self.commit, str) or not self.commit):
        raise ValueError("BenchmarkRunGroup.commit must be a non-empty string or None.")
    if self.environment_fingerprint is not None:
        _validate_fingerprint(
            self.environment_fingerprint,
            field_name="BenchmarkRunGroup.environment_fingerprint",
        )
    if isinstance(self.requested_runs, bool) or not isinstance(self.requested_runs, int):
        raise TypeError("BenchmarkRunGroup.requested_runs must be an integer.")
    if self.requested_runs <= 0:
        raise ValueError("BenchmarkRunGroup.requested_runs must be positive.")
    if tuple(record.index for record in records) != tuple(range(1, len(records) + 1)):
        raise ValueError("BenchmarkRunGroup record indexes must be contiguous and one-based.")
    if len(runs) != sum(record.status == "succeeded" for record in records):
        raise ValueError("BenchmarkRunGroup runs must align with successful records.")
    if len(runs) > self.requested_runs:
        raise ValueError("BenchmarkRunGroup has more successful runs than requested.")
    if len(set(expected_cells)) != len(expected_cells):
        raise ValueError("BenchmarkRunGroup.expected_cells must not contain duplicates.")
    for cell in expected_cells:
        _validate_cell(cell)
    if runs and not expected_cells:
        raise ValueError("BenchmarkRunGroup with successful runs requires expected cells.")
    if not runs and (self.commit is not None or self.environment_fingerprint is not None or expected_cells):
        raise ValueError("BenchmarkRunGroup without successful runs cannot define an anchor.")
    for run in runs:
        if not isinstance(run, BenchmarkRun):
            raise TypeError("BenchmarkRunGroup.runs must contain BenchmarkRun values.")
        if _run_cells(run) != expected_cells:
            raise ValueError("BenchmarkRunGroup run matrix does not match expected_cells.")

    object.__setattr__(self, "runs", runs)
    object.__setattr__(self, "records", records)
    object.__setattr__(self, "command", command)
    object.__setattr__(self, "cwd", Path(self.cwd))
    object.__setattr__(self, "expected_cells", expected_cells)
    object.__setattr__(self, "manifest_path", Path(self.manifest_path))

compare_to

compare_to(
    candidate: BenchmarkRunGroup,
    *,
    compatibility_policy: RunCompatibilityPolicy
    | None = None,
    regression_policy: RegressionPolicy | None = None,
    evidence_policy: EvidencePolicy | None = None,
    inference_policy: InferencePolicy | None = None,
    precision_policy: PrecisionPolicy | None = None,
) -> BenchmarkRunComparison

Compare this repeated baseline collection with a candidate.

Parameters:

Name	Type	Description	Default
candidate	BenchmarkRunGroup	Repeated candidate collection.	required
compatibility_policy	RunCompatibilityPolicy | None	Environment checks to apply.	None
regression_policy	RegressionPolicy | None	Thresholds used to classify cell changes.	None
evidence_policy	EvidencePolicy | None	Minimum repeated-run evidence to require.	None
inference_policy	InferencePolicy | None	Statistical inference and multiplicity controls.	None
precision_policy	PrecisionPolicy | None	Optional precision-planning policy. Independent groups require its planning mode to remain disabled.	None

Returns:

Type	Description
BenchmarkRunComparison	A matrix-aware repeated-run comparison.

Raises:

Type	Description
ValueError	If either collection has no successful runs.

Source code in src/benchmatrix/bench_collection.py

def compare_to(
    self,
    candidate: BenchmarkRunGroup,
    *,
    compatibility_policy: RunCompatibilityPolicy | None = None,
    regression_policy: RegressionPolicy | None = None,
    evidence_policy: EvidencePolicy | None = None,
    inference_policy: InferencePolicy | None = None,
    precision_policy: PrecisionPolicy | None = None,
) -> BenchmarkRunComparison:
    """Compare this repeated baseline collection with a candidate.

    Args:
        candidate: Repeated candidate collection.
        compatibility_policy: Environment checks to apply.
        regression_policy: Thresholds used to classify cell changes.
        evidence_policy: Minimum repeated-run evidence to require.
        inference_policy: Statistical inference and multiplicity controls.
        precision_policy: Optional precision-planning policy. Independent
            groups require its planning mode to remain disabled.

    Returns:
        A matrix-aware repeated-run comparison.

    Raises:
        ValueError: If either collection has no successful runs.
    """
    return compare_benchmark_run_groups(
        self.runs,
        candidate.runs,
        compatibility_policy=compatibility_policy,
        regression_policy=regression_policy,
        evidence_policy=evidence_policy,
        inference_policy=inference_policy,
        precision_policy=precision_policy,
    )

BenchmarkRunPair dataclass

One complete atomic baseline/candidate collection block.

Attributes:

Name	Type	Description
pair_index	int	One-based target-pair index.
block_attempt	int	Successful atomic-block attempt for the pair.
pair_order	PairedOrder	AB or BA execution orientation.
cell_order	tuple[BenchmarkCell, ...]	Balanced matrix order used by both variants.
baseline	BenchmarkRun	Baseline benchmark run.
candidate	BenchmarkRun	Candidate benchmark run.
baseline_record	BenchmarkPairedRunRecord	Manifest record for baseline.
candidate_record	BenchmarkPairedRunRecord	Manifest record for candidate.

Source code in src/benchmatrix/bench_collection.py

@dataclass(frozen=True, slots=True)
class BenchmarkRunPair:
    """One complete atomic baseline/candidate collection block.

    Attributes:
        pair_index: One-based target-pair index.
        block_attempt: Successful atomic-block attempt for the pair.
        pair_order: AB or BA execution orientation.
        cell_order: Balanced matrix order used by both variants.
        baseline: Baseline benchmark run.
        candidate: Candidate benchmark run.
        baseline_record: Manifest record for ``baseline``.
        candidate_record: Manifest record for ``candidate``.
    """

    pair_index: int
    block_attempt: int
    pair_order: PairedOrder
    cell_order: tuple[BenchmarkCell, ...]
    baseline: BenchmarkRun
    candidate: BenchmarkRun
    baseline_record: BenchmarkPairedRunRecord
    candidate_record: BenchmarkPairedRunRecord

    def __post_init__(self) -> None:
        """Validate the matched-block contract."""
        if self.baseline_record.status != "succeeded" or self.candidate_record.status != "succeeded":
            raise ValueError("BenchmarkRunPair requires two successful records.")
        for record, variant in (
            (self.baseline_record, "baseline"),
            (self.candidate_record, "candidate"),
        ):
            if (
                record.pair_index != self.pair_index
                or record.block_attempt != self.block_attempt
                or record.variant != variant
                or record.pair_order != self.pair_order
            ):
                raise ValueError("BenchmarkRunPair records do not match the pair identity.")
        if _run_cell_order(self.baseline) != self.cell_order or _run_cell_order(self.candidate) != self.cell_order:
            raise ValueError("BenchmarkRunPair runs do not match their scheduled cell order.")

__post_init__

__post_init__() -> None

Validate the matched-block contract.

Source code in src/benchmatrix/bench_collection.py

def __post_init__(self) -> None:
    """Validate the matched-block contract."""
    if self.baseline_record.status != "succeeded" or self.candidate_record.status != "succeeded":
        raise ValueError("BenchmarkRunPair requires two successful records.")
    for record, variant in (
        (self.baseline_record, "baseline"),
        (self.candidate_record, "candidate"),
    ):
        if (
            record.pair_index != self.pair_index
            or record.block_attempt != self.block_attempt
            or record.variant != variant
            or record.pair_order != self.pair_order
        ):
            raise ValueError("BenchmarkRunPair records do not match the pair identity.")
    if _run_cell_order(self.baseline) != self.cell_order or _run_cell_order(self.candidate) != self.cell_order:
        raise ValueError("BenchmarkRunPair runs do not match their scheduled cell order.")

BenchmarkRunRecord dataclass

One attempted run recorded in a collection manifest.

Attributes:

Name	Type	Description
index	int	One-based attempt number.
status	CollectionRunStatus	Whether the command produced accepted benchmark evidence.
path	Path	Benchmark JSON path for the attempt.
returncode	int | None	Child-process return code, when the command started.
started_at	str	UTC ISO 8601 timestamp for the attempt.
duration_seconds	float	Child command and validation duration.
error	str | None	Failure reason for an unsuccessful attempt.
warnings	tuple[str, ...]	Non-blocking environment diagnostics.
commit	str | None	Source commit reported by pytest-benchmark, when present.
environment_fingerprint	str | None	SHA-256 fingerprint of environment metadata, when a valid run was produced.

Source code in src/benchmatrix/bench_collection.py

@dataclass(frozen=True, slots=True)
class BenchmarkRunRecord:
    """One attempted run recorded in a collection manifest.

    Attributes:
        index: One-based attempt number.
        status: Whether the command produced accepted benchmark evidence.
        path: Benchmark JSON path for the attempt.
        returncode: Child-process return code, when the command started.
        started_at: UTC ISO 8601 timestamp for the attempt.
        duration_seconds: Child command and validation duration.
        error: Failure reason for an unsuccessful attempt.
        warnings: Non-blocking environment diagnostics.
        commit: Source commit reported by pytest-benchmark, when present.
        environment_fingerprint: SHA-256 fingerprint of environment metadata,
            when a valid run was produced.
    """

    index: int
    status: CollectionRunStatus
    path: Path
    returncode: int | None
    started_at: str
    duration_seconds: float
    error: str | None = None
    warnings: tuple[str, ...] = ()
    commit: str | None = None
    environment_fingerprint: str | None = None

    def __post_init__(self) -> None:
        """Normalize and validate an attempted-run record."""
        if isinstance(self.index, bool) or not isinstance(self.index, int) or self.index <= 0:
            raise ValueError("BenchmarkRunRecord.index must be a positive integer.")
        if self.status not in {"succeeded", "failed"}:
            raise ValueError(f"Unsupported benchmark collection status: {self.status!r}.")
        if self.returncode is not None and (isinstance(self.returncode, bool) or not isinstance(self.returncode, int)):
            raise TypeError("BenchmarkRunRecord.returncode must be an integer or None.")
        _validate_timestamp(self.started_at, field_name="BenchmarkRunRecord.started_at")
        if (
            isinstance(self.duration_seconds, bool)
            or not isinstance(self.duration_seconds, int | float)
            or self.duration_seconds < 0.0
        ):
            raise ValueError("BenchmarkRunRecord.duration_seconds must be a non-negative number.")

        warnings = tuple(self.warnings)
        if any(not isinstance(warning, str) or not warning for warning in warnings):
            raise ValueError("BenchmarkRunRecord.warnings must contain non-empty strings.")
        if self.commit is not None and (not isinstance(self.commit, str) or not self.commit):
            raise ValueError("BenchmarkRunRecord.commit must be a non-empty string or None.")
        if self.environment_fingerprint is not None:
            _validate_fingerprint(
                self.environment_fingerprint,
                field_name="BenchmarkRunRecord.environment_fingerprint",
            )

        if self.status == "succeeded":
            if self.returncode != 0 or self.error is not None:
                raise ValueError("Successful benchmark records require returncode 0 and no error.")
            if self.environment_fingerprint is None:
                raise ValueError("Successful benchmark records require an environment fingerprint.")
        elif not isinstance(self.error, str) or not self.error:
            raise ValueError("Failed benchmark records require a non-empty error.")

        object.__setattr__(self, "path", Path(self.path))
        object.__setattr__(self, "duration_seconds", float(self.duration_seconds))
        object.__setattr__(self, "warnings", warnings)

__post_init__

__post_init__() -> None

Normalize and validate an attempted-run record.

Source code in src/benchmatrix/bench_collection.py

def __post_init__(self) -> None:
    """Normalize and validate an attempted-run record."""
    if isinstance(self.index, bool) or not isinstance(self.index, int) or self.index <= 0:
        raise ValueError("BenchmarkRunRecord.index must be a positive integer.")
    if self.status not in {"succeeded", "failed"}:
        raise ValueError(f"Unsupported benchmark collection status: {self.status!r}.")
    if self.returncode is not None and (isinstance(self.returncode, bool) or not isinstance(self.returncode, int)):
        raise TypeError("BenchmarkRunRecord.returncode must be an integer or None.")
    _validate_timestamp(self.started_at, field_name="BenchmarkRunRecord.started_at")
    if (
        isinstance(self.duration_seconds, bool)
        or not isinstance(self.duration_seconds, int | float)
        or self.duration_seconds < 0.0
    ):
        raise ValueError("BenchmarkRunRecord.duration_seconds must be a non-negative number.")

    warnings = tuple(self.warnings)
    if any(not isinstance(warning, str) or not warning for warning in warnings):
        raise ValueError("BenchmarkRunRecord.warnings must contain non-empty strings.")
    if self.commit is not None and (not isinstance(self.commit, str) or not self.commit):
        raise ValueError("BenchmarkRunRecord.commit must be a non-empty string or None.")
    if self.environment_fingerprint is not None:
        _validate_fingerprint(
            self.environment_fingerprint,
            field_name="BenchmarkRunRecord.environment_fingerprint",
        )

    if self.status == "succeeded":
        if self.returncode != 0 or self.error is not None:
            raise ValueError("Successful benchmark records require returncode 0 and no error.")
        if self.environment_fingerprint is None:
            raise ValueError("Successful benchmark records require an environment fingerprint.")
    elif not isinstance(self.error, str) or not self.error:
        raise ValueError("Failed benchmark records require a non-empty error.")

    object.__setattr__(self, "path", Path(self.path))
    object.__setattr__(self, "duration_seconds", float(self.duration_seconds))
    object.__setattr__(self, "warnings", warnings)

BenchmarkComparison dataclass

Comparison for one implementation, case, and metric matrix cell.

percent_change is the conventional candidate change from baseline, while improvement_percent is direction-aware and therefore positive when the candidate is better.

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class BenchmarkComparison:
    """Comparison for one implementation, case, and metric matrix cell.

    ``percent_change`` is the conventional candidate change from baseline,
    while ``improvement_percent`` is direction-aware and therefore positive
    when the candidate is better.
    """

    implementation_name: str
    case_name: str
    metric_name: MetricName
    statistic: str
    direction: ComparisonDirection
    status: ComparisonStatus
    baseline_value: float | None
    candidate_value: float | None
    ratio: float | None
    percent_change: float | None
    improvement_percent: float | None
    regression: RegressionClassification
    threshold_percent: float
    unit: str
    baseline_evidence: BenchmarkEvidence | None = None
    candidate_evidence: BenchmarkEvidence | None = None
    improvement_low_percent: float | None = None
    improvement_high_percent: float | None = None
    reason: str | None = None
    inference: BenchmarkInference | None = None
    precision: PrecisionPlan | None = None

BenchmarkEvidence dataclass

Trust diagnostics for one side of a matrix-cell comparison.

Attributes:

Name	Type	Description
provided_run_count	int	Files supplied for this side.
observed_run_count	int	Files containing this matrix cell.
rounds	tuple[int | None, ...]	Positive pytest-benchmark round counts aligned to the files.
iterations	tuple[int | None, ...]	Positive iteration counts aligned to the files.
sample_counts	tuple[int, ...]	Raw timing sample counts aligned to the files.
sample_count	int	Total pooled raw timing samples.
iqr	float | None	Interquartile range of pooled timing samples in seconds. Retained as a descriptive compatibility field; evidence gates use the corresponding per-run diagnostics.
coefficient_of_variation	float | None	Pooled timing-sample population standard deviation divided by the absolute mean.
outlier_count	int | None	Samples outside the pooled 1.5-IQR Tukey fences.
outlier_fraction	float | None	Outlier count divided by total sample count.
adequate	bool	Whether the configured evidence policy was satisfied.
issues	tuple[str, ...]	Human-readable reasons evidence is inadequate.
run_iqrs	tuple[float | None, ...]	Per-run timing-sample interquartile ranges.
run_coefficients_of_variation	tuple[float | None, ...]	Per-run coefficients of variation.
run_outlier_counts	tuple[int | None, ...]	Per-run Tukey-outlier counts.
run_outlier_fractions	tuple[float | None, ...]	Per-run Tukey-outlier fractions.

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class BenchmarkEvidence:
    """Trust diagnostics for one side of a matrix-cell comparison.

    Attributes:
        provided_run_count: Files supplied for this side.
        observed_run_count: Files containing this matrix cell.
        rounds: Positive pytest-benchmark round counts aligned to the files.
        iterations: Positive iteration counts aligned to the files.
        sample_counts: Raw timing sample counts aligned to the files.
        sample_count: Total pooled raw timing samples.
        iqr: Interquartile range of pooled timing samples in seconds. Retained
            as a descriptive compatibility field; evidence gates use the
            corresponding per-run diagnostics.
        coefficient_of_variation: Pooled timing-sample population standard
            deviation divided by the absolute mean.
        outlier_count: Samples outside the pooled 1.5-IQR Tukey fences.
        outlier_fraction: Outlier count divided by total sample count.
        adequate: Whether the configured evidence policy was satisfied.
        issues: Human-readable reasons evidence is inadequate.
        run_iqrs: Per-run timing-sample interquartile ranges.
        run_coefficients_of_variation: Per-run coefficients of variation.
        run_outlier_counts: Per-run Tukey-outlier counts.
        run_outlier_fractions: Per-run Tukey-outlier fractions.
    """

    provided_run_count: int
    observed_run_count: int
    rounds: tuple[int | None, ...]
    iterations: tuple[int | None, ...]
    sample_counts: tuple[int, ...]
    sample_count: int
    iqr: float | None
    coefficient_of_variation: float | None
    outlier_count: int | None
    outlier_fraction: float | None
    adequate: bool
    issues: tuple[str, ...]
    run_iqrs: tuple[float | None, ...] = ()
    run_coefficients_of_variation: tuple[float | None, ...] = ()
    run_outlier_counts: tuple[int | None, ...] = ()
    run_outlier_fractions: tuple[float | None, ...] = ()

BenchmarkInference dataclass

Statistical inference for one benchmark matrix cell.

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class BenchmarkInference:
    """Statistical inference for one benchmark matrix cell."""

    method: IntervalMethod
    estimand: str
    design: ComparisonDesign
    confidence_level: float
    adjusted_confidence_level: float
    multiplicity: MultiplicityCorrection
    family_size: int
    resamples: int
    random_seed: int
    estimate_percent: float | None
    confidence_low_percent: float | None
    confidence_high_percent: float | None
    warnings: tuple[str, ...] = ()
    issues: tuple[str, ...] = ()
    pair_count: int | None = None
    strata_count: int | None = None

    def __post_init__(self) -> None:
        """Validate and normalize an inference result."""
        if self.method not in {"bca_bootstrap", "percentile_bootstrap"}:
            raise ValueError(f"Unsupported interval method: {self.method!r}.")
        if self.design not in {"independent", "paired"}:
            raise ValueError(f"Unsupported inference design: {self.design!r}.")
        if not isinstance(self.estimand, str) or not self.estimand:
            raise ValueError("BenchmarkInference.estimand must be a non-empty string.")
        confidence_level = _validate_fraction(
            self.confidence_level,
            field_name="BenchmarkInference.confidence_level",
        )
        adjusted_confidence_level = _validate_fraction(
            self.adjusted_confidence_level,
            field_name="BenchmarkInference.adjusted_confidence_level",
        )
        if confidence_level in {0.0, 1.0} or adjusted_confidence_level in {0.0, 1.0}:
            raise ValueError("BenchmarkInference confidence levels must be between zero and one.")
        if adjusted_confidence_level < confidence_level:
            raise ValueError("Adjusted confidence level must not be lower than the nominal confidence level.")
        if self.multiplicity not in {"bonferroni", "none"}:
            raise ValueError(f"Unsupported multiplicity correction: {self.multiplicity!r}.")
        if isinstance(self.family_size, bool) or not isinstance(self.family_size, int) or self.family_size <= 0:
            raise ValueError("BenchmarkInference.family_size must be a positive integer.")
        if isinstance(self.resamples, bool) or not isinstance(self.resamples, int):
            raise TypeError("BenchmarkInference.resamples must be an integer.")
        if self.resamples < 1_000:
            raise ValueError("BenchmarkInference.resamples must be at least 1000.")
        if isinstance(self.random_seed, bool) or not isinstance(self.random_seed, int):
            raise TypeError("BenchmarkInference.random_seed must be an integer.")
        if self.random_seed < 0:
            raise ValueError("BenchmarkInference.random_seed must be non-negative.")
        if self.design == "independent" and (self.pair_count is not None or self.strata_count is not None):
            raise ValueError("Independent inference must not define paired-design counts.")
        if self.design == "paired" and (
            isinstance(self.pair_count, bool) or not isinstance(self.pair_count, int) or self.pair_count <= 0
        ):
            raise ValueError("Paired inference requires a positive integer pair_count.")
        if self.design == "paired" and (
            isinstance(self.strata_count, bool)
            or not isinstance(self.strata_count, int)
            or self.strata_count <= 0
            or self.pair_count is None
            or self.strata_count > self.pair_count
        ):
            raise ValueError("Paired inference requires a valid positive strata_count no greater than pair_count.")
        for field_name, value in (
            ("estimate_percent", self.estimate_percent),
            ("confidence_low_percent", self.confidence_low_percent),
            ("confidence_high_percent", self.confidence_high_percent),
        ):
            if value is not None and not math.isfinite(value):
                raise ValueError(f"BenchmarkInference.{field_name} must be finite or None.")
        present = (
            self.estimate_percent is not None,
            self.confidence_low_percent is not None,
            self.confidence_high_percent is not None,
        )
        if any(present) and not all(present):
            raise ValueError("BenchmarkInference estimate and confidence bounds must be present together.")
        if (
            self.confidence_low_percent is not None
            and self.confidence_high_percent is not None
            and self.confidence_low_percent > self.confidence_high_percent
        ):
            raise ValueError("BenchmarkInference confidence bounds are reversed.")
        warnings = tuple(self.warnings)
        issues = tuple(self.issues)
        if any(not isinstance(item, str) or not item for item in (*warnings, *issues)):
            raise ValueError("BenchmarkInference warnings and issues must contain non-empty strings.")
        object.__setattr__(self, "warnings", warnings)
        object.__setattr__(self, "issues", issues)
        object.__setattr__(self, "confidence_level", confidence_level)
        object.__setattr__(self, "adjusted_confidence_level", adjusted_confidence_level)

    @property
    def adequate(self) -> bool:
        """Return whether a complete confidence interval is available."""
        return not self.issues and self.confidence_low_percent is not None and self.confidence_high_percent is not None

adequate property

adequate: bool

Return whether a complete confidence interval is available.

__post_init__

__post_init__() -> None

Validate and normalize an inference result.

Source code in src/benchmatrix/bench_compare.py

def __post_init__(self) -> None:
    """Validate and normalize an inference result."""
    if self.method not in {"bca_bootstrap", "percentile_bootstrap"}:
        raise ValueError(f"Unsupported interval method: {self.method!r}.")
    if self.design not in {"independent", "paired"}:
        raise ValueError(f"Unsupported inference design: {self.design!r}.")
    if not isinstance(self.estimand, str) or not self.estimand:
        raise ValueError("BenchmarkInference.estimand must be a non-empty string.")
    confidence_level = _validate_fraction(
        self.confidence_level,
        field_name="BenchmarkInference.confidence_level",
    )
    adjusted_confidence_level = _validate_fraction(
        self.adjusted_confidence_level,
        field_name="BenchmarkInference.adjusted_confidence_level",
    )
    if confidence_level in {0.0, 1.0} or adjusted_confidence_level in {0.0, 1.0}:
        raise ValueError("BenchmarkInference confidence levels must be between zero and one.")
    if adjusted_confidence_level < confidence_level:
        raise ValueError("Adjusted confidence level must not be lower than the nominal confidence level.")
    if self.multiplicity not in {"bonferroni", "none"}:
        raise ValueError(f"Unsupported multiplicity correction: {self.multiplicity!r}.")
    if isinstance(self.family_size, bool) or not isinstance(self.family_size, int) or self.family_size <= 0:
        raise ValueError("BenchmarkInference.family_size must be a positive integer.")
    if isinstance(self.resamples, bool) or not isinstance(self.resamples, int):
        raise TypeError("BenchmarkInference.resamples must be an integer.")
    if self.resamples < 1_000:
        raise ValueError("BenchmarkInference.resamples must be at least 1000.")
    if isinstance(self.random_seed, bool) or not isinstance(self.random_seed, int):
        raise TypeError("BenchmarkInference.random_seed must be an integer.")
    if self.random_seed < 0:
        raise ValueError("BenchmarkInference.random_seed must be non-negative.")
    if self.design == "independent" and (self.pair_count is not None or self.strata_count is not None):
        raise ValueError("Independent inference must not define paired-design counts.")
    if self.design == "paired" and (
        isinstance(self.pair_count, bool) or not isinstance(self.pair_count, int) or self.pair_count <= 0
    ):
        raise ValueError("Paired inference requires a positive integer pair_count.")
    if self.design == "paired" and (
        isinstance(self.strata_count, bool)
        or not isinstance(self.strata_count, int)
        or self.strata_count <= 0
        or self.pair_count is None
        or self.strata_count > self.pair_count
    ):
        raise ValueError("Paired inference requires a valid positive strata_count no greater than pair_count.")
    for field_name, value in (
        ("estimate_percent", self.estimate_percent),
        ("confidence_low_percent", self.confidence_low_percent),
        ("confidence_high_percent", self.confidence_high_percent),
    ):
        if value is not None and not math.isfinite(value):
            raise ValueError(f"BenchmarkInference.{field_name} must be finite or None.")
    present = (
        self.estimate_percent is not None,
        self.confidence_low_percent is not None,
        self.confidence_high_percent is not None,
    )
    if any(present) and not all(present):
        raise ValueError("BenchmarkInference estimate and confidence bounds must be present together.")
    if (
        self.confidence_low_percent is not None
        and self.confidence_high_percent is not None
        and self.confidence_low_percent > self.confidence_high_percent
    ):
        raise ValueError("BenchmarkInference confidence bounds are reversed.")
    warnings = tuple(self.warnings)
    issues = tuple(self.issues)
    if any(not isinstance(item, str) or not item for item in (*warnings, *issues)):
        raise ValueError("BenchmarkInference warnings and issues must contain non-empty strings.")
    object.__setattr__(self, "warnings", warnings)
    object.__setattr__(self, "issues", issues)
    object.__setattr__(self, "confidence_level", confidence_level)
    object.__setattr__(self, "adjusted_confidence_level", adjusted_confidence_level)

BenchmarkRunComparison dataclass

Matrix-aware comparison between a baseline and candidate run.

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class BenchmarkRunComparison:
    """Matrix-aware comparison between a baseline and candidate run."""

    baseline: BenchmarkRun
    candidate: BenchmarkRun
    compatibility: RunCompatibilityReport
    regression_policy: RegressionPolicy
    comparisons: tuple[BenchmarkComparison, ...]
    baseline_runs: tuple[BenchmarkRun, ...] = ()
    candidate_runs: tuple[BenchmarkRun, ...] = ()
    evidence_policy: EvidencePolicy = field(default_factory=EvidencePolicy)
    inference_policy: InferencePolicy = field(default_factory=InferencePolicy)
    design: ComparisonDesign = "independent"
    precision_policy: PrecisionPolicy = field(default_factory=PrecisionPolicy)

    @property
    def matched(self) -> tuple[BenchmarkComparison, ...]:
        """Return cells that were compared successfully."""
        return tuple(comparison for comparison in self.comparisons if comparison.status == "matched")

    @property
    def missing(self) -> tuple[BenchmarkComparison, ...]:
        """Return cells absent from either input run."""
        return tuple(
            comparison
            for comparison in self.comparisons
            if comparison.status in {"missing_baseline", "missing_candidate"}
        )

    @property
    def incompatible(self) -> tuple[BenchmarkComparison, ...]:
        """Return cells whose measurement context cannot be compared."""
        return tuple(comparison for comparison in self.comparisons if comparison.status == "incompatible")

    @property
    def improved(self) -> tuple[BenchmarkComparison, ...]:
        """Return comparable cells that exceeded their improvement threshold."""
        return tuple(comparison for comparison in self.comparisons if comparison.regression == "improved")

    @property
    def unchanged(self) -> tuple[BenchmarkComparison, ...]:
        """Return comparable cells whose changes stayed within threshold."""
        return tuple(comparison for comparison in self.comparisons if comparison.regression == "unchanged")

    @property
    def regressed(self) -> tuple[BenchmarkComparison, ...]:
        """Return comparable cells that exceeded their regression threshold."""
        return tuple(comparison for comparison in self.comparisons if comparison.regression == "regressed")

    @property
    def inconclusive(self) -> tuple[BenchmarkComparison, ...]:
        """Return matched cells whose evidence cannot support a decision."""
        return tuple(comparison for comparison in self.comparisons if comparison.regression == "inconclusive")

    @property
    def not_comparable(self) -> tuple[BenchmarkComparison, ...]:
        """Return cells without a trustworthy regression classification."""
        return tuple(comparison for comparison in self.comparisons if comparison.regression == "not_comparable")

    @property
    def is_complete(self) -> bool:
        """Return whether every matrix cell was compared successfully."""
        return len(self.matched) == len(self.comparisons)

    @property
    def is_comparable(self) -> bool:
        """Return whether environment and every matrix cell are comparable."""
        return self.compatibility.is_compatible and self.is_complete and not self.not_comparable

    @property
    def has_regressions(self) -> bool:
        """Return whether any comparable matrix cell regressed."""
        return bool(self.regressed)

    @property
    def passed(self) -> bool:
        """Return whether the comparison is trustworthy and regression-free."""
        return self.is_comparable and not self.has_regressions and not self.inconclusive

matched property

matched: tuple[BenchmarkComparison, ...]

Return cells that were compared successfully.

missing property

missing: tuple[BenchmarkComparison, ...]

Return cells absent from either input run.

incompatible property

incompatible: tuple[BenchmarkComparison, ...]

Return cells whose measurement context cannot be compared.

improved property

improved: tuple[BenchmarkComparison, ...]

Return comparable cells that exceeded their improvement threshold.

unchanged property

unchanged: tuple[BenchmarkComparison, ...]

Return comparable cells whose changes stayed within threshold.

regressed property

regressed: tuple[BenchmarkComparison, ...]

Return comparable cells that exceeded their regression threshold.

inconclusive property

inconclusive: tuple[BenchmarkComparison, ...]

Return matched cells whose evidence cannot support a decision.

not_comparable property

not_comparable: tuple[BenchmarkComparison, ...]

Return cells without a trustworthy regression classification.

is_complete property

is_complete: bool

Return whether every matrix cell was compared successfully.

is_comparable property

is_comparable: bool

Return whether environment and every matrix cell are comparable.

has_regressions property

has_regressions: bool

Return whether any comparable matrix cell regressed.

passed property

passed: bool

Return whether the comparison is trustworthy and regression-free.

EvidencePolicy dataclass

Minimum evidence required for repeated-run classifications.

Parameters:

Name	Type	Description	Default
minimum_runs	int	Minimum files on each side containing a matrix cell.	5
minimum_samples_per_run	int	Minimum raw timing samples required from each observed file.	5
minimum_rounds_per_run	int	Minimum pytest-benchmark rounds required from each observed file.	5
require_rounds	bool	Whether every row must report a positive round count.	True
require_iterations	bool	Whether every row must report a positive iteration count.	True
require_raw_samples_for_inference	bool	Whether every observed row must retain raw per-round durations.	True
minimum_tail_samples_per_run	int	Minimum round-duration observations required from each tail-latency row.	100
require_tail_iterations_one	bool	Whether tail-latency rows must represent individual calls rather than averages of multiple iterations.	True
maximum_cv	float | None	Optional maximum within-run coefficient of variation.	None
maximum_outlier_fraction	float | None	Optional maximum within-run Tukey-outlier fraction.	None

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class EvidencePolicy:
    """Minimum evidence required for repeated-run classifications.

    Args:
        minimum_runs: Minimum files on each side containing a matrix cell.
        minimum_samples_per_run: Minimum raw timing samples required from each
            observed file.
        minimum_rounds_per_run: Minimum pytest-benchmark rounds required from
            each observed file.
        require_rounds: Whether every row must report a positive round count.
        require_iterations: Whether every row must report a positive iteration
            count.
        require_raw_samples_for_inference: Whether every observed row must
            retain raw per-round durations.
        minimum_tail_samples_per_run: Minimum round-duration observations
            required from each tail-latency row.
        require_tail_iterations_one: Whether tail-latency rows must represent
            individual calls rather than averages of multiple iterations.
        maximum_cv: Optional maximum within-run coefficient of variation.
        maximum_outlier_fraction: Optional maximum within-run Tukey-outlier
            fraction.
    """

    minimum_runs: int = 5
    minimum_samples_per_run: int = 5
    minimum_rounds_per_run: int = 5
    require_rounds: bool = True
    require_iterations: bool = True
    require_raw_samples_for_inference: bool = True
    minimum_tail_samples_per_run: int = 100
    require_tail_iterations_one: bool = True
    maximum_cv: float | None = None
    maximum_outlier_fraction: float | None = None

    def __post_init__(self) -> None:
        """Validate evidence thresholds."""
        if isinstance(self.minimum_runs, bool) or not isinstance(self.minimum_runs, int):
            raise TypeError("EvidencePolicy.minimum_runs must be an integer.")
        if self.minimum_runs <= 0:
            raise ValueError("EvidencePolicy.minimum_runs must be a positive integer.")
        if isinstance(self.minimum_samples_per_run, bool) or not isinstance(self.minimum_samples_per_run, int):
            raise TypeError("EvidencePolicy.minimum_samples_per_run must be an integer.")
        if self.minimum_samples_per_run < 0:
            raise ValueError("EvidencePolicy.minimum_samples_per_run must be a non-negative integer.")
        if isinstance(self.minimum_rounds_per_run, bool) or not isinstance(self.minimum_rounds_per_run, int):
            raise TypeError("EvidencePolicy.minimum_rounds_per_run must be an integer.")
        if self.minimum_rounds_per_run < 0:
            raise ValueError("EvidencePolicy.minimum_rounds_per_run must be a non-negative integer.")
        if not isinstance(self.require_rounds, bool):
            raise TypeError("EvidencePolicy.require_rounds must be a boolean.")
        if not isinstance(self.require_iterations, bool):
            raise TypeError("EvidencePolicy.require_iterations must be a boolean.")
        if not isinstance(self.require_raw_samples_for_inference, bool):
            raise TypeError("EvidencePolicy.require_raw_samples_for_inference must be a boolean.")
        if isinstance(self.minimum_tail_samples_per_run, bool) or not isinstance(
            self.minimum_tail_samples_per_run, int
        ):
            raise TypeError("EvidencePolicy.minimum_tail_samples_per_run must be an integer.")
        if self.minimum_tail_samples_per_run < 0:
            raise ValueError("EvidencePolicy.minimum_tail_samples_per_run must be a non-negative integer.")
        if not isinstance(self.require_tail_iterations_one, bool):
            raise TypeError("EvidencePolicy.require_tail_iterations_one must be a boolean.")
        if self.maximum_cv is not None:
            object.__setattr__(
                self,
                "maximum_cv",
                _validate_non_negative_number(
                    self.maximum_cv,
                    field_name="EvidencePolicy.maximum_cv",
                ),
            )
        if self.maximum_outlier_fraction is not None:
            object.__setattr__(
                self,
                "maximum_outlier_fraction",
                _validate_fraction(
                    self.maximum_outlier_fraction,
                    field_name="EvidencePolicy.maximum_outlier_fraction",
                ),
            )

__post_init__

__post_init__() -> None

Validate evidence thresholds.

Source code in src/benchmatrix/bench_compare.py

def __post_init__(self) -> None:
    """Validate evidence thresholds."""
    if isinstance(self.minimum_runs, bool) or not isinstance(self.minimum_runs, int):
        raise TypeError("EvidencePolicy.minimum_runs must be an integer.")
    if self.minimum_runs <= 0:
        raise ValueError("EvidencePolicy.minimum_runs must be a positive integer.")
    if isinstance(self.minimum_samples_per_run, bool) or not isinstance(self.minimum_samples_per_run, int):
        raise TypeError("EvidencePolicy.minimum_samples_per_run must be an integer.")
    if self.minimum_samples_per_run < 0:
        raise ValueError("EvidencePolicy.minimum_samples_per_run must be a non-negative integer.")
    if isinstance(self.minimum_rounds_per_run, bool) or not isinstance(self.minimum_rounds_per_run, int):
        raise TypeError("EvidencePolicy.minimum_rounds_per_run must be an integer.")
    if self.minimum_rounds_per_run < 0:
        raise ValueError("EvidencePolicy.minimum_rounds_per_run must be a non-negative integer.")
    if not isinstance(self.require_rounds, bool):
        raise TypeError("EvidencePolicy.require_rounds must be a boolean.")
    if not isinstance(self.require_iterations, bool):
        raise TypeError("EvidencePolicy.require_iterations must be a boolean.")
    if not isinstance(self.require_raw_samples_for_inference, bool):
        raise TypeError("EvidencePolicy.require_raw_samples_for_inference must be a boolean.")
    if isinstance(self.minimum_tail_samples_per_run, bool) or not isinstance(
        self.minimum_tail_samples_per_run, int
    ):
        raise TypeError("EvidencePolicy.minimum_tail_samples_per_run must be an integer.")
    if self.minimum_tail_samples_per_run < 0:
        raise ValueError("EvidencePolicy.minimum_tail_samples_per_run must be a non-negative integer.")
    if not isinstance(self.require_tail_iterations_one, bool):
        raise TypeError("EvidencePolicy.require_tail_iterations_one must be a boolean.")
    if self.maximum_cv is not None:
        object.__setattr__(
            self,
            "maximum_cv",
            _validate_non_negative_number(
                self.maximum_cv,
                field_name="EvidencePolicy.maximum_cv",
            ),
        )
    if self.maximum_outlier_fraction is not None:
        object.__setattr__(
            self,
            "maximum_outlier_fraction",
            _validate_fraction(
                self.maximum_outlier_fraction,
                field_name="EvidencePolicy.maximum_outlier_fraction",
            ),
        )

InferencePolicy dataclass

Policy controlling run-level statistical inference.

The default method bootstraps complete process-run statistics, applies a BCa interval, and uses a Bonferroni-adjusted simultaneous confidence level across the reported matrix. legacy_consistency preserves the version 1 observed-pairwise-range decision rule and is intentionally non-inferential.

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class InferencePolicy:
    """Policy controlling run-level statistical inference.

    The default method bootstraps complete process-run statistics, applies a
    BCa interval, and uses a Bonferroni-adjusted simultaneous confidence level
    across the reported matrix. ``legacy_consistency`` preserves the version 1
    observed-pairwise-range decision rule and is intentionally non-inferential.
    """

    method: InferenceMethod = "bca_bootstrap"
    confidence_level: float = 0.95
    resamples: int = 50_000
    random_seed: int = 0
    multiplicity: MultiplicityCorrection = "bonferroni"

    def __post_init__(self) -> None:
        """Validate inference controls."""
        if self.method not in {"bca_bootstrap", "legacy_consistency"}:
            raise ValueError(f"Unsupported inference method: {self.method!r}.")
        confidence_level = _validate_fraction(
            self.confidence_level,
            field_name="InferencePolicy.confidence_level",
        )
        if confidence_level in {0.0, 1.0}:
            raise ValueError("InferencePolicy.confidence_level must be between zero and one.")
        if isinstance(self.resamples, bool) or not isinstance(self.resamples, int):
            raise TypeError("InferencePolicy.resamples must be an integer.")
        if self.resamples < 1_000:
            raise ValueError("InferencePolicy.resamples must be at least 1000.")
        if isinstance(self.random_seed, bool) or not isinstance(self.random_seed, int):
            raise TypeError("InferencePolicy.random_seed must be an integer.")
        if self.random_seed < 0:
            raise ValueError("InferencePolicy.random_seed must be non-negative.")
        if self.multiplicity not in {"bonferroni", "none"}:
            raise ValueError(f"Unsupported multiplicity correction: {self.multiplicity!r}.")
        object.__setattr__(self, "confidence_level", confidence_level)

__post_init__

__post_init__() -> None

Validate inference controls.

Source code in src/benchmatrix/bench_compare.py

def __post_init__(self) -> None:
    """Validate inference controls."""
    if self.method not in {"bca_bootstrap", "legacy_consistency"}:
        raise ValueError(f"Unsupported inference method: {self.method!r}.")
    confidence_level = _validate_fraction(
        self.confidence_level,
        field_name="InferencePolicy.confidence_level",
    )
    if confidence_level in {0.0, 1.0}:
        raise ValueError("InferencePolicy.confidence_level must be between zero and one.")
    if isinstance(self.resamples, bool) or not isinstance(self.resamples, int):
        raise TypeError("InferencePolicy.resamples must be an integer.")
    if self.resamples < 1_000:
        raise ValueError("InferencePolicy.resamples must be at least 1000.")
    if isinstance(self.random_seed, bool) or not isinstance(self.random_seed, int):
        raise TypeError("InferencePolicy.random_seed must be an integer.")
    if self.random_seed < 0:
        raise ValueError("InferencePolicy.random_seed must be non-negative.")
    if self.multiplicity not in {"bonferroni", "none"}:
        raise ValueError(f"Unsupported multiplicity correction: {self.multiplicity!r}.")
    object.__setattr__(self, "confidence_level", confidence_level)

PrecisionPolicy dataclass

Optional fixed-design precision target for paired pilot comparisons.

target_half_width_percent=None disables planning. When enabled, each paired matrix cell estimates the pair count for a fresh future collection; the pilot comparison and its pass/fail decision are never changed by the plan.

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class PrecisionPolicy:
    """Optional fixed-design precision target for paired pilot comparisons.

    ``target_half_width_percent=None`` disables planning. When enabled, each
    paired matrix cell estimates the pair count for a fresh future collection;
    the pilot comparison and its pass/fail decision are never changed by the
    plan.
    """

    target_half_width_percent: float | None = None

    def __post_init__(self) -> None:
        """Validate and normalize the optional percentage target."""
        if self.target_half_width_percent is None:
            return
        target = _validate_non_negative_number(
            self.target_half_width_percent,
            field_name="PrecisionPolicy.target_half_width_percent",
        )
        if target == 0.0:
            raise ValueError("PrecisionPolicy.target_half_width_percent must be positive when enabled.")
        object.__setattr__(self, "target_half_width_percent", target)

    @property
    def enabled(self) -> bool:
        """Return whether paired precision planning is requested."""
        return self.target_half_width_percent is not None

enabled property

enabled: bool

Return whether paired precision planning is requested.

__post_init__

__post_init__() -> None

Validate and normalize the optional percentage target.

Source code in src/benchmatrix/bench_compare.py

def __post_init__(self) -> None:
    """Validate and normalize the optional percentage target."""
    if self.target_half_width_percent is None:
        return
    target = _validate_non_negative_number(
        self.target_half_width_percent,
        field_name="PrecisionPolicy.target_half_width_percent",
    )
    if target == 0.0:
        raise ValueError("PrecisionPolicy.target_half_width_percent must be positive when enabled.")
    object.__setattr__(self, "target_half_width_percent", target)

RegressionPolicy dataclass

Threshold policy for classifying benchmark changes.

Thresholds are percentage points and must be finite and non-negative. More specific mappings override broader ones in this order: exact matrix cell, case, implementation, metric, then the default threshold.

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class RegressionPolicy:
    """Threshold policy for classifying benchmark changes.

    Thresholds are percentage points and must be finite and non-negative. More
    specific mappings override broader ones in this order: exact matrix cell,
    case, implementation, metric, then the default threshold.
    """

    default_threshold_percent: float = 5.0
    by_metric: Mapping[MetricName, float] = field(default_factory=dict)
    by_implementation: Mapping[str, float] = field(default_factory=dict)
    by_case: Mapping[str, float] = field(default_factory=dict)
    by_cell: Mapping[tuple[str, str, MetricName], float] = field(default_factory=dict)

    def __post_init__(self) -> None:
        """Validate thresholds and freeze policy mappings."""
        default_threshold = _validate_threshold(
            self.default_threshold_percent,
            field_name="default_threshold_percent",
        )
        by_metric = {
            _validate_metric_key(metric_name): _validate_threshold(
                threshold,
                field_name=f"by_metric[{metric_name!r}]",
            )
            for metric_name, threshold in self.by_metric.items()
        }
        by_implementation = {
            _validate_selector_name(implementation_name, field_name="implementation"): _validate_threshold(
                threshold,
                field_name=f"by_implementation[{implementation_name!r}]",
            )
            for implementation_name, threshold in self.by_implementation.items()
        }
        by_case = {
            _validate_selector_name(case_name, field_name="case"): _validate_threshold(
                threshold,
                field_name=f"by_case[{case_name!r}]",
            )
            for case_name, threshold in self.by_case.items()
        }
        by_cell = {
            _validate_cell_key(cell): _validate_threshold(
                threshold,
                field_name=f"by_cell[{cell!r}]",
            )
            for cell, threshold in self.by_cell.items()
        }

        object.__setattr__(self, "default_threshold_percent", default_threshold)
        object.__setattr__(self, "by_metric", MappingProxyType(by_metric))
        object.__setattr__(self, "by_implementation", MappingProxyType(by_implementation))
        object.__setattr__(self, "by_case", MappingProxyType(by_case))
        object.__setattr__(self, "by_cell", MappingProxyType(by_cell))

    def threshold_for(
        self,
        implementation_name: str,
        case_name: str,
        metric_name: MetricName,
    ) -> float:
        """Return the effective threshold for one matrix cell."""
        scope = self.threshold_scope_for(implementation_name, case_name, metric_name)
        cell = (implementation_name, case_name, metric_name)
        if scope == "cell":
            return self.by_cell[cell]
        if scope == "case":
            return self.by_case[case_name]
        if scope == "implementation":
            return self.by_implementation[implementation_name]
        if scope == "metric":
            return self.by_metric[metric_name]
        return self.default_threshold_percent

    def threshold_scope_for(
        self,
        implementation_name: str,
        case_name: str,
        metric_name: MetricName,
    ) -> RegressionThresholdScope:
        """Return the selector scope that supplies one cell's threshold."""
        cell = (implementation_name, case_name, metric_name)
        if cell in self.by_cell:
            return "cell"
        if case_name in self.by_case:
            return "case"
        if implementation_name in self.by_implementation:
            return "implementation"
        if metric_name in self.by_metric:
            return "metric"
        return "default"

__post_init__

__post_init__() -> None

Validate thresholds and freeze policy mappings.

Source code in src/benchmatrix/bench_compare.py

def __post_init__(self) -> None:
    """Validate thresholds and freeze policy mappings."""
    default_threshold = _validate_threshold(
        self.default_threshold_percent,
        field_name="default_threshold_percent",
    )
    by_metric = {
        _validate_metric_key(metric_name): _validate_threshold(
            threshold,
            field_name=f"by_metric[{metric_name!r}]",
        )
        for metric_name, threshold in self.by_metric.items()
    }
    by_implementation = {
        _validate_selector_name(implementation_name, field_name="implementation"): _validate_threshold(
            threshold,
            field_name=f"by_implementation[{implementation_name!r}]",
        )
        for implementation_name, threshold in self.by_implementation.items()
    }
    by_case = {
        _validate_selector_name(case_name, field_name="case"): _validate_threshold(
            threshold,
            field_name=f"by_case[{case_name!r}]",
        )
        for case_name, threshold in self.by_case.items()
    }
    by_cell = {
        _validate_cell_key(cell): _validate_threshold(
            threshold,
            field_name=f"by_cell[{cell!r}]",
        )
        for cell, threshold in self.by_cell.items()
    }

    object.__setattr__(self, "default_threshold_percent", default_threshold)
    object.__setattr__(self, "by_metric", MappingProxyType(by_metric))
    object.__setattr__(self, "by_implementation", MappingProxyType(by_implementation))
    object.__setattr__(self, "by_case", MappingProxyType(by_case))
    object.__setattr__(self, "by_cell", MappingProxyType(by_cell))

threshold_for

threshold_for(
    implementation_name: str,
    case_name: str,
    metric_name: MetricName,
) -> float

Return the effective threshold for one matrix cell.

Source code in src/benchmatrix/bench_compare.py

def threshold_for(
    self,
    implementation_name: str,
    case_name: str,
    metric_name: MetricName,
) -> float:
    """Return the effective threshold for one matrix cell."""
    scope = self.threshold_scope_for(implementation_name, case_name, metric_name)
    cell = (implementation_name, case_name, metric_name)
    if scope == "cell":
        return self.by_cell[cell]
    if scope == "case":
        return self.by_case[case_name]
    if scope == "implementation":
        return self.by_implementation[implementation_name]
    if scope == "metric":
        return self.by_metric[metric_name]
    return self.default_threshold_percent

threshold_scope_for

threshold_scope_for(
    implementation_name: str,
    case_name: str,
    metric_name: MetricName,
) -> RegressionThresholdScope

Return the selector scope that supplies one cell's threshold.

Source code in src/benchmatrix/bench_compare.py

def threshold_scope_for(
    self,
    implementation_name: str,
    case_name: str,
    metric_name: MetricName,
) -> RegressionThresholdScope:
    """Return the selector scope that supplies one cell's threshold."""
    cell = (implementation_name, case_name, metric_name)
    if cell in self.by_cell:
        return "cell"
    if case_name in self.by_case:
        return "case"
    if implementation_name in self.by_implementation:
        return "implementation"
    if metric_name in self.by_metric:
        return "metric"
    return "default"

RunCompatibilityFinding dataclass

One material difference between two run environments.

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class RunCompatibilityFinding:
    """One material difference between two run environments."""

    field: str
    baseline_value: object | None
    candidate_value: object | None
    severity: CompatibilitySeverity
    reason: str
    baseline_run: str | None = None
    candidate_run: str | None = None

RunCompatibilityPolicy dataclass

Policy controlling run-environment compatibility checks.

permissive keeps lower-risk differences as warnings, strict promotes every difference or missing environment record to a blocker, and off disables run-level compatibility checks.

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class RunCompatibilityPolicy:
    """Policy controlling run-environment compatibility checks.

    ``permissive`` keeps lower-risk differences as warnings, ``strict``
    promotes every difference or missing environment record to a blocker, and
    ``off`` disables run-level compatibility checks.
    """

    mode: CompatibilityMode = "permissive"

    def __post_init__(self) -> None:
        """Validate the compatibility mode."""
        if self.mode not in {"strict", "permissive", "off"}:
            raise ValueError(f"Unsupported run compatibility mode: {self.mode!r}.")

__post_init__

__post_init__() -> None

Validate the compatibility mode.

Source code in src/benchmatrix/bench_compare.py

def __post_init__(self) -> None:
    """Validate the compatibility mode."""
    if self.mode not in {"strict", "permissive", "off"}:
        raise ValueError(f"Unsupported run compatibility mode: {self.mode!r}.")

RunCompatibilityReport dataclass

Compatibility findings for the baseline and candidate environments.

Source code in src/benchmatrix/bench_compare.py

@dataclass(frozen=True, slots=True)
class RunCompatibilityReport:
    """Compatibility findings for the baseline and candidate environments."""

    policy: RunCompatibilityPolicy
    findings: tuple[RunCompatibilityFinding, ...]
    pairs_checked: int = 1

    @property
    def blocking(self) -> tuple[RunCompatibilityFinding, ...]:
        """Return differences that prevent a trustworthy comparison."""
        return tuple(finding for finding in self.findings if finding.severity == "blocking")

    @property
    def warnings(self) -> tuple[RunCompatibilityFinding, ...]:
        """Return non-blocking environment differences."""
        return tuple(finding for finding in self.findings if finding.severity == "warning")

    @property
    def is_compatible(self) -> bool:
        """Return whether no blocking environment differences were found."""
        return not self.blocking

blocking property

blocking: tuple[RunCompatibilityFinding, ...]

Return differences that prevent a trustworthy comparison.

warnings property

warnings: tuple[RunCompatibilityFinding, ...]

Return non-blocking environment differences.

is_compatible property

is_compatible: bool

Return whether no blocking environment differences were found.

BenchmarkCase dataclass

Named input case and metadata for a pytest-benchmark matrix.

Warning

If fresh_inputs is false, pytest-benchmark may call the target function repeatedly with the same argument objects. That is appropriate only when the target function treats its inputs as immutable or when reuse reflects the workload you want to measure.

If fresh_inputs is true, this harness uses pytest-benchmark pedantic setup so input construction is setup work rather than timed target-function work. That avoids accidentally timing input creation, but it also means the benchmark is not an end-to-end measurement that includes input construction. To benchmark construction cost, put that construction inside the target function itself.

When fresh_inputs is true, BenchmarkConfig.pedantic_iterations is ignored because pytest-benchmark setup mode is used. The harness emits a runtime warning when a non-default value is ignored.

Parameters:

Name	Type	Description	Default
name	str	Human-readable case name used in parameter IDs and metadata.	required
make_args	Callable[[], tuple[object, ...]]	Factory returning positional arguments for the target function.	_empty_args
make_kwargs	Callable[[], dict[str, object]]	Factory returning keyword arguments for the target function.	_empty_kwargs
work_units	float | Callable[[], float] | None	Positive logical amount of work performed by one target call. This can represent items, rows, bytes, tokens, records, events, or any other domain-specific unit.	None
work_unit_name	str	Name of the logical work unit, such as "items", "rows", "bytes", or "tokens". Use a base unit name without spaces, slashes, or "/s"; display code appends "/s" for throughput.	_DEFAULT_WORK_UNIT_NAME
fresh_inputs	bool	Whether each benchmark round needs newly created inputs.	False
metadata	Mapping[str, object]	Additional strict-JSON-renderable metadata describing the case. Reasonable scalar types such as paths, datetimes, enums, and NumPy scalars are coerced; unsupported values raise MetadataSerializationError.	_empty_metadata()

Attributes:

Name	Type	Description
name	str	Human-readable case name used in parameter IDs and metadata.
make_args	Callable[[], tuple[object, ...]]	Factory returning positional arguments for the target function.
make_kwargs	Callable[[], dict[str, object]]	Factory returning keyword arguments for the target function.
work_units	float | Callable[[], float] | None	Positive logical amount of work performed by one target call.
work_unit_name	str	Name of the logical work unit.
fresh_inputs	bool	Whether each benchmark round needs newly created inputs.
metadata	Mapping[str, object]	Strict JSON-safe metadata describing the case.

Source code in src/benchmatrix/bench_harness.py

@dataclass(frozen=True, slots=True)
class BenchmarkCase:
    """Named input case and metadata for a pytest-benchmark matrix.

    Warning:
        If ``fresh_inputs`` is false, pytest-benchmark may call the target
        function repeatedly with the same argument objects. That is appropriate
        only when the target function treats its inputs as immutable or when
        reuse reflects the workload you want to measure.

        If ``fresh_inputs`` is true, this harness uses pytest-benchmark
        pedantic setup so input construction is setup work rather than timed
        target-function work. That avoids accidentally timing input creation,
        but it also means the benchmark is not an end-to-end measurement that
        includes input construction. To benchmark construction cost, put that
        construction inside the target function itself.

        When ``fresh_inputs`` is true, ``BenchmarkConfig.pedantic_iterations``
        is ignored because pytest-benchmark setup mode is used. The harness
        emits a runtime warning when a non-default value is ignored.

    Args:
        name: Human-readable case name used in parameter IDs and metadata.
        make_args: Factory returning positional arguments for the target
            function.
        make_kwargs: Factory returning keyword arguments for the target
            function.
        work_units: Positive logical amount of work performed by one target
            call. This can represent items, rows, bytes, tokens, records,
            events, or any other domain-specific unit.
        work_unit_name: Name of the logical work unit, such as ``"items"``,
            ``"rows"``, ``"bytes"``, or ``"tokens"``. Use a base unit name
            without spaces, slashes, or ``"/s"``; display code appends ``"/s"``
            for throughput.
        fresh_inputs: Whether each benchmark round needs newly created inputs.
        metadata: Additional strict-JSON-renderable metadata describing the
            case. Reasonable scalar types such as paths, datetimes, enums, and
            NumPy scalars are coerced; unsupported values raise
            ``MetadataSerializationError``.

    Attributes:
        name: Human-readable case name used in parameter IDs and metadata.
        make_args: Factory returning positional arguments for the target
            function.
        make_kwargs: Factory returning keyword arguments for the target
            function.
        work_units: Positive logical amount of work performed by one target
            call.
        work_unit_name: Name of the logical work unit.
        fresh_inputs: Whether each benchmark round needs newly created inputs.
        metadata: Strict JSON-safe metadata describing the case.
    """

    name: str
    make_args: Callable[[], tuple[object, ...]] = _empty_args
    make_kwargs: Callable[[], dict[str, object]] = _empty_kwargs
    work_units: float | Callable[[], float] | None = None
    work_unit_name: str = _DEFAULT_WORK_UNIT_NAME
    fresh_inputs: bool = False
    metadata: Mapping[str, object] = field(default_factory=_empty_metadata)

    def __post_init__(self) -> None:
        """Validate benchmark case fields after initialization."""
        object.__setattr__(self, "name", _validate_name(self.name, field="case name"))

        _validate_case_callable(self.make_args, field="make_args")
        _validate_case_callable(self.make_kwargs, field="make_kwargs")

        object.__setattr__(self, "work_unit_name", _validate_work_unit_name(self.work_unit_name))

        if self.work_units is not None:
            if callable(self.work_units):
                _validate_case_callable(self.work_units, field="work_units")
            else:
                _ = _validate_work_units(self.work_units)

        if not isinstance(self.fresh_inputs, bool):
            raise TypeError("BenchmarkCase.fresh_inputs must be a boolean.")

        if not isinstance(self.metadata, Mapping):
            raise TypeError("BenchmarkCase.metadata must be a mapping.")

        coerced_metadata = _coerce_json_mapping(
            self.metadata,
            path="BenchmarkCase.metadata",
        )
        reserved_metadata = sorted(_RESERVED_CASE_METADATA_KEYS.intersection(coerced_metadata))
        if reserved_metadata:
            formatted = ", ".join(repr(key) for key in reserved_metadata)
            raise ValueError(f"BenchmarkCase.metadata uses reserved key(s): {formatted}.")
        object.__setattr__(self, "metadata", coerced_metadata)

    def make_call(self) -> tuple[tuple[object, ...], dict[str, object]]:
        """Return positional and keyword arguments for one target invocation.

        Returns:
            A tuple containing positional arguments and keyword arguments.
        """
        args = self.make_args()
        if not isinstance(args, tuple):
            raise TypeError("BenchmarkCase.make_args must return a tuple.")

        kwargs = self.make_kwargs()
        if not isinstance(kwargs, dict):
            raise TypeError("BenchmarkCase.make_kwargs must return a dictionary.")
        if any(not isinstance(key, str) for key in kwargs):
            raise TypeError("BenchmarkCase.make_kwargs must return a dictionary with string keys.")

        return args, kwargs

    def work_unit_count(self) -> float | None:
        """Return the logical work-unit count for throughput metrics.

        Returns:
            The logical work-unit count, or ``None`` when the case has no work
            unit count.

        Raises:
            ValueError: If the work-unit count is not positive or finite.
        """
        if self.work_units is None:
            return None

        value = self.work_units() if callable(self.work_units) else self.work_units
        return _validate_work_units(value)

    @classmethod
    def from_values(
        cls,
        name: str,
        *args: object,
        work_units: float | Callable[[], float] | None = None,
        work_unit_name: str = _DEFAULT_WORK_UNIT_NAME,
        fresh_inputs: bool = False,
        copier: Callable[[object], object] | None = None,
        metadata: Mapping[str, object] | None = None,
        **kwargs: object,
    ) -> BenchmarkCase:
        """Create a benchmark case from concrete argument values.

        Args:
            name: Case name.
            *args: Positional arguments for the target function.
            work_units: Positive logical amount of work performed by one target
                call.
            work_unit_name: Name of the logical work unit, such as ``"items"``,
                ``"rows"``, ``"bytes"``, or ``"tokens"``. Use a base unit name
                without spaces, slashes, or ``"/s"``.
            fresh_inputs: Whether target invocations need fresh inputs. When
                true and ``copier`` is omitted, a shallow copy is made for each
                argument value.
            copier: Optional copy function applied to each argument value. Use
                ``deep_copy`` or a domain-specific copy function when shallow
                copies are not fresh enough for the benchmarked workload.
            metadata: Optional strict-JSON-renderable case metadata.
            **kwargs: Keyword arguments for the target function.

        Returns:
            A configured benchmark case.
        """

        if not isinstance(fresh_inputs, bool):
            raise TypeError("BenchmarkCase.fresh_inputs must be a boolean.")
        if copier is not None:
            _validate_case_callable(copier, field="copier")

        effective_copier = shallow_copy if fresh_inputs and copier is None else copier

        def make_args() -> tuple[object, ...]:
            """Return case positional arguments."""
            if effective_copier is None:
                return args

            return tuple(effective_copier(arg) for arg in args)

        def make_kwargs() -> dict[str, object]:
            """Return case keyword arguments."""
            if effective_copier is None:
                return dict(kwargs)

            return {key: effective_copier(value) for key, value in kwargs.items()}

        return cls(
            name=name,
            make_args=make_args,
            make_kwargs=make_kwargs,
            work_units=work_units,
            work_unit_name=work_unit_name,
            fresh_inputs=fresh_inputs or effective_copier is not None,
            metadata={} if metadata is None else dict(metadata),
        )

__post_init__

__post_init__() -> None

Validate benchmark case fields after initialization.

Source code in src/benchmatrix/bench_harness.py

def __post_init__(self) -> None:
    """Validate benchmark case fields after initialization."""
    object.__setattr__(self, "name", _validate_name(self.name, field="case name"))

    _validate_case_callable(self.make_args, field="make_args")
    _validate_case_callable(self.make_kwargs, field="make_kwargs")

    object.__setattr__(self, "work_unit_name", _validate_work_unit_name(self.work_unit_name))

    if self.work_units is not None:
        if callable(self.work_units):
            _validate_case_callable(self.work_units, field="work_units")
        else:
            _ = _validate_work_units(self.work_units)

    if not isinstance(self.fresh_inputs, bool):
        raise TypeError("BenchmarkCase.fresh_inputs must be a boolean.")

    if not isinstance(self.metadata, Mapping):
        raise TypeError("BenchmarkCase.metadata must be a mapping.")

    coerced_metadata = _coerce_json_mapping(
        self.metadata,
        path="BenchmarkCase.metadata",
    )
    reserved_metadata = sorted(_RESERVED_CASE_METADATA_KEYS.intersection(coerced_metadata))
    if reserved_metadata:
        formatted = ", ".join(repr(key) for key in reserved_metadata)
        raise ValueError(f"BenchmarkCase.metadata uses reserved key(s): {formatted}.")
    object.__setattr__(self, "metadata", coerced_metadata)

make_call

make_call() -> tuple[tuple[object, ...], dict[str, object]]

Return positional and keyword arguments for one target invocation.

Returns:

Type	Description
tuple[tuple[object, ...], dict[str, object]]	A tuple containing positional arguments and keyword arguments.

Source code in src/benchmatrix/bench_harness.py

def make_call(self) -> tuple[tuple[object, ...], dict[str, object]]:
    """Return positional and keyword arguments for one target invocation.

    Returns:
        A tuple containing positional arguments and keyword arguments.
    """
    args = self.make_args()
    if not isinstance(args, tuple):
        raise TypeError("BenchmarkCase.make_args must return a tuple.")

    kwargs = self.make_kwargs()
    if not isinstance(kwargs, dict):
        raise TypeError("BenchmarkCase.make_kwargs must return a dictionary.")
    if any(not isinstance(key, str) for key in kwargs):
        raise TypeError("BenchmarkCase.make_kwargs must return a dictionary with string keys.")

    return args, kwargs

work_unit_count

work_unit_count() -> float | None

Return the logical work-unit count for throughput metrics.

Returns:

Type	Description
float | None	The logical work-unit count, or None when the case has no work
float | None	unit count.

Raises:

Type	Description
ValueError	If the work-unit count is not positive or finite.

Source code in src/benchmatrix/bench_harness.py

def work_unit_count(self) -> float | None:
    """Return the logical work-unit count for throughput metrics.

    Returns:
        The logical work-unit count, or ``None`` when the case has no work
        unit count.

    Raises:
        ValueError: If the work-unit count is not positive or finite.
    """
    if self.work_units is None:
        return None

    value = self.work_units() if callable(self.work_units) else self.work_units
    return _validate_work_units(value)

from_values classmethod

from_values(
    name: str,
    *args: object,
    work_units: float | Callable[[], float] | None = None,
    work_unit_name: str = _DEFAULT_WORK_UNIT_NAME,
    fresh_inputs: bool = False,
    copier: Callable[[object], object] | None = None,
    metadata: Mapping[str, object] | None = None,
    **kwargs: object,
) -> BenchmarkCase

Create a benchmark case from concrete argument values.

Parameters:

Name	Type	Description	Default
name	str	Case name.	required
*args	object	Positional arguments for the target function.	()
work_units	float | Callable[[], float] | None	Positive logical amount of work performed by one target call.	None
work_unit_name	str	Name of the logical work unit, such as "items", "rows", "bytes", or "tokens". Use a base unit name without spaces, slashes, or "/s".	_DEFAULT_WORK_UNIT_NAME
fresh_inputs	bool	Whether target invocations need fresh inputs. When true and copier is omitted, a shallow copy is made for each argument value.	False
copier	Callable[[object], object] | None	Optional copy function applied to each argument value. Use deep_copy or a domain-specific copy function when shallow copies are not fresh enough for the benchmarked workload.	None
metadata	Mapping[str, object] | None	Optional strict-JSON-renderable case metadata.	None
**kwargs	object	Keyword arguments for the target function.	{}

Returns:

Type	Description
BenchmarkCase	A configured benchmark case.

Source code in src/benchmatrix/bench_harness.py

@classmethod
def from_values(
    cls,
    name: str,
    *args: object,
    work_units: float | Callable[[], float] | None = None,
    work_unit_name: str = _DEFAULT_WORK_UNIT_NAME,
    fresh_inputs: bool = False,
    copier: Callable[[object], object] | None = None,
    metadata: Mapping[str, object] | None = None,
    **kwargs: object,
) -> BenchmarkCase:
    """Create a benchmark case from concrete argument values.

    Args:
        name: Case name.
        *args: Positional arguments for the target function.
        work_units: Positive logical amount of work performed by one target
            call.
        work_unit_name: Name of the logical work unit, such as ``"items"``,
            ``"rows"``, ``"bytes"``, or ``"tokens"``. Use a base unit name
            without spaces, slashes, or ``"/s"``.
        fresh_inputs: Whether target invocations need fresh inputs. When
            true and ``copier`` is omitted, a shallow copy is made for each
            argument value.
        copier: Optional copy function applied to each argument value. Use
            ``deep_copy`` or a domain-specific copy function when shallow
            copies are not fresh enough for the benchmarked workload.
        metadata: Optional strict-JSON-renderable case metadata.
        **kwargs: Keyword arguments for the target function.

    Returns:
        A configured benchmark case.
    """

    if not isinstance(fresh_inputs, bool):
        raise TypeError("BenchmarkCase.fresh_inputs must be a boolean.")
    if copier is not None:
        _validate_case_callable(copier, field="copier")

    effective_copier = shallow_copy if fresh_inputs and copier is None else copier

    def make_args() -> tuple[object, ...]:
        """Return case positional arguments."""
        if effective_copier is None:
            return args

        return tuple(effective_copier(arg) for arg in args)

    def make_kwargs() -> dict[str, object]:
        """Return case keyword arguments."""
        if effective_copier is None:
            return dict(kwargs)

        return {key: effective_copier(value) for key, value in kwargs.items()}

    return cls(
        name=name,
        make_args=make_args,
        make_kwargs=make_kwargs,
        work_units=work_units,
        work_unit_name=work_unit_name,
        fresh_inputs=fresh_inputs or effective_copier is not None,
        metadata={} if metadata is None else dict(metadata),
    )

BenchmarkConfig dataclass

Configuration passed from benchmatrix to pytest-benchmark.

Parameters:

Name	Type	Description	Default
pedantic_rounds	int	Number of pedantic benchmark rounds to request.	_DEFAULT_PEDANTIC_ROUNDS
warmup_rounds	int	Number of pedantic warmup rounds to request.	_DEFAULT_WARMUP_ROUNDS
pedantic_iterations	int	Number of function calls per pedantic round when inputs are reused. This value is intentionally ignored when BenchmarkCase.fresh_inputs is true because pytest-benchmark setup mode is used to keep input construction outside the timed target-function body.	_DEFAULT_PEDANTIC_ITERATIONS
stream_progress	bool	Whether benchmark helpers should print one progress line per benchmark invocation.	True
before_benchmark	BenchmarkLifecycleHook | None	Optional synchronous hook called immediately before pytest-benchmark starts an invocation.	None
validate_result	BenchmarkResultValidator | None	Optional synchronous correctness hook called with the result returned by pytest-benchmark. The hook should raise when the result is invalid.	None
after_benchmark	BenchmarkLifecycleHook | None	Optional synchronous hook called after result validation, or during cleanup if benchmarking or validation raises.	None

Attributes:

Name	Type	Description
pedantic_rounds	int	Number of pedantic benchmark rounds to request.
warmup_rounds	int	Number of pedantic warmup rounds to request.
pedantic_iterations	int	Number of function calls per pedantic round when inputs are reused.
stream_progress	bool	Whether benchmark helpers should print one progress line per benchmark invocation.
before_benchmark	BenchmarkLifecycleHook | None	Optional untimed setup hook for a benchmark invocation.
validate_result	BenchmarkResultValidator | None	Optional untimed correctness hook for the returned target result.
after_benchmark	BenchmarkLifecycleHook | None	Optional untimed cleanup hook for a benchmark invocation.

Raises:

Type	Description
TypeError	If a timing control has the wrong type, progress output is not boolean, or a configured hook is not callable or is asynchronous.
ValueError	If rounds or iterations are not positive, or if warmup rounds are negative.

Warning

For tail_latency benchmarks, setting pedantic_iterations above one means raw samples are per-round averages of multiple calls rather than individual-call latency samples. The harness emits a runtime warning for this configuration.

Source code in src/benchmatrix/bench_harness.py

@dataclass(frozen=True, slots=True)
class BenchmarkConfig:
    """Configuration passed from benchmatrix to pytest-benchmark.

    Args:
        pedantic_rounds: Number of pedantic benchmark rounds to request.
        warmup_rounds: Number of pedantic warmup rounds to request.
        pedantic_iterations: Number of function calls per pedantic round when
            inputs are reused. This value is intentionally ignored when
            ``BenchmarkCase.fresh_inputs`` is true because pytest-benchmark
            setup mode is used to keep input construction outside the timed
            target-function body.
        stream_progress: Whether benchmark helpers should print one progress
            line per benchmark invocation.
        before_benchmark: Optional synchronous hook called immediately before
            pytest-benchmark starts an invocation.
        validate_result: Optional synchronous correctness hook called with the
            result returned by pytest-benchmark. The hook should raise when the
            result is invalid.
        after_benchmark: Optional synchronous hook called after result
            validation, or during cleanup if benchmarking or validation raises.

    Attributes:
        pedantic_rounds: Number of pedantic benchmark rounds to request.
        warmup_rounds: Number of pedantic warmup rounds to request.
        pedantic_iterations: Number of function calls per pedantic round when
            inputs are reused.
        stream_progress: Whether benchmark helpers should print one progress
            line per benchmark invocation.
        before_benchmark: Optional untimed setup hook for a benchmark
            invocation.
        validate_result: Optional untimed correctness hook for the returned
            target result.
        after_benchmark: Optional untimed cleanup hook for a benchmark
            invocation.

    Raises:
        TypeError: If a timing control has the wrong type, progress output is
            not boolean, or a configured hook is not callable or is
            asynchronous.
        ValueError: If rounds or iterations are not positive, or if warmup
            rounds are negative.

    Warning:
        For ``tail_latency`` benchmarks, setting ``pedantic_iterations`` above
        one means raw samples are per-round averages of multiple calls rather
        than individual-call latency samples. The harness emits a runtime
        warning for this configuration.
    """

    pedantic_rounds: int = _DEFAULT_PEDANTIC_ROUNDS
    warmup_rounds: int = _DEFAULT_WARMUP_ROUNDS
    pedantic_iterations: int = _DEFAULT_PEDANTIC_ITERATIONS
    stream_progress: bool = True
    before_benchmark: BenchmarkLifecycleHook | None = None
    validate_result: BenchmarkResultValidator | None = None
    after_benchmark: BenchmarkLifecycleHook | None = None

    def __post_init__(self) -> None:
        """Validate benchmark configuration after initialization."""
        if isinstance(self.pedantic_rounds, bool) or not isinstance(self.pedantic_rounds, int):
            raise TypeError("BenchmarkConfig.pedantic_rounds must be an integer.")
        if self.pedantic_rounds <= 0:
            raise ValueError("BenchmarkConfig.pedantic_rounds must be positive.")

        if isinstance(self.warmup_rounds, bool) or not isinstance(self.warmup_rounds, int):
            raise TypeError("BenchmarkConfig.warmup_rounds must be an integer.")
        if self.warmup_rounds < 0:
            raise ValueError("BenchmarkConfig.warmup_rounds must be non-negative.")

        if isinstance(self.pedantic_iterations, bool) or not isinstance(self.pedantic_iterations, int):
            raise TypeError("BenchmarkConfig.pedantic_iterations must be an integer.")
        if self.pedantic_iterations <= 0:
            raise ValueError("BenchmarkConfig.pedantic_iterations must be positive.")

        if not isinstance(self.stream_progress, bool):
            raise TypeError("BenchmarkConfig.stream_progress must be a boolean.")

        _validate_hook(self.before_benchmark, field="before_benchmark")
        _validate_hook(self.validate_result, field="validate_result")
        _validate_hook(self.after_benchmark, field="after_benchmark")

__post_init__

__post_init__() -> None

Validate benchmark configuration after initialization.

Source code in src/benchmatrix/bench_harness.py

def __post_init__(self) -> None:
    """Validate benchmark configuration after initialization."""
    if isinstance(self.pedantic_rounds, bool) or not isinstance(self.pedantic_rounds, int):
        raise TypeError("BenchmarkConfig.pedantic_rounds must be an integer.")
    if self.pedantic_rounds <= 0:
        raise ValueError("BenchmarkConfig.pedantic_rounds must be positive.")

    if isinstance(self.warmup_rounds, bool) or not isinstance(self.warmup_rounds, int):
        raise TypeError("BenchmarkConfig.warmup_rounds must be an integer.")
    if self.warmup_rounds < 0:
        raise ValueError("BenchmarkConfig.warmup_rounds must be non-negative.")

    if isinstance(self.pedantic_iterations, bool) or not isinstance(self.pedantic_iterations, int):
        raise TypeError("BenchmarkConfig.pedantic_iterations must be an integer.")
    if self.pedantic_iterations <= 0:
        raise ValueError("BenchmarkConfig.pedantic_iterations must be positive.")

    if not isinstance(self.stream_progress, bool):
        raise TypeError("BenchmarkConfig.stream_progress must be a boolean.")

    _validate_hook(self.before_benchmark, field="before_benchmark")
    _validate_hook(self.validate_result, field="validate_result")
    _validate_hook(self.after_benchmark, field="after_benchmark")

BenchmarkFixture

Bases: Protocol

pytest-benchmark fixture surface used by benchmatrix.

Attributes:

Name	Type	Description
extra_info	MutableMapping[str, object]	Mutable metadata attached to pytest-benchmark output.

Source code in src/benchmatrix/bench_harness.py

class BenchmarkFixture(Protocol):
    """pytest-benchmark fixture surface used by benchmatrix.

    Attributes:
        extra_info: Mutable metadata attached to pytest-benchmark output.
    """

    extra_info: MutableMapping[str, object]

    def __call__(self, target: Callable[..., T], *args: object, **kwargs: object) -> T:
        """Benchmark ``target`` with pytest-benchmark automatic calibration."""
        ...

    def pedantic(
        self,
        target: Callable[..., T],
        *,
        args: Sequence[object] | None = None,
        kwargs: Mapping[str, object] | None = None,
        setup: Callable[[], tuple[Sequence[object], Mapping[str, object]]] | None = None,
        teardown: Callable[..., object] | None = None,
        rounds: int = _DEFAULT_PEDANTIC_ROUNDS,
        warmup_rounds: int = _DEFAULT_WARMUP_ROUNDS,
        iterations: int = _DEFAULT_PEDANTIC_ITERATIONS,
    ) -> T:
        """Benchmark ``target`` with pytest-benchmark pedantic mode."""
        ...

__call__

__call__(
    target: Callable[..., T],
    *args: object,
    **kwargs: object,
) -> T

Benchmark target with pytest-benchmark automatic calibration.

Source code in src/benchmatrix/bench_harness.py

def __call__(self, target: Callable[..., T], *args: object, **kwargs: object) -> T:
    """Benchmark ``target`` with pytest-benchmark automatic calibration."""
    ...

pedantic

pedantic(
    target: Callable[..., T],
    *,
    args: Sequence[object] | None = None,
    kwargs: Mapping[str, object] | None = None,
    setup: Callable[
        [], tuple[Sequence[object], Mapping[str, object]]
    ]
    | None = None,
    teardown: Callable[..., object] | None = None,
    rounds: int = _DEFAULT_PEDANTIC_ROUNDS,
    warmup_rounds: int = _DEFAULT_WARMUP_ROUNDS,
    iterations: int = _DEFAULT_PEDANTIC_ITERATIONS,
) -> T

Benchmark target with pytest-benchmark pedantic mode.

Source code in src/benchmatrix/bench_harness.py

def pedantic(
    self,
    target: Callable[..., T],
    *,
    args: Sequence[object] | None = None,
    kwargs: Mapping[str, object] | None = None,
    setup: Callable[[], tuple[Sequence[object], Mapping[str, object]]] | None = None,
    teardown: Callable[..., object] | None = None,
    rounds: int = _DEFAULT_PEDANTIC_ROUNDS,
    warmup_rounds: int = _DEFAULT_WARMUP_ROUNDS,
    iterations: int = _DEFAULT_PEDANTIC_ITERATIONS,
) -> T:
    """Benchmark ``target`` with pytest-benchmark pedantic mode."""
    ...

BenchmarkHookContext dataclass

Identity and inputs available to benchmark lifecycle hooks.

Attributes:

Name	Type	Description
metric_name	MetricName	Metric requested for this benchmark invocation.
implementation_name	str	Name of the implementation under test.
case_name	str	Matrix case name under test.
function	TargetFunction	Synchronous target function under test.
case	BenchmarkCase	Benchmark case definition used by the invocation.

Source code in src/benchmatrix/bench_harness.py

@dataclass(frozen=True, slots=True)
class BenchmarkHookContext:
    """Identity and inputs available to benchmark lifecycle hooks.

    Attributes:
        metric_name: Metric requested for this benchmark invocation.
        implementation_name: Name of the implementation under test.
        case_name: Matrix case name under test.
        function: Synchronous target function under test.
        case: Benchmark case definition used by the invocation.
    """

    metric_name: MetricName
    implementation_name: str
    case_name: str
    function: TargetFunction
    case: BenchmarkCase

BenchmarkInvocationRecord dataclass

Lightweight record returned after one benchmark invocation.

This record is not a timing result. Timing results come from pytest-benchmark's report, saved runs, CSV output, or JSON output.

Attributes:

Name	Type	Description
metric_name	MetricName	Metric requested for this benchmark invocation.
implementation_name	str	Name of the implementation under test.
case_name	str	Name of the input case under test.
extra_info	Mapping[str, object]	Strict JSON-safe metadata attached to pytest-benchmark output. Values are limited to JSON primitives, lists, and string-keyed mappings after metadata coercion. The metadata includes benchmatrix producer and schema-version markers.

Source code in src/benchmatrix/bench_harness.py

@dataclass(frozen=True, slots=True)
class BenchmarkInvocationRecord:
    """Lightweight record returned after one benchmark invocation.

    This record is not a timing result. Timing results come from
    pytest-benchmark's report, saved runs, CSV output, or JSON output.

    Attributes:
        metric_name: Metric requested for this benchmark invocation.
        implementation_name: Name of the implementation under test.
        case_name: Name of the input case under test.
        extra_info: Strict JSON-safe metadata attached to pytest-benchmark
            output. Values are limited to JSON primitives, lists, and
            string-keyed mappings after metadata coercion. The metadata includes
            benchmatrix producer and schema-version markers.
    """

    metric_name: MetricName
    implementation_name: str
    case_name: str
    extra_info: Mapping[str, object]

BenchmarkPolicyConfig dataclass

Resolved benchmatrix policy configuration.

Attributes:

Name	Type	Description
compatibility	RunCompatibilityPolicy	Run-environment compatibility policy.
evidence	EvidencePolicy	Repeated-run evidence policy.
inference	InferencePolicy	Run-level inference and multiplicity policy.
precision	PrecisionPolicy	Optional fixed-design precision-planning policy.
regression	RegressionPolicy	Regression threshold policy.
source	Path | None	Selected TOML file, or None when using built-in defaults.
configured_fields	frozenset[str]	Explicit tool.benchmatrix field paths.

Source code in src/benchmatrix/bench_policy.py

@dataclass(frozen=True, slots=True)
class BenchmarkPolicyConfig:
    """Resolved benchmatrix policy configuration.

    Attributes:
        compatibility: Run-environment compatibility policy.
        evidence: Repeated-run evidence policy.
        inference: Run-level inference and multiplicity policy.
        precision: Optional fixed-design precision-planning policy.
        regression: Regression threshold policy.
        source: Selected TOML file, or ``None`` when using built-in defaults.
        configured_fields: Explicit ``tool.benchmatrix`` field paths.
    """

    compatibility: RunCompatibilityPolicy
    evidence: EvidencePolicy
    regression: RegressionPolicy
    source: Path | None = None
    configured_fields: frozenset[str] = frozenset()
    inference: InferencePolicy = dataclass_field(default_factory=InferencePolicy)
    precision: PrecisionPolicy = dataclass_field(default_factory=PrecisionPolicy)

    def __post_init__(self) -> None:
        """Normalize the optional source path and configured field set."""
        if not isinstance(self.compatibility, RunCompatibilityPolicy):
            raise TypeError("BenchmarkPolicyConfig.compatibility must be a RunCompatibilityPolicy.")
        if not isinstance(self.evidence, EvidencePolicy):
            raise TypeError("BenchmarkPolicyConfig.evidence must be an EvidencePolicy.")
        if not isinstance(self.inference, InferencePolicy):
            raise TypeError("BenchmarkPolicyConfig.inference must be an InferencePolicy.")
        if not isinstance(self.precision, PrecisionPolicy):
            raise TypeError("BenchmarkPolicyConfig.precision must be a PrecisionPolicy.")
        if not isinstance(self.regression, RegressionPolicy):
            raise TypeError("BenchmarkPolicyConfig.regression must be a RegressionPolicy.")
        fields = frozenset(self.configured_fields)
        if any(not isinstance(field, str) or not field for field in fields):
            raise ValueError("BenchmarkPolicyConfig.configured_fields must contain non-empty strings.")
        if self.source is not None:
            object.__setattr__(self, "source", Path(self.source))
        object.__setattr__(self, "configured_fields", fields)

    @property
    def is_configured(self) -> bool:
        """Return whether a ``tool.benchmatrix`` table was loaded."""
        return self.source is not None

is_configured property

is_configured: bool

Return whether a tool.benchmatrix table was loaded.

__post_init__

__post_init__() -> None

Normalize the optional source path and configured field set.

Source code in src/benchmatrix/bench_policy.py

def __post_init__(self) -> None:
    """Normalize the optional source path and configured field set."""
    if not isinstance(self.compatibility, RunCompatibilityPolicy):
        raise TypeError("BenchmarkPolicyConfig.compatibility must be a RunCompatibilityPolicy.")
    if not isinstance(self.evidence, EvidencePolicy):
        raise TypeError("BenchmarkPolicyConfig.evidence must be an EvidencePolicy.")
    if not isinstance(self.inference, InferencePolicy):
        raise TypeError("BenchmarkPolicyConfig.inference must be an InferencePolicy.")
    if not isinstance(self.precision, PrecisionPolicy):
        raise TypeError("BenchmarkPolicyConfig.precision must be a PrecisionPolicy.")
    if not isinstance(self.regression, RegressionPolicy):
        raise TypeError("BenchmarkPolicyConfig.regression must be a RegressionPolicy.")
    fields = frozenset(self.configured_fields)
    if any(not isinstance(field, str) or not field for field in fields):
        raise ValueError("BenchmarkPolicyConfig.configured_fields must contain non-empty strings.")
    if self.source is not None:
        object.__setattr__(self, "source", Path(self.source))
    object.__setattr__(self, "configured_fields", fields)

BenchmarkCollectionSnapshot dataclass

Portable collection provenance embedded in a comparison report.

Source code in src/benchmatrix/bench_report.py

@dataclass(frozen=True, slots=True)
class BenchmarkCollectionSnapshot:
    """Portable collection provenance embedded in a comparison report."""

    manifest: str
    created_at: str
    command: tuple[str, ...]
    cwd: str
    commit: str | None
    environment_fingerprint: str | None
    requested_runs: int
    expected_cells: tuple[BenchmarkCell, ...]
    records: tuple[BenchmarkRunRecord, ...]

    def __post_init__(self) -> None:
        """Validate and normalize the portable collection snapshot."""
        if not isinstance(self.manifest, str) or not self.manifest:
            raise ValueError("Collection manifest must be a non-empty string.")
        if not isinstance(self.created_at, str) or not self.created_at:
            raise ValueError("Collection created_at must be a non-empty string.")
        command = tuple(self.command)
        if not command or any(not isinstance(argument, str) or not argument for argument in command):
            raise ValueError("Collection command must contain non-empty strings.")
        if not isinstance(self.cwd, str) or not self.cwd:
            raise ValueError("Collection cwd must be a non-empty string.")
        if self.commit is not None and (not isinstance(self.commit, str) or not self.commit):
            raise ValueError("Collection commit must be a non-empty string or None.")
        if self.environment_fingerprint is not None and (
            not isinstance(self.environment_fingerprint, str) or not self.environment_fingerprint
        ):
            raise ValueError("Collection environment fingerprint must be a non-empty string or None.")
        if (
            isinstance(self.requested_runs, bool)
            or not isinstance(self.requested_runs, int)
            or self.requested_runs <= 0
        ):
            raise ValueError("Collection requested_runs must be a positive integer.")
        expected_cells = tuple(self.expected_cells)
        records = tuple(self.records)
        if tuple(record.index for record in records) != tuple(range(1, len(records) + 1)):
            raise ValueError("Collection record indexes must be contiguous and one-based.")
        successful_runs = sum(record.status == "succeeded" for record in records)
        if successful_runs > self.requested_runs:
            raise ValueError("Collection has more successful runs than requested.")
        if len(set(expected_cells)) != len(expected_cells):
            raise ValueError("Collection expected_cells must not contain duplicates.")
        for implementation, case, metric in expected_cells:
            if not implementation or not case or metric not in KNOWN_METRICS:
                raise ValueError(f"Invalid collection expected cell: {(implementation, case, metric)!r}.")
        object.__setattr__(self, "command", command)
        object.__setattr__(self, "expected_cells", expected_cells)
        object.__setattr__(self, "records", records)

    @classmethod
    def from_group(cls, group: BenchmarkRunGroup) -> BenchmarkCollectionSnapshot:
        """Create a portable snapshot from a loaded run group."""
        return cls(
            manifest=str(group.manifest_path),
            created_at=group.created_at,
            command=group.command,
            cwd=str(group.cwd),
            commit=group.commit,
            environment_fingerprint=group.environment_fingerprint,
            requested_runs=group.requested_runs,
            expected_cells=group.expected_cells,
            records=group.records,
        )

    @property
    def attempted_runs(self) -> int:
        """Return the number of completed collection attempts."""
        return len(self.records)

    @property
    def successful_runs(self) -> int:
        """Return the number of accepted collection attempts."""
        return sum(record.status == "succeeded" for record in self.records)

    @property
    def failed_runs(self) -> int:
        """Return the number of failed collection attempts."""
        return sum(record.status == "failed" for record in self.records)

    @property
    def pending_runs(self) -> int:
        """Return initial collection slots that were never attempted."""
        return max(0, self.requested_runs - self.attempted_runs)

    @property
    def retry_attempts(self) -> int:
        """Return attempts appended after the initial collection slots."""
        return max(0, self.attempted_runs - self.requested_runs)

    @property
    def remaining_runs(self) -> int:
        """Return additional successful runs needed for completeness."""
        return self.requested_runs - self.successful_runs

    @property
    def complete(self) -> bool:
        """Return whether the requested successful-run target was reached."""
        return self.successful_runs == self.requested_runs

    def to_dict(self) -> dict[str, JsonValue]:
        """Return the stable JSON representation of this collection."""
        return {
            "manifest": self.manifest,
            "created_at": self.created_at,
            "command": list(self.command),
            "cwd": self.cwd,
            "commit": self.commit,
            "environment_fingerprint": self.environment_fingerprint,
            "requested_runs": self.requested_runs,
            "attempted_runs": self.attempted_runs,
            "successful_runs": self.successful_runs,
            "failed_runs": self.failed_runs,
            "complete": self.complete,
            "expected_cells": [
                {
                    "implementation_name": implementation,
                    "case_name": case,
                    "metric_name": metric,
                }
                for implementation, case, metric in self.expected_cells
            ],
            "runs": [_record_dict(record) for record in self.records],
        }

attempted_runs property

attempted_runs: int

Return the number of completed collection attempts.

successful_runs property

successful_runs: int

Return the number of accepted collection attempts.

failed_runs property

failed_runs: int

Return the number of failed collection attempts.

pending_runs property

pending_runs: int

Return initial collection slots that were never attempted.

retry_attempts property

retry_attempts: int

Return attempts appended after the initial collection slots.

remaining_runs property

remaining_runs: int

Return additional successful runs needed for completeness.

complete property

complete: bool

Return whether the requested successful-run target was reached.

__post_init__

__post_init__() -> None

Validate and normalize the portable collection snapshot.

Source code in src/benchmatrix/bench_report.py

def __post_init__(self) -> None:
    """Validate and normalize the portable collection snapshot."""
    if not isinstance(self.manifest, str) or not self.manifest:
        raise ValueError("Collection manifest must be a non-empty string.")
    if not isinstance(self.created_at, str) or not self.created_at:
        raise ValueError("Collection created_at must be a non-empty string.")
    command = tuple(self.command)
    if not command or any(not isinstance(argument, str) or not argument for argument in command):
        raise ValueError("Collection command must contain non-empty strings.")
    if not isinstance(self.cwd, str) or not self.cwd:
        raise ValueError("Collection cwd must be a non-empty string.")
    if self.commit is not None and (not isinstance(self.commit, str) or not self.commit):
        raise ValueError("Collection commit must be a non-empty string or None.")
    if self.environment_fingerprint is not None and (
        not isinstance(self.environment_fingerprint, str) or not self.environment_fingerprint
    ):
        raise ValueError("Collection environment fingerprint must be a non-empty string or None.")
    if (
        isinstance(self.requested_runs, bool)
        or not isinstance(self.requested_runs, int)
        or self.requested_runs <= 0
    ):
        raise ValueError("Collection requested_runs must be a positive integer.")
    expected_cells = tuple(self.expected_cells)
    records = tuple(self.records)
    if tuple(record.index for record in records) != tuple(range(1, len(records) + 1)):
        raise ValueError("Collection record indexes must be contiguous and one-based.")
    successful_runs = sum(record.status == "succeeded" for record in records)
    if successful_runs > self.requested_runs:
        raise ValueError("Collection has more successful runs than requested.")
    if len(set(expected_cells)) != len(expected_cells):
        raise ValueError("Collection expected_cells must not contain duplicates.")
    for implementation, case, metric in expected_cells:
        if not implementation or not case or metric not in KNOWN_METRICS:
            raise ValueError(f"Invalid collection expected cell: {(implementation, case, metric)!r}.")
    object.__setattr__(self, "command", command)
    object.__setattr__(self, "expected_cells", expected_cells)
    object.__setattr__(self, "records", records)

from_group classmethod

from_group(
    group: BenchmarkRunGroup,
) -> BenchmarkCollectionSnapshot

Create a portable snapshot from a loaded run group.

Source code in src/benchmatrix/bench_report.py

@classmethod
def from_group(cls, group: BenchmarkRunGroup) -> BenchmarkCollectionSnapshot:
    """Create a portable snapshot from a loaded run group."""
    return cls(
        manifest=str(group.manifest_path),
        created_at=group.created_at,
        command=group.command,
        cwd=str(group.cwd),
        commit=group.commit,
        environment_fingerprint=group.environment_fingerprint,
        requested_runs=group.requested_runs,
        expected_cells=group.expected_cells,
        records=group.records,
    )

to_dict

to_dict() -> dict[str, JsonValue]

Return the stable JSON representation of this collection.

Source code in src/benchmatrix/bench_report.py

def to_dict(self) -> dict[str, JsonValue]:
    """Return the stable JSON representation of this collection."""
    return {
        "manifest": self.manifest,
        "created_at": self.created_at,
        "command": list(self.command),
        "cwd": self.cwd,
        "commit": self.commit,
        "environment_fingerprint": self.environment_fingerprint,
        "requested_runs": self.requested_runs,
        "attempted_runs": self.attempted_runs,
        "successful_runs": self.successful_runs,
        "failed_runs": self.failed_runs,
        "complete": self.complete,
        "expected_cells": [
            {
                "implementation_name": implementation,
                "case_name": case,
                "metric_name": metric,
            }
            for implementation, case, metric in self.expected_cells
        ],
        "runs": [_record_dict(record) for record in self.records],
    }

BenchmarkComparisonReport dataclass

One portable, versioned benchmark comparison result.

Source code in src/benchmatrix/bench_report.py

@dataclass(frozen=True, slots=True)
class BenchmarkComparisonReport:
    """One portable, versioned benchmark comparison result."""

    baselines: tuple[str, ...]
    candidates: tuple[str, ...]
    baseline_collections: tuple[BenchmarkCollectionSnapshot, ...]
    candidate_collections: tuple[BenchmarkCollectionSnapshot, ...]
    compatibility: RunCompatibilityReport
    evidence_policy: EvidencePolicy
    regression_policy: RegressionPolicy
    policy_provenance: BenchmarkPolicyProvenance
    comparisons: tuple[BenchmarkComparison, ...]
    threshold_provenance: tuple[BenchmarkThresholdProvenance, ...]
    inference_policy: InferencePolicy = field(
        default_factory=lambda: InferencePolicy(method="legacy_consistency", multiplicity="none")
    )
    design: ComparisonDesign = "independent"
    precision_policy: PrecisionPolicy = field(default_factory=PrecisionPolicy)
    paired_collections: tuple[BenchmarkPairedCollectionSnapshot, ...] = ()

    def __post_init__(self) -> None:
        """Validate and normalize report containers."""
        baselines = tuple(self.baselines)
        candidates = tuple(self.candidates)
        baseline_collections = tuple(self.baseline_collections)
        candidate_collections = tuple(self.candidate_collections)
        paired_collections = tuple(self.paired_collections)
        comparisons = tuple(self.comparisons)
        threshold_provenance = tuple(self.threshold_provenance)
        if not baselines or any(not isinstance(source, str) or not source for source in baselines):
            raise ValueError("Report baselines must contain non-empty source strings.")
        if not candidates or any(not isinstance(source, str) or not source for source in candidates):
            raise ValueError("Report candidates must contain non-empty source strings.")
        if len(comparisons) != len(threshold_provenance):
            raise ValueError("Every reported comparison requires threshold provenance.")
        if any(
            not isinstance(item, BenchmarkCollectionSnapshot)
            for item in (*baseline_collections, *candidate_collections)
        ):
            raise TypeError("Report collections must contain BenchmarkCollectionSnapshot values.")
        if any(not isinstance(item, BenchmarkPairedCollectionSnapshot) for item in paired_collections):
            raise TypeError("Report paired_collections must contain BenchmarkPairedCollectionSnapshot values.")
        if any(not isinstance(item, BenchmarkComparison) for item in comparisons):
            raise TypeError("Report comparisons must contain BenchmarkComparison values.")
        if any(not isinstance(item, BenchmarkThresholdProvenance) for item in threshold_provenance):
            raise TypeError("Report threshold_provenance must contain BenchmarkThresholdProvenance values.")
        if not isinstance(self.compatibility, RunCompatibilityReport):
            raise TypeError("Report compatibility must be a RunCompatibilityReport.")
        if not isinstance(self.evidence_policy, EvidencePolicy):
            raise TypeError("Report evidence_policy must be an EvidencePolicy.")
        if not isinstance(self.regression_policy, RegressionPolicy):
            raise TypeError("Report regression_policy must be a RegressionPolicy.")
        if not isinstance(self.inference_policy, InferencePolicy):
            raise TypeError("Report inference_policy must be an InferencePolicy.")
        if self.design not in {"independent", "paired"}:
            raise ValueError(f"Unsupported report comparison design: {self.design!r}.")
        if not isinstance(self.precision_policy, PrecisionPolicy):
            raise TypeError("Report precision_policy must be a PrecisionPolicy.")
        if not isinstance(self.policy_provenance, BenchmarkPolicyProvenance):
            raise TypeError("Report policy_provenance must be BenchmarkPolicyProvenance.")
        paired_strata_count: int | None = None
        paired_supercycle_multiple: int | None = None
        if self.design == "independent":
            if self.precision_policy.enabled:
                raise ValueError("Independent reports cannot enable paired precision planning.")
            if paired_collections:
                raise ValueError("Independent reports cannot include paired collection provenance.")
        else:
            if baseline_collections or candidate_collections:
                raise ValueError("Paired reports cannot include independent collection provenance.")
            if len(baselines) != len(candidates):
                raise ValueError("Paired reports require equal baseline and candidate source counts.")
            evidence_paths = tuple(Path(source).resolve() for source in (*baselines, *candidates))
            if len(set(evidence_paths)) != len(evidence_paths):
                raise ValueError("Paired report complete-pair sources cannot reuse a file as independent evidence.")
            if paired_collections:
                supercycle_multiples = {
                    balanced_order_supercycle_length(len(collection.expected_cells))
                    for collection in paired_collections
                }
                if len(supercycle_multiples) != 1:
                    raise ValueError("Paired collection provenance has conflicting joint-design supercycles.")
                paired_supercycle_multiple = next(iter(supercycle_multiples))
                paired_strata_count = len(
                    {
                        baseline.pair_order
                        for collection in paired_collections
                        for baseline, _candidate in collection.complete_pair_records
                    }
                )
                expected_baselines = tuple(
                    source for collection in paired_collections for source in collection.baseline_sources
                )
                expected_candidates = tuple(
                    source for collection in paired_collections for source in collection.candidate_sources
                )
                if baselines != expected_baselines or candidates != expected_candidates:
                    raise ValueError("Paired report sources must match complete same-block record paths in pair order.")
        object.__setattr__(self, "baselines", baselines)
        object.__setattr__(self, "candidates", candidates)
        object.__setattr__(self, "baseline_collections", baseline_collections)
        object.__setattr__(self, "candidate_collections", candidate_collections)
        object.__setattr__(self, "paired_collections", paired_collections)
        object.__setattr__(self, "comparisons", comparisons)
        object.__setattr__(self, "threshold_provenance", threshold_provenance)
        for cell, threshold in zip(comparisons, threshold_provenance, strict=True):
            _validate_threshold_provenance(
                cell,
                threshold,
                regression_policy=self.regression_policy,
                policy_provenance=self.policy_provenance,
            )
            _validate_cell_inference(
                cell,
                inference_policy=self.inference_policy,
                environment_compatible=self.compatibility.is_compatible,
                design=self.design,
                expected_pair_count=(len(baselines) if self.design == "paired" else None),
                expected_strata_count=paired_strata_count,
            )
            _validate_cell_precision(
                cell,
                precision_policy=self.precision_policy,
                inference_policy=self.inference_policy,
                evidence_policy=self.evidence_policy,
                design=self.design,
                expected_pair_count=(len(baselines) if self.design == "paired" else None),
                expected_strata_count=paired_strata_count,
                expected_pair_count_multiple=paired_supercycle_multiple,
            )
        inferences = tuple(cell.inference for cell in comparisons if cell.inference is not None)
        if inferences:
            family_sizes = {inference.family_size for inference in inferences}
            if len(family_sizes) != 1 or next(iter(family_sizes)) != len(inferences):
                raise ValueError("Reported inference family size is inconsistent with inferred cells.")
        plans = tuple(cell.precision for cell in comparisons if cell.precision is not None)
        if plans:
            family_sizes = {plan.family_size for plan in plans}
            if len(family_sizes) != 1:
                raise ValueError("Reported precision-plan family sizes are inconsistent across cells.")
            if self.compatibility.is_compatible and next(iter(family_sizes)) != len(plans):
                raise ValueError("Reported precision-plan family size is inconsistent with planned cells.")

    @classmethod
    def from_comparison(
        cls,
        comparison: BenchmarkRunComparison,
        *,
        baselines: Sequence[str | Path],
        candidates: Sequence[str | Path],
        policy_provenance: BenchmarkPolicyProvenance,
        threshold_provenance: Sequence[BenchmarkThresholdProvenance],
        baseline_collections: Sequence[BenchmarkRunGroup] = (),
        candidate_collections: Sequence[BenchmarkRunGroup] = (),
        paired_collections: Sequence[BenchmarkPairedRunGroup] = (),
    ) -> BenchmarkComparisonReport:
        """Create a portable report from a live comparison result."""
        return cls(
            baselines=tuple(str(source) for source in baselines),
            candidates=tuple(str(source) for source in candidates),
            baseline_collections=tuple(BenchmarkCollectionSnapshot.from_group(group) for group in baseline_collections),
            candidate_collections=tuple(
                BenchmarkCollectionSnapshot.from_group(group) for group in candidate_collections
            ),
            paired_collections=tuple(
                BenchmarkPairedCollectionSnapshot.from_group(group) for group in paired_collections
            ),
            compatibility=comparison.compatibility,
            evidence_policy=comparison.evidence_policy,
            regression_policy=comparison.regression_policy,
            policy_provenance=policy_provenance,
            comparisons=comparison.comparisons,
            threshold_provenance=tuple(threshold_provenance),
            inference_policy=comparison.inference_policy,
            design=comparison.design,
            precision_policy=comparison.precision_policy,
        )

    @property
    def producer(self) -> str:
        """Return the comparison report producer identifier."""
        return PRODUCER

    @property
    def kind(self) -> str:
        """Return the comparison report document kind."""
        return COMPARISON_REPORT_KIND

    @property
    def schema_version(self) -> int:
        """Return the comparison report schema version."""
        return COMPARISON_REPORT_SCHEMA_VERSION

    @property
    def baseline(self) -> str:
        """Return the primary baseline source."""
        return self.baselines[0]

    @property
    def candidate(self) -> str:
        """Return the primary candidate source."""
        return self.candidates[0]

    @property
    def improved(self) -> tuple[BenchmarkComparison, ...]:
        """Return cells classified as improvements."""
        return tuple(cell for cell in self.comparisons if cell.regression == "improved")

    @property
    def unchanged(self) -> tuple[BenchmarkComparison, ...]:
        """Return cells classified as unchanged."""
        return tuple(cell for cell in self.comparisons if cell.regression == "unchanged")

    @property
    def regressed(self) -> tuple[BenchmarkComparison, ...]:
        """Return cells classified as regressions."""
        return tuple(cell for cell in self.comparisons if cell.regression == "regressed")

    @property
    def inconclusive(self) -> tuple[BenchmarkComparison, ...]:
        """Return cells with inconclusive evidence."""
        return tuple(cell for cell in self.comparisons if cell.regression == "inconclusive")

    @property
    def not_comparable(self) -> tuple[BenchmarkComparison, ...]:
        """Return cells without a trustworthy comparison."""
        return tuple(cell for cell in self.comparisons if cell.regression == "not_comparable")

    @property
    def comparison_passed(self) -> bool:
        """Return whether the benchmark comparison itself passed."""
        complete = all(cell.status == "matched" for cell in self.comparisons)
        comparable = self.compatibility.is_compatible and complete and not self.not_comparable
        return comparable and not self.regressed and not self.inconclusive

    @property
    def passed(self) -> bool:
        """Return whether comparison and collection lifecycle gates passed."""
        collections = (*self.baseline_collections, *self.candidate_collections, *self.paired_collections)
        return self.comparison_passed and all(collection.complete for collection in collections)

    @property
    def is_comparable(self) -> bool:
        """Return whether all report cells and environments are comparable."""
        complete = all(cell.status == "matched" for cell in self.comparisons)
        return self.compatibility.is_compatible and complete and not self.not_comparable

    @property
    def has_regressions(self) -> bool:
        """Return whether any matrix cell regressed."""
        return bool(self.regressed)

    def to_dict(self) -> dict[str, JsonValue]:
        """Return the complete stable comparison report document."""
        return {
            "producer": self.producer,
            "kind": self.kind,
            "schema_version": self.schema_version,
            "baseline": self.baseline,
            "candidate": self.candidate,
            "baselines": list(self.baselines),
            "candidates": list(self.candidates),
            "baseline_collections": [collection.to_dict() for collection in self.baseline_collections],
            "candidate_collections": [collection.to_dict() for collection in self.candidate_collections],
            "paired_collections": [collection.to_dict() for collection in self.paired_collections],
            "design": self.design,
            "passed": self.passed,
            "comparison_passed": self.comparison_passed,
            "is_comparable": self.is_comparable,
            "has_regressions": self.has_regressions,
            "compatibility": _compatibility_dict(self.compatibility),
            "summary": {
                "improved": len(self.improved),
                "unchanged": len(self.unchanged),
                "regressed": len(self.regressed),
                "inconclusive": len(self.inconclusive),
                "not_comparable": len(self.not_comparable),
            },
            "evidence_policy": _evidence_policy_dict(self.evidence_policy),
            "inference_policy": _inference_policy_dict(self.inference_policy),
            "precision_policy": _precision_policy_dict(self.precision_policy),
            "policy": _policy_dict(self.policy_provenance, self.regression_policy),
            "comparisons": [
                _comparison_dict(cell, threshold)
                for cell, threshold in zip(
                    self.comparisons,
                    self.threshold_provenance,
                    strict=True,
                )
            ],
        }

producer property

producer: str

Return the comparison report producer identifier.

kind property

kind: str

Return the comparison report document kind.

schema_version property

schema_version: int

Return the comparison report schema version.

baseline property

baseline: str

Return the primary baseline source.

candidate property

candidate: str

Return the primary candidate source.

improved property

improved: tuple[BenchmarkComparison, ...]

Return cells classified as improvements.

unchanged property

unchanged: tuple[BenchmarkComparison, ...]

Return cells classified as unchanged.

regressed property

regressed: tuple[BenchmarkComparison, ...]

Return cells classified as regressions.

inconclusive property

inconclusive: tuple[BenchmarkComparison, ...]

Return cells with inconclusive evidence.

not_comparable property

not_comparable: tuple[BenchmarkComparison, ...]

Return cells without a trustworthy comparison.

comparison_passed property

comparison_passed: bool

Return whether the benchmark comparison itself passed.

passed property

passed: bool

Return whether comparison and collection lifecycle gates passed.

is_comparable property

is_comparable: bool

Return whether all report cells and environments are comparable.

has_regressions property

has_regressions: bool

Return whether any matrix cell regressed.

__post_init__

__post_init__() -> None

Validate and normalize report containers.

Source code in src/benchmatrix/bench_report.py

def __post_init__(self) -> None:
    """Validate and normalize report containers."""
    baselines = tuple(self.baselines)
    candidates = tuple(self.candidates)
    baseline_collections = tuple(self.baseline_collections)
    candidate_collections = tuple(self.candidate_collections)
    paired_collections = tuple(self.paired_collections)
    comparisons = tuple(self.comparisons)
    threshold_provenance = tuple(self.threshold_provenance)
    if not baselines or any(not isinstance(source, str) or not source for source in baselines):
        raise ValueError("Report baselines must contain non-empty source strings.")
    if not candidates or any(not isinstance(source, str) or not source for source in candidates):
        raise ValueError("Report candidates must contain non-empty source strings.")
    if len(comparisons) != len(threshold_provenance):
        raise ValueError("Every reported comparison requires threshold provenance.")
    if any(
        not isinstance(item, BenchmarkCollectionSnapshot)
        for item in (*baseline_collections, *candidate_collections)
    ):
        raise TypeError("Report collections must contain BenchmarkCollectionSnapshot values.")
    if any(not isinstance(item, BenchmarkPairedCollectionSnapshot) for item in paired_collections):
        raise TypeError("Report paired_collections must contain BenchmarkPairedCollectionSnapshot values.")
    if any(not isinstance(item, BenchmarkComparison) for item in comparisons):
        raise TypeError("Report comparisons must contain BenchmarkComparison values.")
    if any(not isinstance(item, BenchmarkThresholdProvenance) for item in threshold_provenance):
        raise TypeError("Report threshold_provenance must contain BenchmarkThresholdProvenance values.")
    if not isinstance(self.compatibility, RunCompatibilityReport):
        raise TypeError("Report compatibility must be a RunCompatibilityReport.")
    if not isinstance(self.evidence_policy, EvidencePolicy):
        raise TypeError("Report evidence_policy must be an EvidencePolicy.")
    if not isinstance(self.regression_policy, RegressionPolicy):
        raise TypeError("Report regression_policy must be a RegressionPolicy.")
    if not isinstance(self.inference_policy, InferencePolicy):
        raise TypeError("Report inference_policy must be an InferencePolicy.")
    if self.design not in {"independent", "paired"}:
        raise ValueError(f"Unsupported report comparison design: {self.design!r}.")
    if not isinstance(self.precision_policy, PrecisionPolicy):
        raise TypeError("Report precision_policy must be a PrecisionPolicy.")
    if not isinstance(self.policy_provenance, BenchmarkPolicyProvenance):
        raise TypeError("Report policy_provenance must be BenchmarkPolicyProvenance.")
    paired_strata_count: int | None = None
    paired_supercycle_multiple: int | None = None
    if self.design == "independent":
        if self.precision_policy.enabled:
            raise ValueError("Independent reports cannot enable paired precision planning.")
        if paired_collections:
            raise ValueError("Independent reports cannot include paired collection provenance.")
    else:
        if baseline_collections or candidate_collections:
            raise ValueError("Paired reports cannot include independent collection provenance.")
        if len(baselines) != len(candidates):
            raise ValueError("Paired reports require equal baseline and candidate source counts.")
        evidence_paths = tuple(Path(source).resolve() for source in (*baselines, *candidates))
        if len(set(evidence_paths)) != len(evidence_paths):
            raise ValueError("Paired report complete-pair sources cannot reuse a file as independent evidence.")
        if paired_collections:
            supercycle_multiples = {
                balanced_order_supercycle_length(len(collection.expected_cells))
                for collection in paired_collections
            }
            if len(supercycle_multiples) != 1:
                raise ValueError("Paired collection provenance has conflicting joint-design supercycles.")
            paired_supercycle_multiple = next(iter(supercycle_multiples))
            paired_strata_count = len(
                {
                    baseline.pair_order
                    for collection in paired_collections
                    for baseline, _candidate in collection.complete_pair_records
                }
            )
            expected_baselines = tuple(
                source for collection in paired_collections for source in collection.baseline_sources
            )
            expected_candidates = tuple(
                source for collection in paired_collections for source in collection.candidate_sources
            )
            if baselines != expected_baselines or candidates != expected_candidates:
                raise ValueError("Paired report sources must match complete same-block record paths in pair order.")
    object.__setattr__(self, "baselines", baselines)
    object.__setattr__(self, "candidates", candidates)
    object.__setattr__(self, "baseline_collections", baseline_collections)
    object.__setattr__(self, "candidate_collections", candidate_collections)
    object.__setattr__(self, "paired_collections", paired_collections)
    object.__setattr__(self, "comparisons", comparisons)
    object.__setattr__(self, "threshold_provenance", threshold_provenance)
    for cell, threshold in zip(comparisons, threshold_provenance, strict=True):
        _validate_threshold_provenance(
            cell,
            threshold,
            regression_policy=self.regression_policy,
            policy_provenance=self.policy_provenance,
        )
        _validate_cell_inference(
            cell,
            inference_policy=self.inference_policy,
            environment_compatible=self.compatibility.is_compatible,
            design=self.design,
            expected_pair_count=(len(baselines) if self.design == "paired" else None),
            expected_strata_count=paired_strata_count,
        )
        _validate_cell_precision(
            cell,
            precision_policy=self.precision_policy,
            inference_policy=self.inference_policy,
            evidence_policy=self.evidence_policy,
            design=self.design,
            expected_pair_count=(len(baselines) if self.design == "paired" else None),
            expected_strata_count=paired_strata_count,
            expected_pair_count_multiple=paired_supercycle_multiple,
        )
    inferences = tuple(cell.inference for cell in comparisons if cell.inference is not None)
    if inferences:
        family_sizes = {inference.family_size for inference in inferences}
        if len(family_sizes) != 1 or next(iter(family_sizes)) != len(inferences):
            raise ValueError("Reported inference family size is inconsistent with inferred cells.")
    plans = tuple(cell.precision for cell in comparisons if cell.precision is not None)
    if plans:
        family_sizes = {plan.family_size for plan in plans}
        if len(family_sizes) != 1:
            raise ValueError("Reported precision-plan family sizes are inconsistent across cells.")
        if self.compatibility.is_compatible and next(iter(family_sizes)) != len(plans):
            raise ValueError("Reported precision-plan family size is inconsistent with planned cells.")

from_comparison classmethod

from_comparison(
    comparison: BenchmarkRunComparison,
    *,
    baselines: Sequence[str | Path],
    candidates: Sequence[str | Path],
    policy_provenance: BenchmarkPolicyProvenance,
    threshold_provenance: Sequence[
        BenchmarkThresholdProvenance
    ],
    baseline_collections: Sequence[BenchmarkRunGroup] = (),
    candidate_collections: Sequence[BenchmarkRunGroup] = (),
    paired_collections: Sequence[
        BenchmarkPairedRunGroup
    ] = (),
) -> BenchmarkComparisonReport

Create a portable report from a live comparison result.

Source code in src/benchmatrix/bench_report.py

@classmethod
def from_comparison(
    cls,
    comparison: BenchmarkRunComparison,
    *,
    baselines: Sequence[str | Path],
    candidates: Sequence[str | Path],
    policy_provenance: BenchmarkPolicyProvenance,
    threshold_provenance: Sequence[BenchmarkThresholdProvenance],
    baseline_collections: Sequence[BenchmarkRunGroup] = (),
    candidate_collections: Sequence[BenchmarkRunGroup] = (),
    paired_collections: Sequence[BenchmarkPairedRunGroup] = (),
) -> BenchmarkComparisonReport:
    """Create a portable report from a live comparison result."""
    return cls(
        baselines=tuple(str(source) for source in baselines),
        candidates=tuple(str(source) for source in candidates),
        baseline_collections=tuple(BenchmarkCollectionSnapshot.from_group(group) for group in baseline_collections),
        candidate_collections=tuple(
            BenchmarkCollectionSnapshot.from_group(group) for group in candidate_collections
        ),
        paired_collections=tuple(
            BenchmarkPairedCollectionSnapshot.from_group(group) for group in paired_collections
        ),
        compatibility=comparison.compatibility,
        evidence_policy=comparison.evidence_policy,
        regression_policy=comparison.regression_policy,
        policy_provenance=policy_provenance,
        comparisons=comparison.comparisons,
        threshold_provenance=tuple(threshold_provenance),
        inference_policy=comparison.inference_policy,
        design=comparison.design,
        precision_policy=comparison.precision_policy,
    )

to_dict

to_dict() -> dict[str, JsonValue]

Return the complete stable comparison report document.

Source code in src/benchmatrix/bench_report.py

def to_dict(self) -> dict[str, JsonValue]:
    """Return the complete stable comparison report document."""
    return {
        "producer": self.producer,
        "kind": self.kind,
        "schema_version": self.schema_version,
        "baseline": self.baseline,
        "candidate": self.candidate,
        "baselines": list(self.baselines),
        "candidates": list(self.candidates),
        "baseline_collections": [collection.to_dict() for collection in self.baseline_collections],
        "candidate_collections": [collection.to_dict() for collection in self.candidate_collections],
        "paired_collections": [collection.to_dict() for collection in self.paired_collections],
        "design": self.design,
        "passed": self.passed,
        "comparison_passed": self.comparison_passed,
        "is_comparable": self.is_comparable,
        "has_regressions": self.has_regressions,
        "compatibility": _compatibility_dict(self.compatibility),
        "summary": {
            "improved": len(self.improved),
            "unchanged": len(self.unchanged),
            "regressed": len(self.regressed),
            "inconclusive": len(self.inconclusive),
            "not_comparable": len(self.not_comparable),
        },
        "evidence_policy": _evidence_policy_dict(self.evidence_policy),
        "inference_policy": _inference_policy_dict(self.inference_policy),
        "precision_policy": _precision_policy_dict(self.precision_policy),
        "policy": _policy_dict(self.policy_provenance, self.regression_policy),
        "comparisons": [
            _comparison_dict(cell, threshold)
            for cell, threshold in zip(
                self.comparisons,
                self.threshold_provenance,
                strict=True,
            )
        ],
    }

BenchmarkPairedCollectionSnapshot dataclass

Portable paired AB/BA collection provenance embedded in a report.

Source code in src/benchmatrix/bench_report.py

@dataclass(frozen=True, slots=True)
class BenchmarkPairedCollectionSnapshot:
    """Portable paired AB/BA collection provenance embedded in a report."""

    manifest: str
    created_at: str
    baseline_command: tuple[str, ...]
    candidate_command: tuple[str, ...]
    baseline_cwd: str
    candidate_cwd: str
    baseline_commit: str | None
    candidate_commit: str | None
    baseline_environment_fingerprint: str | None
    candidate_environment_fingerprint: str | None
    requested_pairs: int
    random_seed: int
    expected_cells: tuple[BenchmarkCell, ...]
    records: tuple[BenchmarkPairedRunRecord, ...]
    automatic_pairs: bool = False

    def __post_init__(self) -> None:
        """Validate and normalize the portable paired snapshot."""
        if not isinstance(self.manifest, str) or not self.manifest:
            raise ValueError("Paired collection manifest must be a non-empty string.")
        if not isinstance(self.created_at, str) or not self.created_at:
            raise ValueError("Paired collection created_at must be a non-empty string.")
        baseline_command = tuple(self.baseline_command)
        candidate_command = tuple(self.candidate_command)
        for name, command in (
            ("baseline_command", baseline_command),
            ("candidate_command", candidate_command),
        ):
            if not command or any(not isinstance(argument, str) or not argument for argument in command):
                raise ValueError(f"Paired collection {name} must contain non-empty strings.")
        for name, cwd in (("baseline_cwd", self.baseline_cwd), ("candidate_cwd", self.candidate_cwd)):
            if not isinstance(cwd, str) or not cwd:
                raise ValueError(f"Paired collection {name} must be a non-empty string.")
        for name, value in (
            ("baseline_commit", self.baseline_commit),
            ("candidate_commit", self.candidate_commit),
            ("baseline_environment_fingerprint", self.baseline_environment_fingerprint),
            ("candidate_environment_fingerprint", self.candidate_environment_fingerprint),
        ):
            if value is not None and (not isinstance(value, str) or not value):
                raise ValueError(f"Paired collection {name} must be a non-empty string or None.")
        if (
            isinstance(self.requested_pairs, bool)
            or not isinstance(self.requested_pairs, int)
            or self.requested_pairs <= 0
        ):
            raise ValueError("Paired collection requested_pairs must be a positive integer.")
        if isinstance(self.random_seed, bool) or not isinstance(self.random_seed, int) or self.random_seed < 0:
            raise ValueError("Paired collection random_seed must be a non-negative integer.")
        if not isinstance(self.automatic_pairs, bool):
            raise TypeError("Paired collection automatic_pairs must be a boolean.")

        expected_cells = tuple(self.expected_cells)
        records = tuple(self.records)
        if len(set(expected_cells)) != len(expected_cells):
            raise ValueError("Paired collection expected_cells must not contain duplicates.")
        for implementation, case, metric in expected_cells:
            if not implementation or not case or metric not in KNOWN_METRICS:
                raise ValueError(f"Invalid paired collection expected cell: {(implementation, case, metric)!r}.")
        if any(not isinstance(record, BenchmarkPairedRunRecord) for record in records):
            raise TypeError("Paired collection records must contain BenchmarkPairedRunRecord values.")
        if tuple(record.index for record in records) != tuple(range(1, len(records) + 1)):
            raise ValueError("Paired collection record indexes must be contiguous and one-based.")
        canonical_paths = tuple(record.path.resolve() for record in records)
        if len(set(canonical_paths)) != len(canonical_paths):
            raise ValueError("Paired collection record paths must be unique.")

        successful_records = tuple(record for record in records if record.status == "succeeded")
        if successful_records and not expected_cells:
            raise ValueError("Paired collection with successful records requires expected cells.")
        if expected_cells and not successful_records:
            raise ValueError("Paired collection without successful records cannot define expected cells.")
        if self.automatic_pairs:
            if expected_cells:
                supercycle = balanced_order_supercycle_length(len(expected_cells))
                minimum_pairs = EvidencePolicy().minimum_runs
                expected_pair_target = ((minimum_pairs + supercycle - 1) // supercycle) * supercycle
                if self.requested_pairs != expected_pair_target:
                    raise ValueError(
                        "Automatic paired collection target must be the smallest complete joint-design "
                        "supercycle at or above the evidence minimum."
                    )
            elif self.requested_pairs != 6:
                raise ValueError("Automatic paired collection without successful records must retain target 6.")

        schedule = {
            entry.pair_index: entry
            for entry in make_paired_ab_ba_schedule(
                self.requested_pairs,
                random_seed=self.random_seed,
                cell_count=(len(expected_cells) if expected_cells else None),
            )
        }
        block_variants: dict[tuple[int, int], set[str]] = {}
        attempts_by_pair: dict[int, set[int]] = {}
        latest_attempt_by_pair: dict[int, int] = {}
        first_seen_pairs: set[int] = set()
        matrix_anchor_seen = False
        for record in records:
            if record.pair_index > self.requested_pairs:
                raise ValueError("Paired collection record pair_index exceeds requested_pairs.")
            if record.pair_index not in first_seen_pairs:
                expected_pair_index = len(first_seen_pairs) + 1
                if record.pair_index != expected_pair_index:
                    raise ValueError("Paired collection target pairs must first appear as a one-based prefix.")
                first_seen_pairs.add(record.pair_index)
            if record.pair_index > 1 and not matrix_anchor_seen:
                raise ValueError("Paired collection cannot start later target pairs before a successful matrix exists.")
            expected = schedule[record.pair_index]
            if record.pair_order != expected.pair_order or record.cell_order_index != expected.cell_order_index:
                raise ValueError("Paired collection record differs from its deterministic schedule.")
            previous_attempt = latest_attempt_by_pair.get(record.pair_index, record.block_attempt)
            if record.block_attempt < previous_attempt:
                raise ValueError("Paired collection block attempts must be chronological per pair.")
            latest_attempt_by_pair[record.pair_index] = record.block_attempt
            key = (record.pair_index, record.block_attempt)
            variants = block_variants.setdefault(key, set())
            if record.variant in variants:
                raise ValueError("Paired collection block attempts cannot repeat a variant.")
            variants.add(record.variant)
            attempts_by_pair.setdefault(record.pair_index, set()).add(record.block_attempt)
            matrix_anchor_seen = matrix_anchor_seen or record.status == "succeeded"
        records_by_block: dict[tuple[int, int], list[BenchmarkPairedRunRecord]] = {}
        for record in records:
            records_by_block.setdefault((record.pair_index, record.block_attempt), []).append(record)
        for (pair_index, _block_attempt), block_records in records_by_block.items():
            scheduled_variants = schedule[pair_index].variants
            observed_variants = tuple(record.variant for record in block_records)
            if observed_variants != scheduled_variants[: len(observed_variants)]:
                raise ValueError("Paired collection block records are not in scheduled AB/BA order.")
            if len(block_records) == 2 and block_records[1].index != block_records[0].index + 1:
                raise ValueError("Paired collection block members must be adjacent records.")
            if len(block_records) == 1 and block_records[0].index < len(records):
                following = records[block_records[0].index]
                if following.pair_index != pair_index or following.block_attempt != block_records[0].block_attempt + 1:
                    raise ValueError("Paired collection a partial block must be followed by a retry of the same pair.")
        for pair_index, attempts in attempts_by_pair.items():
            if sorted(attempts) != list(range(1, max(attempts) + 1)):
                raise ValueError("Paired collection block attempts must be contiguous per pair.")
            complete_attempts = sum(
                {
                    record.variant
                    for record in records
                    if record.pair_index == pair_index
                    and record.block_attempt == block_attempt
                    and record.status == "succeeded"
                }
                == {"baseline", "candidate"}
                for block_attempt in attempts
            )
            if complete_attempts > 1:
                raise ValueError("Paired collection cannot contain two complete blocks for one pair.")
            complete_attempt = next(
                (
                    block_attempt
                    for block_attempt in attempts
                    if {
                        record.variant
                        for record in records
                        if record.pair_index == pair_index
                        and record.block_attempt == block_attempt
                        and record.status == "succeeded"
                    }
                    == {"baseline", "candidate"}
                ),
                None,
            )
            if complete_attempt is not None and any(attempt > complete_attempt for attempt in attempts):
                raise ValueError("Paired collection cannot retry a pair after a complete block.")

        for variant, commit, fingerprint in (
            (
                "baseline",
                self.baseline_commit,
                self.baseline_environment_fingerprint,
            ),
            (
                "candidate",
                self.candidate_commit,
                self.candidate_environment_fingerprint,
            ),
        ):
            variant_records = tuple(record for record in successful_records if record.variant == variant)
            if not variant_records:
                if commit is not None or fingerprint is not None:
                    raise ValueError(f"Paired collection {variant} anchors require a successful record.")
                continue
            if variant_records[0].environment_fingerprint != fingerprint:
                raise ValueError(f"First successful {variant} fingerprint does not match the collection anchor.")
            if any(record.commit != commit for record in variant_records):
                raise ValueError(f"Successful {variant} commits do not match the collection anchor.")

        object.__setattr__(self, "baseline_command", baseline_command)
        object.__setattr__(self, "candidate_command", candidate_command)
        object.__setattr__(self, "expected_cells", expected_cells)
        object.__setattr__(self, "records", records)

    @classmethod
    def from_group(cls, group: BenchmarkPairedRunGroup) -> BenchmarkPairedCollectionSnapshot:
        """Create a portable snapshot from a loaded paired run group."""
        return cls(
            manifest=str(group.manifest_path),
            created_at=group.created_at,
            baseline_command=group.baseline_command,
            candidate_command=group.candidate_command,
            baseline_cwd=str(group.baseline_cwd),
            candidate_cwd=str(group.candidate_cwd),
            baseline_commit=group.baseline_commit,
            candidate_commit=group.candidate_commit,
            baseline_environment_fingerprint=group.baseline_environment_fingerprint,
            candidate_environment_fingerprint=group.candidate_environment_fingerprint,
            requested_pairs=group.requested_pairs,
            random_seed=group.random_seed,
            expected_cells=group.expected_cells,
            records=group.records,
            automatic_pairs=group.automatic_pairs,
        )

    @property
    def attempted_commands(self) -> int:
        """Return the number of completed command attempts."""
        return len(self.records)

    @property
    def successful_commands(self) -> int:
        """Return the number of accepted command attempts."""
        return sum(record.status == "succeeded" for record in self.records)

    @property
    def failed_commands(self) -> int:
        """Return the number of failed command attempts."""
        return sum(record.status == "failed" for record in self.records)

    @property
    def complete_pair_count(self) -> int:
        """Return the number of target pairs with one complete atomic block."""
        return len(self.complete_pair_records)

    @property
    def complete_pair_records(
        self,
    ) -> tuple[tuple[BenchmarkPairedRunRecord, BenchmarkPairedRunRecord], ...]:
        """Return baseline/candidate records for complete blocks in pair order."""
        successful = {
            (record.pair_index, record.block_attempt, record.variant): record
            for record in self.records
            if record.status == "succeeded"
        }
        pairs: list[tuple[BenchmarkPairedRunRecord, BenchmarkPairedRunRecord]] = []
        for pair_index in range(1, self.requested_pairs + 1):
            attempts = sorted(
                block_attempt
                for pair, block_attempt, variant in successful
                if pair == pair_index and variant == "baseline"
            )
            for block_attempt in attempts:
                baseline = successful.get((pair_index, block_attempt, "baseline"))
                candidate = successful.get((pair_index, block_attempt, "candidate"))
                if baseline is not None and candidate is not None:
                    pairs.append((baseline, candidate))
                    break
        return tuple(pairs)

    @property
    def baseline_sources(self) -> tuple[str, ...]:
        """Return complete-block baseline paths in target-pair order."""
        return tuple(str(baseline.path) for baseline, _candidate in self.complete_pair_records)

    @property
    def candidate_sources(self) -> tuple[str, ...]:
        """Return complete-block candidate paths in target-pair order."""
        return tuple(str(candidate.path) for _baseline, candidate in self.complete_pair_records)

    @property
    def orphan_success_count(self) -> int:
        """Return successful commands excluded from complete pairs."""
        return self.successful_commands - 2 * self.complete_pair_count

    @property
    def complete(self) -> bool:
        """Return whether every requested pair has a complete block."""
        return self.complete_pair_count == self.requested_pairs

    def to_dict(self) -> dict[str, JsonValue]:
        """Return the stable JSON representation of this paired collection."""
        return {
            "manifest": self.manifest,
            "created_at": self.created_at,
            "baseline_command": list(self.baseline_command),
            "candidate_command": list(self.candidate_command),
            "baseline_cwd": self.baseline_cwd,
            "candidate_cwd": self.candidate_cwd,
            "baseline_commit": self.baseline_commit,
            "candidate_commit": self.candidate_commit,
            "baseline_environment_fingerprint": self.baseline_environment_fingerprint,
            "candidate_environment_fingerprint": self.candidate_environment_fingerprint,
            "requested_pairs": self.requested_pairs,
            "random_seed": self.random_seed,
            "automatic_pairs": self.automatic_pairs,
            "attempted_commands": self.attempted_commands,
            "successful_commands": self.successful_commands,
            "failed_commands": self.failed_commands,
            "complete_pairs": self.complete_pair_count,
            "orphan_successes": self.orphan_success_count,
            "complete": self.complete,
            "expected_cells": [
                {
                    "implementation_name": implementation,
                    "case_name": case,
                    "metric_name": metric,
                }
                for implementation, case, metric in self.expected_cells
            ],
            "runs": [_paired_record_dict(record) for record in self.records],
        }

attempted_commands property

attempted_commands: int

Return the number of completed command attempts.

successful_commands property

successful_commands: int

Return the number of accepted command attempts.

failed_commands property

failed_commands: int

Return the number of failed command attempts.

complete_pair_count property

complete_pair_count: int

Return the number of target pairs with one complete atomic block.

complete_pair_records property

complete_pair_records: tuple[
    tuple[
        BenchmarkPairedRunRecord, BenchmarkPairedRunRecord
    ],
    ...,
]

Return baseline/candidate records for complete blocks in pair order.

baseline_sources property

baseline_sources: tuple[str, ...]

Return complete-block baseline paths in target-pair order.

candidate_sources property

candidate_sources: tuple[str, ...]

Return complete-block candidate paths in target-pair order.

orphan_success_count property

orphan_success_count: int

Return successful commands excluded from complete pairs.

complete property

complete: bool

Return whether every requested pair has a complete block.

__post_init__

__post_init__() -> None

Validate and normalize the portable paired snapshot.

Source code in src/benchmatrix/bench_report.py

def __post_init__(self) -> None:
    """Validate and normalize the portable paired snapshot."""
    if not isinstance(self.manifest, str) or not self.manifest:
        raise ValueError("Paired collection manifest must be a non-empty string.")
    if not isinstance(self.created_at, str) or not self.created_at:
        raise ValueError("Paired collection created_at must be a non-empty string.")
    baseline_command = tuple(self.baseline_command)
    candidate_command = tuple(self.candidate_command)
    for name, command in (
        ("baseline_command", baseline_command),
        ("candidate_command", candidate_command),
    ):
        if not command or any(not isinstance(argument, str) or not argument for argument in command):
            raise ValueError(f"Paired collection {name} must contain non-empty strings.")
    for name, cwd in (("baseline_cwd", self.baseline_cwd), ("candidate_cwd", self.candidate_cwd)):
        if not isinstance(cwd, str) or not cwd:
            raise ValueError(f"Paired collection {name} must be a non-empty string.")
    for name, value in (
        ("baseline_commit", self.baseline_commit),
        ("candidate_commit", self.candidate_commit),
        ("baseline_environment_fingerprint", self.baseline_environment_fingerprint),
        ("candidate_environment_fingerprint", self.candidate_environment_fingerprint),
    ):
        if value is not None and (not isinstance(value, str) or not value):
            raise ValueError(f"Paired collection {name} must be a non-empty string or None.")
    if (
        isinstance(self.requested_pairs, bool)
        or not isinstance(self.requested_pairs, int)
        or self.requested_pairs <= 0
    ):
        raise ValueError("Paired collection requested_pairs must be a positive integer.")
    if isinstance(self.random_seed, bool) or not isinstance(self.random_seed, int) or self.random_seed < 0:
        raise ValueError("Paired collection random_seed must be a non-negative integer.")
    if not isinstance(self.automatic_pairs, bool):
        raise TypeError("Paired collection automatic_pairs must be a boolean.")

    expected_cells = tuple(self.expected_cells)
    records = tuple(self.records)
    if len(set(expected_cells)) != len(expected_cells):
        raise ValueError("Paired collection expected_cells must not contain duplicates.")
    for implementation, case, metric in expected_cells:
        if not implementation or not case or metric not in KNOWN_METRICS:
            raise ValueError(f"Invalid paired collection expected cell: {(implementation, case, metric)!r}.")
    if any(not isinstance(record, BenchmarkPairedRunRecord) for record in records):
        raise TypeError("Paired collection records must contain BenchmarkPairedRunRecord values.")
    if tuple(record.index for record in records) != tuple(range(1, len(records) + 1)):
        raise ValueError("Paired collection record indexes must be contiguous and one-based.")
    canonical_paths = tuple(record.path.resolve() for record in records)
    if len(set(canonical_paths)) != len(canonical_paths):
        raise ValueError("Paired collection record paths must be unique.")

    successful_records = tuple(record for record in records if record.status == "succeeded")
    if successful_records and not expected_cells:
        raise ValueError("Paired collection with successful records requires expected cells.")
    if expected_cells and not successful_records:
        raise ValueError("Paired collection without successful records cannot define expected cells.")
    if self.automatic_pairs:
        if expected_cells:
            supercycle = balanced_order_supercycle_length(len(expected_cells))
            minimum_pairs = EvidencePolicy().minimum_runs
            expected_pair_target = ((minimum_pairs + supercycle - 1) // supercycle) * supercycle
            if self.requested_pairs != expected_pair_target:
                raise ValueError(
                    "Automatic paired collection target must be the smallest complete joint-design "
                    "supercycle at or above the evidence minimum."
                )
        elif self.requested_pairs != 6:
            raise ValueError("Automatic paired collection without successful records must retain target 6.")

    schedule = {
        entry.pair_index: entry
        for entry in make_paired_ab_ba_schedule(
            self.requested_pairs,
            random_seed=self.random_seed,
            cell_count=(len(expected_cells) if expected_cells else None),
        )
    }
    block_variants: dict[tuple[int, int], set[str]] = {}
    attempts_by_pair: dict[int, set[int]] = {}
    latest_attempt_by_pair: dict[int, int] = {}
    first_seen_pairs: set[int] = set()
    matrix_anchor_seen = False
    for record in records:
        if record.pair_index > self.requested_pairs:
            raise ValueError("Paired collection record pair_index exceeds requested_pairs.")
        if record.pair_index not in first_seen_pairs:
            expected_pair_index = len(first_seen_pairs) + 1
            if record.pair_index != expected_pair_index:
                raise ValueError("Paired collection target pairs must first appear as a one-based prefix.")
            first_seen_pairs.add(record.pair_index)
        if record.pair_index > 1 and not matrix_anchor_seen:
            raise ValueError("Paired collection cannot start later target pairs before a successful matrix exists.")
        expected = schedule[record.pair_index]
        if record.pair_order != expected.pair_order or record.cell_order_index != expected.cell_order_index:
            raise ValueError("Paired collection record differs from its deterministic schedule.")
        previous_attempt = latest_attempt_by_pair.get(record.pair_index, record.block_attempt)
        if record.block_attempt < previous_attempt:
            raise ValueError("Paired collection block attempts must be chronological per pair.")
        latest_attempt_by_pair[record.pair_index] = record.block_attempt
        key = (record.pair_index, record.block_attempt)
        variants = block_variants.setdefault(key, set())
        if record.variant in variants:
            raise ValueError("Paired collection block attempts cannot repeat a variant.")
        variants.add(record.variant)
        attempts_by_pair.setdefault(record.pair_index, set()).add(record.block_attempt)
        matrix_anchor_seen = matrix_anchor_seen or record.status == "succeeded"
    records_by_block: dict[tuple[int, int], list[BenchmarkPairedRunRecord]] = {}
    for record in records:
        records_by_block.setdefault((record.pair_index, record.block_attempt), []).append(record)
    for (pair_index, _block_attempt), block_records in records_by_block.items():
        scheduled_variants = schedule[pair_index].variants
        observed_variants = tuple(record.variant for record in block_records)
        if observed_variants != scheduled_variants[: len(observed_variants)]:
            raise ValueError("Paired collection block records are not in scheduled AB/BA order.")
        if len(block_records) == 2 and block_records[1].index != block_records[0].index + 1:
            raise ValueError("Paired collection block members must be adjacent records.")
        if len(block_records) == 1 and block_records[0].index < len(records):
            following = records[block_records[0].index]
            if following.pair_index != pair_index or following.block_attempt != block_records[0].block_attempt + 1:
                raise ValueError("Paired collection a partial block must be followed by a retry of the same pair.")
    for pair_index, attempts in attempts_by_pair.items():
        if sorted(attempts) != list(range(1, max(attempts) + 1)):
            raise ValueError("Paired collection block attempts must be contiguous per pair.")
        complete_attempts = sum(
            {
                record.variant
                for record in records
                if record.pair_index == pair_index
                and record.block_attempt == block_attempt
                and record.status == "succeeded"
            }
            == {"baseline", "candidate"}
            for block_attempt in attempts
        )
        if complete_attempts > 1:
            raise ValueError("Paired collection cannot contain two complete blocks for one pair.")
        complete_attempt = next(
            (
                block_attempt
                for block_attempt in attempts
                if {
                    record.variant
                    for record in records
                    if record.pair_index == pair_index
                    and record.block_attempt == block_attempt
                    and record.status == "succeeded"
                }
                == {"baseline", "candidate"}
            ),
            None,
        )
        if complete_attempt is not None and any(attempt > complete_attempt for attempt in attempts):
            raise ValueError("Paired collection cannot retry a pair after a complete block.")

    for variant, commit, fingerprint in (
        (
            "baseline",
            self.baseline_commit,
            self.baseline_environment_fingerprint,
        ),
        (
            "candidate",
            self.candidate_commit,
            self.candidate_environment_fingerprint,
        ),
    ):
        variant_records = tuple(record for record in successful_records if record.variant == variant)
        if not variant_records:
            if commit is not None or fingerprint is not None:
                raise ValueError(f"Paired collection {variant} anchors require a successful record.")
            continue
        if variant_records[0].environment_fingerprint != fingerprint:
            raise ValueError(f"First successful {variant} fingerprint does not match the collection anchor.")
        if any(record.commit != commit for record in variant_records):
            raise ValueError(f"Successful {variant} commits do not match the collection anchor.")

    object.__setattr__(self, "baseline_command", baseline_command)
    object.__setattr__(self, "candidate_command", candidate_command)
    object.__setattr__(self, "expected_cells", expected_cells)
    object.__setattr__(self, "records", records)

from_group classmethod

from_group(
    group: BenchmarkPairedRunGroup,
) -> BenchmarkPairedCollectionSnapshot

Create a portable snapshot from a loaded paired run group.

Source code in src/benchmatrix/bench_report.py

@classmethod
def from_group(cls, group: BenchmarkPairedRunGroup) -> BenchmarkPairedCollectionSnapshot:
    """Create a portable snapshot from a loaded paired run group."""
    return cls(
        manifest=str(group.manifest_path),
        created_at=group.created_at,
        baseline_command=group.baseline_command,
        candidate_command=group.candidate_command,
        baseline_cwd=str(group.baseline_cwd),
        candidate_cwd=str(group.candidate_cwd),
        baseline_commit=group.baseline_commit,
        candidate_commit=group.candidate_commit,
        baseline_environment_fingerprint=group.baseline_environment_fingerprint,
        candidate_environment_fingerprint=group.candidate_environment_fingerprint,
        requested_pairs=group.requested_pairs,
        random_seed=group.random_seed,
        expected_cells=group.expected_cells,
        records=group.records,
        automatic_pairs=group.automatic_pairs,
    )

to_dict

to_dict() -> dict[str, JsonValue]

Return the stable JSON representation of this paired collection.

Source code in src/benchmatrix/bench_report.py

def to_dict(self) -> dict[str, JsonValue]:
    """Return the stable JSON representation of this paired collection."""
    return {
        "manifest": self.manifest,
        "created_at": self.created_at,
        "baseline_command": list(self.baseline_command),
        "candidate_command": list(self.candidate_command),
        "baseline_cwd": self.baseline_cwd,
        "candidate_cwd": self.candidate_cwd,
        "baseline_commit": self.baseline_commit,
        "candidate_commit": self.candidate_commit,
        "baseline_environment_fingerprint": self.baseline_environment_fingerprint,
        "candidate_environment_fingerprint": self.candidate_environment_fingerprint,
        "requested_pairs": self.requested_pairs,
        "random_seed": self.random_seed,
        "automatic_pairs": self.automatic_pairs,
        "attempted_commands": self.attempted_commands,
        "successful_commands": self.successful_commands,
        "failed_commands": self.failed_commands,
        "complete_pairs": self.complete_pair_count,
        "orphan_successes": self.orphan_success_count,
        "complete": self.complete,
        "expected_cells": [
            {
                "implementation_name": implementation,
                "case_name": case,
                "metric_name": metric,
            }
            for implementation, case, metric in self.expected_cells
        ],
        "runs": [_paired_record_dict(record) for record in self.records],
    }

BenchmarkPolicyProvenance dataclass

Configuration provenance embedded in a comparison report.

Source code in src/benchmatrix/bench_report.py

@dataclass(frozen=True, slots=True)
class BenchmarkPolicyProvenance:
    """Configuration provenance embedded in a comparison report."""

    selection: PolicySelection
    configuration_file: str | None = None
    configured_fields: tuple[str, ...] = ()
    cli_overrides: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        """Validate and normalize policy provenance."""
        if self.selection not in {"defaults", "disabled", "discovered", "explicit"}:
            raise ValueError(f"Unsupported policy selection: {self.selection!r}.")
        if self.configuration_file is not None and (
            not isinstance(self.configuration_file, str) or not self.configuration_file
        ):
            raise ValueError("configuration_file must be a non-empty string or None.")
        configured_fields = tuple(self.configured_fields)
        cli_overrides = tuple(self.cli_overrides)
        if any(not isinstance(field, str) or not field for field in (*configured_fields, *cli_overrides)):
            raise ValueError("Policy provenance fields must be non-empty strings.")
        object.__setattr__(self, "configured_fields", configured_fields)
        object.__setattr__(self, "cli_overrides", cli_overrides)

__post_init__

__post_init__() -> None

Validate and normalize policy provenance.

Source code in src/benchmatrix/bench_report.py

def __post_init__(self) -> None:
    """Validate and normalize policy provenance."""
    if self.selection not in {"defaults", "disabled", "discovered", "explicit"}:
        raise ValueError(f"Unsupported policy selection: {self.selection!r}.")
    if self.configuration_file is not None and (
        not isinstance(self.configuration_file, str) or not self.configuration_file
    ):
        raise ValueError("configuration_file must be a non-empty string or None.")
    configured_fields = tuple(self.configured_fields)
    cli_overrides = tuple(self.cli_overrides)
    if any(not isinstance(field, str) or not field for field in (*configured_fields, *cli_overrides)):
        raise ValueError("Policy provenance fields must be non-empty strings.")
    object.__setattr__(self, "configured_fields", configured_fields)
    object.__setattr__(self, "cli_overrides", cli_overrides)

BenchmarkThresholdProvenance dataclass

Rule scope and origin supplying one reported cell threshold.

Source code in src/benchmatrix/bench_report.py

@dataclass(frozen=True, slots=True)
class BenchmarkThresholdProvenance:
    """Rule scope and origin supplying one reported cell threshold."""

    scope: RegressionThresholdScope
    origin: ThresholdOrigin
    field: str

    def __post_init__(self) -> None:
        """Validate threshold provenance."""
        if self.scope not in {"cell", "case", "implementation", "metric", "default"}:
            raise ValueError(f"Unsupported threshold scope: {self.scope!r}.")
        if self.origin not in {"built_in", "configuration", "cli"}:
            raise ValueError(f"Unsupported threshold origin: {self.origin!r}.")
        if not isinstance(self.field, str) or not self.field:
            raise ValueError("Threshold provenance field must be a non-empty string.")

__post_init__

__post_init__() -> None

Validate threshold provenance.

Source code in src/benchmatrix/bench_report.py

def __post_init__(self) -> None:
    """Validate threshold provenance."""
    if self.scope not in {"cell", "case", "implementation", "metric", "default"}:
        raise ValueError(f"Unsupported threshold scope: {self.scope!r}.")
    if self.origin not in {"built_in", "configuration", "cli"}:
        raise ValueError(f"Unsupported threshold origin: {self.origin!r}.")
    if not isinstance(self.field, str) or not self.field:
        raise ValueError("Threshold provenance field must be a non-empty string.")

BenchmarkRun dataclass

One parsed pytest-benchmark run containing a benchmark matrix.

Attributes:

Name	Type	Description
rows	tuple[ParsedBenchmarkRow, ...]	Benchmatrix rows in their source-file order.
metadata	Mapping[str, object]	Top-level pytest-benchmark metadata excluding benchmarks.
source	Path | None	Source JSON path, when the run was loaded from a file.

Source code in src/benchmatrix/bench_results.py

@dataclass(frozen=True, slots=True)
class BenchmarkRun:
    """One parsed pytest-benchmark run containing a benchmark matrix.

    Attributes:
        rows: Benchmatrix rows in their source-file order.
        metadata: Top-level pytest-benchmark metadata excluding ``benchmarks``.
        source: Source JSON path, when the run was loaded from a file.
    """

    rows: tuple[ParsedBenchmarkRow, ...]
    metadata: Mapping[str, object]
    source: Path | None = None

    def __post_init__(self) -> None:
        """Normalize run containers and reject duplicate matrix cells."""
        rows = tuple(self.rows)
        metadata = MappingProxyType(dict(self.metadata))
        seen: set[tuple[str, str, MetricName]] = set()

        if not rows:
            raise BenchmarkJsonError("Benchmark run must contain at least one benchmatrix row.")

        for row in rows:
            if not row.implementation_name or not row.case_name:
                raise BenchmarkJsonError("Benchmark run matrix identifiers must not be empty.")
            if row.metric_name not in KNOWN_METRICS:
                raise BenchmarkJsonError(f"Unsupported benchmatrix metric in benchmark run: {row.metric_name!r}.")

            key = (row.implementation_name, row.case_name, row.metric_name)
            if key in seen:
                implementation_name, case_name, metric_name = key
                message = (
                    "Duplicate benchmark matrix cell for "
                    + f"implementation={implementation_name!r}, case={case_name!r}, metric={metric_name!r}."
                )
                raise BenchmarkJsonError(message)
            seen.add(key)

        object.__setattr__(self, "rows", rows)
        object.__setattr__(self, "metadata", metadata)
        if self.source is not None:
            object.__setattr__(self, "source", Path(self.source))

    @property
    def implementations(self) -> tuple[str, ...]:
        """Return sorted implementation names represented in this run."""
        return tuple(sorted({row.implementation_name for row in self.rows}))

    @property
    def cases(self) -> tuple[str, ...]:
        """Return sorted case names represented in this run."""
        return tuple(sorted({row.case_name for row in self.rows}))

    @property
    def metrics(self) -> tuple[MetricName, ...]:
        """Return sorted metric names represented in this run."""
        return tuple(sorted({row.metric_name for row in self.rows}))

    def compare_to(
        self,
        candidate: BenchmarkRun,
        *,
        compatibility_policy: RunCompatibilityPolicy | None = None,
        regression_policy: RegressionPolicy | None = None,
        inference_policy: InferencePolicy | None = None,
    ) -> BenchmarkRunComparison:
        """Compare this baseline run with a candidate run.

        Args:
            candidate: Run whose values should be compared with this baseline.
            compatibility_policy: Environment checks to apply.
            regression_policy: Thresholds used to classify cell changes.
            inference_policy: Statistical inference and multiplicity controls.

        Returns:
            A matrix-aware comparison containing matched, missing, and
            incompatible cells.
        """
        from .bench_compare import compare_benchmark_runs

        return compare_benchmark_runs(
            self,
            candidate,
            compatibility_policy=compatibility_policy,
            regression_policy=regression_policy,
            inference_policy=inference_policy,
        )

implementations property

implementations: tuple[str, ...]

Return sorted implementation names represented in this run.

cases property

cases: tuple[str, ...]

Return sorted case names represented in this run.

metrics property

metrics: tuple[MetricName, ...]

Return sorted metric names represented in this run.

__post_init__

__post_init__() -> None

Normalize run containers and reject duplicate matrix cells.

Source code in src/benchmatrix/bench_results.py

def __post_init__(self) -> None:
    """Normalize run containers and reject duplicate matrix cells."""
    rows = tuple(self.rows)
    metadata = MappingProxyType(dict(self.metadata))
    seen: set[tuple[str, str, MetricName]] = set()

    if not rows:
        raise BenchmarkJsonError("Benchmark run must contain at least one benchmatrix row.")

    for row in rows:
        if not row.implementation_name or not row.case_name:
            raise BenchmarkJsonError("Benchmark run matrix identifiers must not be empty.")
        if row.metric_name not in KNOWN_METRICS:
            raise BenchmarkJsonError(f"Unsupported benchmatrix metric in benchmark run: {row.metric_name!r}.")

        key = (row.implementation_name, row.case_name, row.metric_name)
        if key in seen:
            implementation_name, case_name, metric_name = key
            message = (
                "Duplicate benchmark matrix cell for "
                + f"implementation={implementation_name!r}, case={case_name!r}, metric={metric_name!r}."
            )
            raise BenchmarkJsonError(message)
        seen.add(key)

    object.__setattr__(self, "rows", rows)
    object.__setattr__(self, "metadata", metadata)
    if self.source is not None:
        object.__setattr__(self, "source", Path(self.source))

compare_to

compare_to(
    candidate: BenchmarkRun,
    *,
    compatibility_policy: RunCompatibilityPolicy
    | None = None,
    regression_policy: RegressionPolicy | None = None,
    inference_policy: InferencePolicy | None = None,
) -> BenchmarkRunComparison

Compare this baseline run with a candidate run.

Parameters:

Name	Type	Description	Default
candidate	BenchmarkRun	Run whose values should be compared with this baseline.	required
compatibility_policy	RunCompatibilityPolicy | None	Environment checks to apply.	None
regression_policy	RegressionPolicy | None	Thresholds used to classify cell changes.	None
inference_policy	InferencePolicy | None	Statistical inference and multiplicity controls.	None

Returns:

Type	Description
BenchmarkRunComparison	A matrix-aware comparison containing matched, missing, and
BenchmarkRunComparison	incompatible cells.

Source code in src/benchmatrix/bench_results.py

def compare_to(
    self,
    candidate: BenchmarkRun,
    *,
    compatibility_policy: RunCompatibilityPolicy | None = None,
    regression_policy: RegressionPolicy | None = None,
    inference_policy: InferencePolicy | None = None,
) -> BenchmarkRunComparison:
    """Compare this baseline run with a candidate run.

    Args:
        candidate: Run whose values should be compared with this baseline.
        compatibility_policy: Environment checks to apply.
        regression_policy: Thresholds used to classify cell changes.
        inference_policy: Statistical inference and multiplicity controls.

    Returns:
        A matrix-aware comparison containing matched, missing, and
        incompatible cells.
    """
    from .bench_compare import compare_benchmark_runs

    return compare_benchmark_runs(
        self,
        candidate,
        compatibility_policy=compatibility_policy,
        regression_policy=regression_policy,
        inference_policy=inference_policy,
    )

ParsedBenchmarkRow dataclass

One benchmatrix-tagged row parsed from pytest-benchmark JSON output.

Attributes:

Name	Type	Description
benchmark_name	str	Name assigned by pytest-benchmark to this benchmark.
metric_name	MetricName	Benchmatrix metric name from extra_info.
implementation_name	str	Implementation name from extra_info.
case_name	str	Case name from extra_info.
stats	Mapping[str, object]	Raw pytest-benchmark timing statistics.
extra_info	Mapping[str, object]	Custom metadata from benchmark.extra_info.
derived	Mapping[str, object]	Derived metric-specific statistics computed from JSON output.
samples	tuple[float, ...]	Raw per-round timing samples in seconds.

Source code in src/benchmatrix/bench_results.py

@dataclass(frozen=True, slots=True)
class ParsedBenchmarkRow:
    """One benchmatrix-tagged row parsed from pytest-benchmark JSON output.

    Attributes:
        benchmark_name: Name assigned by pytest-benchmark to this benchmark.
        metric_name: Benchmatrix metric name from ``extra_info``.
        implementation_name: Implementation name from ``extra_info``.
        case_name: Case name from ``extra_info``.
        stats: Raw pytest-benchmark timing statistics.
        extra_info: Custom metadata from ``benchmark.extra_info``.
        derived: Derived metric-specific statistics computed from JSON output.
        samples: Raw per-round timing samples in seconds.
    """

    benchmark_name: str
    metric_name: MetricName
    implementation_name: str
    case_name: str
    stats: Mapping[str, object]
    extra_info: Mapping[str, object]
    derived: Mapping[str, object]
    samples: tuple[float, ...] = ()

PrecisionPlan dataclass

Fixed-design pair-count plan derived from pilot paired log ratios.

The planning estimand is the mean signed paired log ratio, a variance-based proxy rather than the ratio-of-marginal-medians estimand used by formal BCa inference. This is a precision calculation, not power analysis. Its pair-count result assumes that pilot variability is representative and that a fresh confirmatory collection uses the complete planned pair count fixed before examining its results. additional_pairs is only the arithmetic difference from the pilot size; it does not endorse reusing pilot outcomes.

Source code in src/benchmatrix/bench_statistics.py

@dataclass(frozen=True, slots=True)
class PrecisionPlan:
    """Fixed-design pair-count plan derived from pilot paired log ratios.

    The planning estimand is the mean signed paired log ratio, a variance-based
    proxy rather than the ratio-of-marginal-medians estimand used by formal BCa
    inference. This is a precision calculation, not power analysis. Its
    pair-count result assumes that pilot variability is representative and that
    a fresh confirmatory collection uses the complete planned pair count fixed
    before examining its results. ``additional_pairs`` is only the arithmetic
    difference from the pilot size; it does not endorse reusing pilot outcomes.
    """

    method: Literal["paired_log_ratio_t"]
    pilot_pairs: int
    target_half_width_percent: float
    confidence_level: float
    adjusted_confidence_level: float
    multiplicity: MultiplicityCorrection
    family_size: int
    pilot_log_ratio_standard_deviation: float | None
    critical_value: float | None
    required_pairs: int | None
    additional_pairs: int | None
    assumptions: tuple[str, ...]
    minimum_pairs: int = _MINIMUM_GROUP_SIZE
    pair_count_multiple: int = 1
    unconstrained_required_pairs: int | None = None
    strata_count: int = 1
    warnings: tuple[str, ...] = ()
    issues: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        """Validate and normalize a precision plan."""
        if self.method != "paired_log_ratio_t":
            raise ValueError(f"Unsupported precision-planning method: {self.method!r}.")
        if isinstance(self.pilot_pairs, bool) or not isinstance(self.pilot_pairs, int) or self.pilot_pairs < 0:
            raise ValueError("PrecisionPlan.pilot_pairs must be a non-negative integer.")
        if not math.isfinite(self.target_half_width_percent) or self.target_half_width_percent <= 0.0:
            raise ValueError("PrecisionPlan.target_half_width_percent must be finite and positive.")
        for field_name, value in (
            ("confidence_level", self.confidence_level),
            ("adjusted_confidence_level", self.adjusted_confidence_level),
        ):
            if not math.isfinite(value) or not 0.0 < value < 1.0:
                raise ValueError(f"PrecisionPlan.{field_name} must be finite and between zero and one.")
        if self.adjusted_confidence_level < self.confidence_level:
            raise ValueError("Adjusted confidence level must not be lower than the nominal confidence level.")
        if self.multiplicity not in {"bonferroni", "none"}:
            raise ValueError(f"Unsupported multiplicity correction: {self.multiplicity!r}.")
        if isinstance(self.family_size, bool) or not isinstance(self.family_size, int) or self.family_size <= 0:
            raise ValueError("PrecisionPlan.family_size must be a positive integer.")
        if (
            isinstance(self.minimum_pairs, bool)
            or not isinstance(self.minimum_pairs, int)
            or self.minimum_pairs < _MINIMUM_GROUP_SIZE
        ):
            raise ValueError(f"PrecisionPlan.minimum_pairs must be at least {_MINIMUM_GROUP_SIZE}.")
        if (
            isinstance(self.pair_count_multiple, bool)
            or not isinstance(self.pair_count_multiple, int)
            or self.pair_count_multiple <= 0
        ):
            raise ValueError("PrecisionPlan.pair_count_multiple must be a positive integer.")
        if isinstance(self.strata_count, bool) or not isinstance(self.strata_count, int) or self.strata_count < 0:
            raise ValueError("PrecisionPlan.strata_count must be a non-negative integer.")
        if self.multiplicity == "none" and self.adjusted_confidence_level != self.confidence_level:
            raise ValueError("Unadjusted precision plans must use the nominal confidence level.")
        if self.pilot_log_ratio_standard_deviation is not None and (
            not math.isfinite(self.pilot_log_ratio_standard_deviation) or self.pilot_log_ratio_standard_deviation < 0.0
        ):
            raise ValueError("PrecisionPlan pilot variability must be finite and non-negative or None.")
        if self.critical_value is not None and (not math.isfinite(self.critical_value) or self.critical_value <= 0.0):
            raise ValueError("PrecisionPlan.critical_value must be finite and positive or None.")
        unconstrained_required_pairs = self.unconstrained_required_pairs
        if self.required_pairs is not None and unconstrained_required_pairs is None:
            # Preserve compatibility for callers that constructed the original
            # value object directly. Planner-created values always persist the
            # independently calculated unconstrained count.
            unconstrained_required_pairs = self.required_pairs
            object.__setattr__(self, "unconstrained_required_pairs", unconstrained_required_pairs)
        result_fields = (
            self.critical_value,
            unconstrained_required_pairs,
            self.required_pairs,
            self.additional_pairs,
        )
        if any(value is not None for value in result_fields) and not all(value is not None for value in result_fields):
            raise ValueError("PrecisionPlan critical value and pair-count results must be present together.")
        if self.required_pairs is not None:
            if self.strata_count <= 0:
                raise ValueError("Complete PrecisionPlan results require at least one fitted stratum.")
            if self.pilot_log_ratio_standard_deviation is None or self.pilot_log_ratio_standard_deviation == 0.0:
                raise ValueError("Complete PrecisionPlan results require positive pilot residual variability.")
            minimum_estimable_pairs = self.strata_count + 1
            if (
                isinstance(unconstrained_required_pairs, bool)
                or not isinstance(unconstrained_required_pairs, int)
                or unconstrained_required_pairs < max(_MINIMUM_GROUP_SIZE, minimum_estimable_pairs)
            ):
                raise ValueError(
                    "PrecisionPlan.unconstrained_required_pairs must leave positive residual degrees of freedom."
                )
            if (
                isinstance(self.required_pairs, bool)
                or not isinstance(self.required_pairs, int)
                or self.required_pairs < max(_MINIMUM_GROUP_SIZE, minimum_estimable_pairs)
            ):
                raise ValueError("PrecisionPlan.required_pairs must leave positive residual degrees of freedom.")
            expected_required = _round_up_to_multiple(
                max(unconstrained_required_pairs, self.minimum_pairs),
                self.pair_count_multiple,
            )
            if self.required_pairs != expected_required:
                raise ValueError(
                    "PrecisionPlan.required_pairs is inconsistent with the unconstrained count and design constraints."
                )
            expected_unconstrained = _required_pairs_for_precision(
                self.pilot_log_ratio_standard_deviation,
                target_log_half_width=math.log1p(self.target_half_width_percent / 100.0),
                confidence_level=self.adjusted_confidence_level,
                strata_count=self.strata_count,
            )
            if unconstrained_required_pairs != expected_unconstrained:
                raise ValueError(
                    "PrecisionPlan.unconstrained_required_pairs is inconsistent with its variability and target."
                )
            expected_critical = _student_t_critical(
                self.adjusted_confidence_level,
                degrees_of_freedom=self.required_pairs - self.strata_count,
            )
            critical_value = self.critical_value
            if critical_value is None or not math.isclose(
                critical_value,
                expected_critical,
                rel_tol=1e-12,
                abs_tol=0.0,
            ):
                raise ValueError(
                    "PrecisionPlan.critical_value is inconsistent with confidence and residual degrees of freedom."
                )
            expected_additional = max(0, self.required_pairs - self.pilot_pairs)
            if self.additional_pairs != expected_additional:
                raise ValueError("PrecisionPlan.additional_pairs is inconsistent with the pilot and required counts.")
        assumptions = tuple(self.assumptions)
        warnings = tuple(self.warnings)
        issues = tuple(self.issues)
        if not assumptions or any(not isinstance(item, str) or not item for item in assumptions):
            raise ValueError("PrecisionPlan.assumptions must contain non-empty strings.")
        if any(not isinstance(item, str) or not item for item in (*warnings, *issues)):
            raise ValueError("PrecisionPlan warnings and issues must contain non-empty strings.")
        object.__setattr__(self, "assumptions", assumptions)
        object.__setattr__(self, "warnings", warnings)
        object.__setattr__(self, "issues", issues)

    @property
    def adequate(self) -> bool:
        """Return whether a complete fixed-design pair-count estimate exists."""
        return not self.issues and self.required_pairs is not None and self.additional_pairs is not None

adequate property

adequate: bool

Return whether a complete fixed-design pair-count estimate exists.

__post_init__

__post_init__() -> None

Validate and normalize a precision plan.

Source code in src/benchmatrix/bench_statistics.py

def __post_init__(self) -> None:
    """Validate and normalize a precision plan."""
    if self.method != "paired_log_ratio_t":
        raise ValueError(f"Unsupported precision-planning method: {self.method!r}.")
    if isinstance(self.pilot_pairs, bool) or not isinstance(self.pilot_pairs, int) or self.pilot_pairs < 0:
        raise ValueError("PrecisionPlan.pilot_pairs must be a non-negative integer.")
    if not math.isfinite(self.target_half_width_percent) or self.target_half_width_percent <= 0.0:
        raise ValueError("PrecisionPlan.target_half_width_percent must be finite and positive.")
    for field_name, value in (
        ("confidence_level", self.confidence_level),
        ("adjusted_confidence_level", self.adjusted_confidence_level),
    ):
        if not math.isfinite(value) or not 0.0 < value < 1.0:
            raise ValueError(f"PrecisionPlan.{field_name} must be finite and between zero and one.")
    if self.adjusted_confidence_level < self.confidence_level:
        raise ValueError("Adjusted confidence level must not be lower than the nominal confidence level.")
    if self.multiplicity not in {"bonferroni", "none"}:
        raise ValueError(f"Unsupported multiplicity correction: {self.multiplicity!r}.")
    if isinstance(self.family_size, bool) or not isinstance(self.family_size, int) or self.family_size <= 0:
        raise ValueError("PrecisionPlan.family_size must be a positive integer.")
    if (
        isinstance(self.minimum_pairs, bool)
        or not isinstance(self.minimum_pairs, int)
        or self.minimum_pairs < _MINIMUM_GROUP_SIZE
    ):
        raise ValueError(f"PrecisionPlan.minimum_pairs must be at least {_MINIMUM_GROUP_SIZE}.")
    if (
        isinstance(self.pair_count_multiple, bool)
        or not isinstance(self.pair_count_multiple, int)
        or self.pair_count_multiple <= 0
    ):
        raise ValueError("PrecisionPlan.pair_count_multiple must be a positive integer.")
    if isinstance(self.strata_count, bool) or not isinstance(self.strata_count, int) or self.strata_count < 0:
        raise ValueError("PrecisionPlan.strata_count must be a non-negative integer.")
    if self.multiplicity == "none" and self.adjusted_confidence_level != self.confidence_level:
        raise ValueError("Unadjusted precision plans must use the nominal confidence level.")
    if self.pilot_log_ratio_standard_deviation is not None and (
        not math.isfinite(self.pilot_log_ratio_standard_deviation) or self.pilot_log_ratio_standard_deviation < 0.0
    ):
        raise ValueError("PrecisionPlan pilot variability must be finite and non-negative or None.")
    if self.critical_value is not None and (not math.isfinite(self.critical_value) or self.critical_value <= 0.0):
        raise ValueError("PrecisionPlan.critical_value must be finite and positive or None.")
    unconstrained_required_pairs = self.unconstrained_required_pairs
    if self.required_pairs is not None and unconstrained_required_pairs is None:
        # Preserve compatibility for callers that constructed the original
        # value object directly. Planner-created values always persist the
        # independently calculated unconstrained count.
        unconstrained_required_pairs = self.required_pairs
        object.__setattr__(self, "unconstrained_required_pairs", unconstrained_required_pairs)
    result_fields = (
        self.critical_value,
        unconstrained_required_pairs,
        self.required_pairs,
        self.additional_pairs,
    )
    if any(value is not None for value in result_fields) and not all(value is not None for value in result_fields):
        raise ValueError("PrecisionPlan critical value and pair-count results must be present together.")
    if self.required_pairs is not None:
        if self.strata_count <= 0:
            raise ValueError("Complete PrecisionPlan results require at least one fitted stratum.")
        if self.pilot_log_ratio_standard_deviation is None or self.pilot_log_ratio_standard_deviation == 0.0:
            raise ValueError("Complete PrecisionPlan results require positive pilot residual variability.")
        minimum_estimable_pairs = self.strata_count + 1
        if (
            isinstance(unconstrained_required_pairs, bool)
            or not isinstance(unconstrained_required_pairs, int)
            or unconstrained_required_pairs < max(_MINIMUM_GROUP_SIZE, minimum_estimable_pairs)
        ):
            raise ValueError(
                "PrecisionPlan.unconstrained_required_pairs must leave positive residual degrees of freedom."
            )
        if (
            isinstance(self.required_pairs, bool)
            or not isinstance(self.required_pairs, int)
            or self.required_pairs < max(_MINIMUM_GROUP_SIZE, minimum_estimable_pairs)
        ):
            raise ValueError("PrecisionPlan.required_pairs must leave positive residual degrees of freedom.")
        expected_required = _round_up_to_multiple(
            max(unconstrained_required_pairs, self.minimum_pairs),
            self.pair_count_multiple,
        )
        if self.required_pairs != expected_required:
            raise ValueError(
                "PrecisionPlan.required_pairs is inconsistent with the unconstrained count and design constraints."
            )
        expected_unconstrained = _required_pairs_for_precision(
            self.pilot_log_ratio_standard_deviation,
            target_log_half_width=math.log1p(self.target_half_width_percent / 100.0),
            confidence_level=self.adjusted_confidence_level,
            strata_count=self.strata_count,
        )
        if unconstrained_required_pairs != expected_unconstrained:
            raise ValueError(
                "PrecisionPlan.unconstrained_required_pairs is inconsistent with its variability and target."
            )
        expected_critical = _student_t_critical(
            self.adjusted_confidence_level,
            degrees_of_freedom=self.required_pairs - self.strata_count,
        )
        critical_value = self.critical_value
        if critical_value is None or not math.isclose(
            critical_value,
            expected_critical,
            rel_tol=1e-12,
            abs_tol=0.0,
        ):
            raise ValueError(
                "PrecisionPlan.critical_value is inconsistent with confidence and residual degrees of freedom."
            )
        expected_additional = max(0, self.required_pairs - self.pilot_pairs)
        if self.additional_pairs != expected_additional:
            raise ValueError("PrecisionPlan.additional_pairs is inconsistent with the pilot and required counts.")
    assumptions = tuple(self.assumptions)
    warnings = tuple(self.warnings)
    issues = tuple(self.issues)
    if not assumptions or any(not isinstance(item, str) or not item for item in assumptions):
        raise ValueError("PrecisionPlan.assumptions must contain non-empty strings.")
    if any(not isinstance(item, str) or not item for item in (*warnings, *issues)):
        raise ValueError("PrecisionPlan warnings and issues must contain non-empty strings.")
    object.__setattr__(self, "assumptions", assumptions)
    object.__setattr__(self, "warnings", warnings)
    object.__setattr__(self, "issues", issues)

BenchmarkCollectionError

Bases: BenchmatrixError, RuntimeError

Raised when a repeated-run collection cannot be created.

Source code in src/benchmatrix/exceptions.py

class BenchmarkCollectionError(BenchmatrixError, RuntimeError):
    """Raised when a repeated-run collection cannot be created."""

BenchmarkJsonError

Bases: BenchmatrixError, ValueError

Raised when pytest-benchmark JSON cannot be parsed as benchmatrix output.

Source code in src/benchmatrix/exceptions.py

class BenchmarkJsonError(BenchmatrixError, ValueError):
    """Raised when pytest-benchmark JSON cannot be parsed as benchmatrix output."""

BenchmarkPolicyError

Bases: BenchmatrixError, ValueError

Raised when benchmark policy configuration is invalid.

Source code in src/benchmatrix/exceptions.py

class BenchmarkPolicyError(BenchmatrixError, ValueError):
    """Raised when benchmark policy configuration is invalid."""

BenchmatrixError

Bases: Exception

Base class for benchmatrix matrix, metadata, and result errors.

Source code in src/benchmatrix/exceptions.py

class BenchmatrixError(Exception):
    """Base class for benchmatrix matrix, metadata, and result errors."""

MetadataSerializationError

Bases: BenchmatrixError, ValueError

Raised when benchmark metadata cannot be represented as strict JSON.

Source code in src/benchmatrix/exceptions.py

class MetadataSerializationError(BenchmatrixError, ValueError):
    """Raised when benchmark metadata cannot be represented as strict JSON."""

balanced_cell_order

balanced_cell_order(
    cells: Sequence[tuple[str, str, MetricName]],
    *,
    order_index: int,
    random_seed: int = 0,
) -> tuple[tuple[str, str, MetricName], ...]

Return one deterministic position- and carryover-balanced cell order.

Even-sized matrices repeat after n order indexes. Odd-sized matrices larger than one repeat after 2n indexes because every cyclic row is followed by a reversed cycle. A one-cell matrix has a one-row cycle.

Parameters:

Name	Type	Description	Default
cells	Sequence[tuple[str, str, MetricName]]	Unique (implementation, case, metric) matrix cells.	required
order_index	int	One-based schedule row.	required
random_seed	int	Non-negative seed for the stable base-label permutation.	0

Returns:

Type	Description
tuple[tuple[str, str, MetricName], ...]	The cells in the scheduled execution order.

Raises:

Type	Description
TypeError	If an index or seed is not an integer.
ValueError	If an index, seed, or cell is invalid, or cells repeat.

Source code in src/benchmatrix/_collection_design.py

def balanced_cell_order(
    cells: Sequence[tuple[str, str, MetricName]],
    *,
    order_index: int,
    random_seed: int = 0,
) -> tuple[tuple[str, str, MetricName], ...]:
    """Return one deterministic position- and carryover-balanced cell order.

    Even-sized matrices repeat after ``n`` order indexes. Odd-sized matrices
    larger than one repeat after ``2n`` indexes because every cyclic row is
    followed by a reversed cycle. A one-cell matrix has a one-row cycle.

    Args:
        cells: Unique ``(implementation, case, metric)`` matrix cells.
        order_index: One-based schedule row.
        random_seed: Non-negative seed for the stable base-label permutation.

    Returns:
        The cells in the scheduled execution order.

    Raises:
        TypeError: If an index or seed is not an integer.
        ValueError: If an index, seed, or cell is invalid, or cells repeat.
    """
    frozen = tuple(cells)
    for cell in frozen:
        if (
            not isinstance(cell, tuple)
            or len(cell) != 3
            or not all(isinstance(value, str) and value for value in cell)
            or cell[2] not in KNOWN_METRICS
        ):
            raise ValueError(f"Invalid benchmark matrix cell: {cell!r}.")
    indices = balanced_order_indices(
        [(implementation, case, metric) for implementation, case, metric in frozen],
        order_index=order_index,
        random_seed=random_seed,
    )
    return tuple(frozen[index] for index in indices)

balanced_order_cycle_length

balanced_order_cycle_length(cell_count: int) -> int

Return the number of rows in a complete balanced-order cycle.

Source code in src/benchmatrix/_collection_design.py

def balanced_order_cycle_length(cell_count: int) -> int:
    """Return the number of rows in a complete balanced-order cycle."""
    if isinstance(cell_count, bool) or not isinstance(cell_count, int):
        raise TypeError("cell_count must be an integer.")
    if cell_count < 0:
        raise ValueError("cell_count must be non-negative.")
    if cell_count <= 1 or cell_count % 2 == 0:
        return cell_count
    return cell_count * 2

balanced_order_supercycle_length

balanced_order_supercycle_length(cell_count: int) -> int

Return the AB/BA-by-balanced-row joint-design cycle length.

Source code in src/benchmatrix/_collection_design.py

def balanced_order_supercycle_length(cell_count: int) -> int:
    """Return the AB/BA-by-balanced-row joint-design cycle length."""
    return 2 * balanced_order_cycle_length(cell_count)

collect_benchmark_runs

collect_benchmark_runs(
    command: Sequence[str],
    output_dir: str | Path,
    *,
    run_count: int | None = None,
    resume: bool = False,
    retry_failed: bool = False,
) -> BenchmarkRunGroup

Execute or resume pytest runs and persist a run-group manifest.

--benchmark-json is injected once per attempt. Attempts run sequentially and collection continues after failures so the manifest preserves complete lifecycle diagnostics. Resuming fills initial attempts that were never recorded. Retrying appends attempts until each currently missing successful run has received one new attempt; prior failures are never replaced.

Parameters:

Name	Type	Description	Default
command	Sequence[str]	Pytest command and arguments without --benchmark-json. May be empty when resuming, in which case the manifest command is reused. A supplied resume command must exactly match the manifest.	required
output_dir	str | Path	New or empty directory for a new collection, or an existing collection directory when resuming.	required
run_count	int | None	Successful-run target. New collections default to five. When resuming, an omitted value preserves the manifest target and a supplied value must match it.	None
resume	bool	Continue an existing manifest-backed collection.	False
retry_failed	bool	After resuming unattempted slots, append one new attempt for each successful run still needed. Requires resume.	False

Returns:

Type	Description
BenchmarkRunGroup	The completed collection, including successful runs and failed records.

Raises:

Type	Description
BenchmarkCollectionError	If collection cannot be initialized or resumed, or the command and requested count do not match.

Source code in src/benchmatrix/bench_collection.py

def collect_benchmark_runs(
    command: Sequence[str],
    output_dir: str | Path,
    *,
    run_count: int | None = None,
    resume: bool = False,
    retry_failed: bool = False,
) -> BenchmarkRunGroup:
    """Execute or resume pytest runs and persist a run-group manifest.

    ``--benchmark-json`` is injected once per attempt. Attempts run
    sequentially and collection continues after failures so the manifest
    preserves complete lifecycle diagnostics. Resuming fills initial attempts
    that were never recorded. Retrying appends attempts until each currently
    missing successful run has received one new attempt; prior failures are
    never replaced.

    Args:
        command: Pytest command and arguments without ``--benchmark-json``.
            May be empty when resuming, in which case the manifest command is
            reused. A supplied resume command must exactly match the manifest.
        output_dir: New or empty directory for a new collection, or an existing
            collection directory when resuming.
        run_count: Successful-run target. New collections default to five.
            When resuming, an omitted value preserves the manifest target and
            a supplied value must match it.
        resume: Continue an existing manifest-backed collection.
        retry_failed: After resuming unattempted slots, append one new attempt
            for each successful run still needed. Requires ``resume``.

    Returns:
        The completed collection, including successful runs and failed records.

    Raises:
        BenchmarkCollectionError: If collection cannot be initialized or
            resumed, or the command and requested count do not match.
    """
    if retry_failed and not resume:
        raise BenchmarkCollectionError("retry_failed requires resume=True.")
    if run_count is not None and (isinstance(run_count, bool) or not isinstance(run_count, int) or run_count <= 0):
        raise BenchmarkCollectionError("run_count must be a positive integer.")

    output = Path(output_dir).resolve()
    if resume:
        group = _load_resumable_group(output)
        normalized_command = _resume_command(command, group)
        requested_runs = group.requested_runs
        if run_count is not None and run_count != requested_runs:
            raise BenchmarkCollectionError(
                f"run_count {run_count} does not match the manifest target {requested_runs}."
            )
        manifest_path = group.manifest_path
        created_at = group.created_at
        cwd = group.cwd
        if not cwd.is_dir():
            raise BenchmarkCollectionError(f"Collection working directory is unavailable: {cwd}")
        records = list(group.records)
        runs = list(group.runs)
        expected_cells = group.expected_cells
        anchor_commit = group.commit
        anchor_fingerprint = group.environment_fingerprint
    else:
        normalized_command = _validate_collection_command(command)
        requested_runs = 5 if run_count is None else run_count
        _initialize_output_directory(output)
        manifest_path = output / RUN_GROUP_MANIFEST
        created_at = _utc_now()
        cwd = Path.cwd().resolve()
        records = []
        runs = []
        expected_cells = ()
        anchor_commit = None
        anchor_fingerprint = None
        _write_manifest(
            manifest_path,
            command=normalized_command,
            created_at=created_at,
            cwd=cwd,
            commit=anchor_commit,
            environment_fingerprint=anchor_fingerprint,
            expected_cells=expected_cells,
            requested_runs=requested_runs,
            records=records,
        )

    def execute_attempts(count: int) -> None:
        """Execute and persist a bounded set of collection attempts."""
        nonlocal expected_cells, anchor_commit, anchor_fingerprint
        for _ in range(count):
            index = len(records) + 1
            result_path = _available_result_path(output, index)
            started_at = _utc_now()
            started = time.monotonic()
            returncode: int | None = None
            error: str | None = None
            warnings: tuple[str, ...] = ()
            run: BenchmarkRun | None = None
            commit: str | None = None
            fingerprint: str | None = None

            try:
                # The CLI intentionally executes user-supplied argv without a shell.
                argv = [*normalized_command, f"--benchmark-json={result_path}"]
                command_stdout = _COLLECTION_COMMAND_STDOUT.get()
                if command_stdout is None:
                    completed = subprocess.run(argv, check=False, cwd=cwd)  # nosec B603
                else:
                    completed = subprocess.run(  # nosec B603
                        argv,
                        check=False,
                        cwd=cwd,
                        stdout=subprocess.PIPE,
                        text=True,
                    )
                    if completed.stdout:
                        _ = command_stdout.write(completed.stdout)
                        command_stdout.flush()
                returncode = completed.returncode
                if returncode != 0:
                    error = f"Benchmark command exited with status {returncode}."
                else:
                    run = load_benchmark_run(result_path)
                    commit = _run_commit(run)
                    fingerprint = _environment_fingerprint(run)
                    if runs:
                        error, warnings = _validate_collected_run(
                            runs[0],
                            run,
                            expected_cells=expected_cells,
                            anchor_commit=anchor_commit,
                            commit=commit,
                        )
                    if error is None and not runs:
                        expected_cells = _run_cells(run)
                        anchor_commit = commit
                        anchor_fingerprint = fingerprint
            except (OSError, BenchmarkJsonError, TypeError, ValueError) as exc:
                error = str(exc)

            duration = time.monotonic() - started
            if error is None and run is not None and fingerprint is not None:
                runs.append(run)
                record = BenchmarkRunRecord(
                    index=index,
                    status="succeeded",
                    path=result_path,
                    returncode=returncode,
                    started_at=started_at,
                    duration_seconds=duration,
                    warnings=warnings,
                    commit=commit,
                    environment_fingerprint=fingerprint,
                )
            else:
                record = BenchmarkRunRecord(
                    index=index,
                    status="failed",
                    path=result_path,
                    returncode=returncode,
                    started_at=started_at,
                    duration_seconds=duration,
                    error=error or "Benchmark command did not produce a valid run.",
                )
            records.append(record)
            _write_manifest(
                manifest_path,
                command=normalized_command,
                created_at=created_at,
                cwd=cwd,
                commit=anchor_commit,
                environment_fingerprint=anchor_fingerprint,
                expected_cells=expected_cells,
                requested_runs=requested_runs,
                records=records,
            )

    execute_attempts(max(0, requested_runs - len(records)))
    if retry_failed:
        execute_attempts(requested_runs - len(runs))

    return BenchmarkRunGroup(
        runs=tuple(runs),
        records=tuple(records),
        command=normalized_command,
        created_at=created_at,
        cwd=cwd,
        commit=anchor_commit,
        environment_fingerprint=anchor_fingerprint,
        expected_cells=expected_cells,
        requested_runs=requested_runs,
        manifest_path=manifest_path,
    )

collect_paired_benchmark_runs

collect_paired_benchmark_runs(
    baseline_command: Sequence[str],
    candidate_command: Sequence[str],
    output_dir: str | Path,
    *,
    pair_count: int | None = None,
    random_seed: int | None = None,
    baseline_cwd: str | Path | None = None,
    candidate_cwd: str | Path | None = None,
    resume: bool = False,
    retry_failed: bool = False,
) -> BenchmarkPairedRunGroup

Collect adjacent paired runs using a deterministic AB/BA schedule.

Every target pair is one atomic two-command block. AB and BA orientations alternate, with random_seed choosing the first orientation. Both variants use the same balanced Williams-style matrix order. A failed or interrupted block contributes no pair; retrying reruns both variants as a new adjacent block attempt while retaining all earlier records.

Parameters:

Name	Type	Description	Default
baseline_command	Sequence[str]	Baseline pytest command without --benchmark-json.	required
candidate_command	Sequence[str]	Candidate pytest command without --benchmark-json.	required
output_dir	str | Path	New collection directory, or an existing one when resuming.	required
pair_count	int | None	Complete-pair target. When omitted, collection starts with a provisional target of six and expands after matrix discovery to the smallest complete joint supercycle meeting the five-pair evidence default.	None
random_seed	int | None	Deterministic AB/BA and matrix-order seed. New collections default to zero; an omitted resume value preserves the manifest.	None
baseline_cwd	str | Path | None	Baseline child working directory. Defaults to the current directory for a new collection and the manifest value on resume.	None
candidate_cwd	str | Path | None	Candidate child working directory, with the same rules.	None
resume	bool	Continue a manifest-backed paired collection.	False
retry_failed	bool	Append one atomic block attempt for every pair still incomplete after interrupted work is resumed. Requires resume.	False

Returns:

Type	Description
BenchmarkPairedRunGroup	The paired collection with complete pairs and full lifecycle records.

Raises:

Type	Description
BenchmarkCollectionError	If collection cannot be initialized or resumed, or supplied settings disagree with the manifest.

Source code in src/benchmatrix/bench_collection.py

def collect_paired_benchmark_runs(
    baseline_command: Sequence[str],
    candidate_command: Sequence[str],
    output_dir: str | Path,
    *,
    pair_count: int | None = None,
    random_seed: int | None = None,
    baseline_cwd: str | Path | None = None,
    candidate_cwd: str | Path | None = None,
    resume: bool = False,
    retry_failed: bool = False,
) -> BenchmarkPairedRunGroup:
    """Collect adjacent paired runs using a deterministic AB/BA schedule.

    Every target pair is one atomic two-command block. AB and BA orientations
    alternate, with ``random_seed`` choosing the first orientation. Both
    variants use the same balanced Williams-style matrix order. A failed or
    interrupted block contributes no pair; retrying reruns both variants as a
    new adjacent block attempt while retaining all earlier records.

    Args:
        baseline_command: Baseline pytest command without ``--benchmark-json``.
        candidate_command: Candidate pytest command without
            ``--benchmark-json``.
        output_dir: New collection directory, or an existing one when resuming.
        pair_count: Complete-pair target. When omitted, collection starts with
            a provisional target of six and expands after matrix discovery to
            the smallest complete joint supercycle meeting the five-pair
            evidence default.
        random_seed: Deterministic AB/BA and matrix-order seed. New collections
            default to zero; an omitted resume value preserves the manifest.
        baseline_cwd: Baseline child working directory. Defaults to the current
            directory for a new collection and the manifest value on resume.
        candidate_cwd: Candidate child working directory, with the same rules.
        resume: Continue a manifest-backed paired collection.
        retry_failed: Append one atomic block attempt for every pair still
            incomplete after interrupted work is resumed. Requires ``resume``.

    Returns:
        The paired collection with complete pairs and full lifecycle records.

    Raises:
        BenchmarkCollectionError: If collection cannot be initialized or
            resumed, or supplied settings disagree with the manifest.
    """
    if retry_failed and not resume:
        raise BenchmarkCollectionError("retry_failed requires resume=True.")
    if pair_count is not None and (isinstance(pair_count, bool) or not isinstance(pair_count, int) or pair_count <= 0):
        raise BenchmarkCollectionError("pair_count must be a positive integer.")
    if random_seed is not None and (
        isinstance(random_seed, bool) or not isinstance(random_seed, int) or random_seed < 0
    ):
        raise BenchmarkCollectionError("random_seed must be a non-negative integer.")

    output = Path(output_dir).resolve()
    if resume:
        group = _load_resumable_paired_group(output)
        normalized_baseline = _resume_variant_command(baseline_command, group.baseline_command, variant="baseline")
        normalized_candidate = _resume_variant_command(candidate_command, group.candidate_command, variant="candidate")
        requested_pairs = group.requested_pairs
        resolved_seed = group.random_seed
        if pair_count is not None and pair_count != requested_pairs:
            raise BenchmarkCollectionError(
                f"pair_count {pair_count} does not match the manifest target {requested_pairs}."
            )
        if random_seed is not None and random_seed != resolved_seed:
            raise BenchmarkCollectionError(
                f"random_seed {random_seed} does not match the manifest seed {resolved_seed}."
            )
        resolved_baseline_cwd = _resume_variant_cwd(baseline_cwd, group.baseline_cwd, variant="baseline")
        resolved_candidate_cwd = _resume_variant_cwd(candidate_cwd, group.candidate_cwd, variant="candidate")
        manifest_path = group.manifest_path
        created_at = group.created_at
        records = list(group.records)
        runs = list(group.runs)
        expected_cells = group.expected_cells
        baseline_commit = group.baseline_commit
        candidate_commit = group.candidate_commit
        baseline_fingerprint = group.baseline_environment_fingerprint
        candidate_fingerprint = group.candidate_environment_fingerprint
        automatic_pairs = group.automatic_pairs
    else:
        normalized_baseline = _validate_collection_command(baseline_command)
        normalized_candidate = _validate_collection_command(candidate_command)
        automatic_pairs = pair_count is None
        requested_pairs = _PROVISIONAL_AUTOMATIC_PAIRS if automatic_pairs else cast(int, pair_count)
        resolved_seed = 0 if random_seed is None else random_seed
        resolved_baseline_cwd = _resolve_new_collection_cwd(baseline_cwd, variant="baseline")
        resolved_candidate_cwd = _resolve_new_collection_cwd(candidate_cwd, variant="candidate")
        _initialize_output_directory(output)
        manifest_path = output / RUN_GROUP_MANIFEST
        created_at = _utc_now()
        records = []
        runs = []
        expected_cells = ()
        baseline_commit = None
        candidate_commit = None
        baseline_fingerprint = None
        candidate_fingerprint = None

    def write_manifest() -> None:
        """Persist current paired state after every completed command."""
        _write_paired_manifest(
            manifest_path,
            baseline_command=normalized_baseline,
            candidate_command=normalized_candidate,
            created_at=created_at,
            baseline_cwd=resolved_baseline_cwd,
            candidate_cwd=resolved_candidate_cwd,
            baseline_commit=baseline_commit,
            candidate_commit=candidate_commit,
            baseline_environment_fingerprint=baseline_fingerprint,
            candidate_environment_fingerprint=candidate_fingerprint,
            expected_cells=expected_cells,
            requested_pairs=requested_pairs,
            automatic_pairs=automatic_pairs,
            random_seed=resolved_seed,
            records=records,
        )

    if not resume:
        write_manifest()

    def execute_block(entry: BenchmarkPairSchedule, block_attempt: int) -> None:
        """Execute and persist one complete adjacent AB/BA block attempt."""
        nonlocal expected_cells, baseline_commit, candidate_commit
        nonlocal baseline_fingerprint, candidate_fingerprint
        nonlocal requested_pairs
        for variant in entry.variants:
            command = normalized_baseline if variant == "baseline" else normalized_candidate
            cwd = resolved_baseline_cwd if variant == "baseline" else resolved_candidate_cwd
            index = len(records) + 1
            result_path = _available_paired_result_path(
                output,
                pair_index=entry.pair_index,
                block_attempt=block_attempt,
                variant=variant,
            )
            started_at = _utc_now()
            started = time.monotonic()
            returncode: int | None = None
            error: str | None = None
            warnings: tuple[str, ...] = ()
            run: BenchmarkRun | None = None
            commit: str | None = None
            fingerprint: str | None = None
            try:
                argv = [*command, f"--benchmark-json={result_path}"]
                command_stdout = _COLLECTION_COMMAND_STDOUT.get()
                with _collection_order_environment(
                    random_seed=resolved_seed,
                    order_index=entry.cell_order_index,
                ):
                    if command_stdout is None:
                        completed = subprocess.run(argv, check=False, cwd=cwd)  # nosec B603
                    else:
                        completed = subprocess.run(  # nosec B603
                            argv,
                            check=False,
                            cwd=cwd,
                            stdout=subprocess.PIPE,
                            text=True,
                        )
                        if completed.stdout:
                            _ = command_stdout.write(completed.stdout)
                            command_stdout.flush()
                returncode = completed.returncode
                if returncode != 0:
                    error = f"Benchmark command exited with status {returncode}."
                else:
                    run = load_benchmark_run(result_path)
                    candidate_cells = expected_cells or _run_cells(run)
                    scheduled_cells = balanced_cell_order(
                        candidate_cells,
                        order_index=entry.cell_order_index,
                        random_seed=resolved_seed,
                    )
                    if _run_cell_order(run) != scheduled_cells:
                        error = "Benchmark matrix cells were not executed in the scheduled balanced order."
                    commit = _run_commit(run)
                    fingerprint = _environment_fingerprint(run)
                    anchor_commit = baseline_commit if variant == "baseline" else candidate_commit
                    if error is None and anchor_commit is not None and commit != anchor_commit:
                        error = f"{variant.capitalize()} commit differs from its first successful run."
                    if error is None and runs:
                        error, warnings = _validate_paired_collected_run(
                            runs[0],
                            run,
                            expected_cells=candidate_cells,
                        )
                    if error is None and not expected_cells:
                        expected_cells = candidate_cells
                        if automatic_pairs:
                            requested_pairs = _automatic_pair_target(len(expected_cells))
                    if error is None and variant == "baseline" and baseline_fingerprint is None:
                        baseline_commit = commit
                        baseline_fingerprint = fingerprint
                    if error is None and variant == "candidate" and candidate_fingerprint is None:
                        candidate_commit = commit
                        candidate_fingerprint = fingerprint
            except (OSError, BenchmarkJsonError, TypeError, ValueError) as exc:
                error = str(exc)

            duration = time.monotonic() - started
            common = {
                "index": index,
                "pair_index": entry.pair_index,
                "block_attempt": block_attempt,
                "variant": variant,
                "pair_order": entry.pair_order,
                "order_position": _variant_order_position(variant, entry.pair_order),
                "cell_order_index": entry.cell_order_index,
                "path": result_path,
                "returncode": returncode,
                "started_at": started_at,
                "duration_seconds": duration,
            }
            if error is None and run is not None and fingerprint is not None:
                runs.append(run)
                record = BenchmarkPairedRunRecord(
                    **common,
                    status="succeeded",
                    warnings=warnings,
                    commit=commit,
                    environment_fingerprint=fingerprint,
                )
            else:
                record = BenchmarkPairedRunRecord(
                    **common,
                    status="failed",
                    error=error or "Benchmark command did not produce a valid run.",
                )
            records.append(record)
            write_manifest()

    attempted_pairs_this_call: set[int] = set()

    # The first target pair establishes the matrix size. Until at least one
    # command succeeds, no later pair has a well-defined balanced-order row.
    if not expected_cells:
        first_entry = make_paired_ab_ba_schedule(
            requested_pairs,
            random_seed=resolved_seed,
        )[0]
        first_records = [record for record in records if record.pair_index == 1]
        should_attempt_first = (
            not first_records
            or _latest_block_is_partial(first_records)
            or (retry_failed and not _pair_has_complete_block(first_records))
        )
        if should_attempt_first:
            next_attempt = max((record.block_attempt for record in first_records), default=0) + 1
            execute_block(first_entry, next_attempt)
            attempted_pairs_this_call.add(1)

    # Start every never-attempted target pair only after learning cell_count.
    # A partially persisted block is abandoned and replaced by a new adjacent
    # atomic block before collection advances.
    schedule = (
        make_paired_ab_ba_schedule(
            requested_pairs,
            random_seed=resolved_seed,
            cell_count=len(expected_cells),
        )
        if expected_cells
        else ()
    )
    for entry in schedule:
        pair_records = [record for record in records if record.pair_index == entry.pair_index]
        if not pair_records:
            execute_block(entry, 1)
            attempted_pairs_this_call.add(entry.pair_index)
        elif not _pair_has_complete_block(pair_records) and _latest_block_is_partial(pair_records):
            execute_block(entry, max(record.block_attempt for record in pair_records) + 1)
            attempted_pairs_this_call.add(entry.pair_index)

    if retry_failed:
        for entry in schedule:
            pair_records = [record for record in records if record.pair_index == entry.pair_index]
            if entry.pair_index not in attempted_pairs_this_call and not _pair_has_complete_block(pair_records):
                next_attempt = max((record.block_attempt for record in pair_records), default=0) + 1
                execute_block(entry, next_attempt)

    return BenchmarkPairedRunGroup(
        runs=tuple(runs),
        records=tuple(records),
        baseline_command=normalized_baseline,
        candidate_command=normalized_candidate,
        created_at=created_at,
        baseline_cwd=resolved_baseline_cwd,
        candidate_cwd=resolved_candidate_cwd,
        baseline_commit=baseline_commit,
        candidate_commit=candidate_commit,
        baseline_environment_fingerprint=baseline_fingerprint,
        candidate_environment_fingerprint=candidate_fingerprint,
        expected_cells=expected_cells,
        requested_pairs=requested_pairs,
        random_seed=resolved_seed,
        manifest_path=manifest_path,
        automatic_pairs=automatic_pairs,
    )

load_benchmark_run_group

load_benchmark_run_group(
    path: str | Path,
) -> BenchmarkRunGroup

Load a repeated-run collection from a manifest or its directory.

Parameters:

Name	Type	Description	Default
path	str | Path	Collection directory or benchmatrix-manifest.json path.	required

Returns:

Type	Description
BenchmarkRunGroup	A validated run group. Failed attempts are retained as records but do
BenchmarkRunGroup	not appear in runs.

Raises:

Type	Description
BenchmarkJsonError	If the manifest or a successful run is invalid.

Source code in src/benchmatrix/bench_collection.py

def load_benchmark_run_group(path: str | Path) -> BenchmarkRunGroup:
    """Load a repeated-run collection from a manifest or its directory.

    Args:
        path: Collection directory or ``benchmatrix-manifest.json`` path.

    Returns:
        A validated run group. Failed attempts are retained as records but do
        not appear in ``runs``.

    Raises:
        BenchmarkJsonError: If the manifest or a successful run is invalid.
    """
    source = Path(path)
    manifest_path = source / RUN_GROUP_MANIFEST if source.is_dir() else source

    try:
        payload = cast(object, json.loads(manifest_path.read_text(encoding="utf-8")))
    except OSError as exc:
        raise BenchmarkJsonError(f"Could not read benchmark run-group manifest: {manifest_path}") from exc
    except json.JSONDecodeError as exc:
        raise BenchmarkJsonError(f"Invalid JSON in benchmark run-group manifest: {manifest_path}") from exc

    root = _require_mapping(payload, path="manifest")
    _require_exact_keys(root, _ROOT_KEYS, path="manifest")
    if _require_string(root["producer"], path="manifest.producer") != PRODUCER:
        raise BenchmarkJsonError("Unsupported producer in benchmark run-group manifest.")
    if _require_string(root["kind"], path="manifest.kind") != RUN_GROUP_KIND:
        raise BenchmarkJsonError("Unsupported benchmark run-group manifest kind.")
    schema_version = _require_int(root["schema_version"], path="manifest.schema_version")
    if schema_version not in RUN_GROUP_SCHEMA_READ_VERSIONS:
        raise BenchmarkJsonError("Unsupported benchmark run-group manifest schema version.")

    created_at = _require_string(root["created_at"], path="manifest.created_at")
    _validate_manifest_timestamp(created_at, path="manifest.created_at")
    command = tuple(_require_string_list(root["command"], path="manifest.command", non_empty=True))
    cwd = Path(_require_string(root["cwd"], path="manifest.cwd"))
    commit = _require_optional_string(root["commit"], path="manifest.commit")
    fingerprint = _require_optional_string(
        root["environment_fingerprint"],
        path="manifest.environment_fingerprint",
    )
    if fingerprint is not None:
        _validate_manifest_fingerprint(fingerprint, path="manifest.environment_fingerprint")
    requested_runs = _require_positive_int(root["requested_runs"], path="manifest.requested_runs")
    expected_cells = _parse_cells(root["expected_cells"])
    record_payloads = _require_list(root["runs"], path="manifest.runs")

    records: list[BenchmarkRunRecord] = []
    runs: list[BenchmarkRun] = []
    record_paths: set[Path] = set()
    for position, record_payload in enumerate(record_payloads):
        record_path = f"manifest.runs[{position}]"
        record_mapping = _require_mapping(record_payload, path=record_path)
        _require_exact_keys(record_mapping, _RECORD_KEYS, path=record_path)
        record = _parse_record(record_mapping, manifest_path=manifest_path, path=record_path)
        canonical_record_path = record.path.resolve()
        if canonical_record_path in record_paths:
            raise BenchmarkJsonError("Manifest run paths must be unique.")
        record_paths.add(canonical_record_path)
        records.append(record)
        if record.status == "succeeded":
            run = load_benchmark_run(record.path)
            _validate_loaded_run(
                run,
                record=record,
                expected_cells=expected_cells,
                anchor_commit=commit,
                anchor_fingerprint=fingerprint,
            )
            runs.append(run)

    successful_records = tuple(record for record in records if record.status == "succeeded")
    if schema_version == 1 and len(records) > requested_runs:
        raise BenchmarkJsonError("Version 1 run-group manifests cannot contain retry attempts.")
    if successful_records and successful_records[0].environment_fingerprint != fingerprint:
        raise BenchmarkJsonError("First successful run environment fingerprint does not match the manifest anchor.")
    if runs:
        compatibility = compare_benchmark_run_groups(
            (runs[0],),
            tuple(runs[1:]) or (runs[0],),
            compatibility_policy=RunCompatibilityPolicy(mode="permissive"),
            evidence_policy=EvidencePolicy(minimum_runs=1, minimum_samples_per_run=0),
            inference_policy=InferencePolicy(method="legacy_consistency"),
        ).compatibility
        if compatibility.blocking:
            fields = ", ".join(finding.field for finding in compatibility.blocking)
            raise BenchmarkJsonError(f"Manifest contains incompatible successful environments: {fields}.")

    try:
        return BenchmarkRunGroup(
            runs=tuple(runs),
            records=tuple(records),
            command=command,
            created_at=created_at,
            cwd=cwd,
            commit=commit,
            environment_fingerprint=fingerprint,
            expected_cells=expected_cells,
            requested_runs=requested_runs,
            manifest_path=manifest_path,
        )
    except (TypeError, ValueError) as exc:
        raise BenchmarkJsonError(f"Invalid benchmark run-group manifest: {exc}") from exc

load_paired_benchmark_run_group

load_paired_benchmark_run_group(
    path: str | Path,
) -> BenchmarkPairedRunGroup

Load and validate a paired AB/BA collection manifest.

Parameters:

Name	Type	Description	Default
path	str | Path	Collection directory or benchmatrix-manifest.json path.	required

Returns:

Type	Description
BenchmarkPairedRunGroup	A paired collection whose complete pairs contain only atomic blocks in
BenchmarkPairedRunGroup	which both scheduled commands succeeded.

Raises:

Type	Description
BenchmarkJsonError	If the manifest or a successful run is invalid.

Source code in src/benchmatrix/bench_collection.py

def load_paired_benchmark_run_group(path: str | Path) -> BenchmarkPairedRunGroup:
    """Load and validate a paired AB/BA collection manifest.

    Args:
        path: Collection directory or ``benchmatrix-manifest.json`` path.

    Returns:
        A paired collection whose complete pairs contain only atomic blocks in
        which both scheduled commands succeeded.

    Raises:
        BenchmarkJsonError: If the manifest or a successful run is invalid.
    """
    source = Path(path)
    manifest_path = source / RUN_GROUP_MANIFEST if source.is_dir() else source
    try:
        payload = cast(object, json.loads(manifest_path.read_text(encoding="utf-8")))
    except OSError as exc:
        raise BenchmarkJsonError(f"Could not read paired benchmark manifest: {manifest_path}") from exc
    except json.JSONDecodeError as exc:
        raise BenchmarkJsonError(f"Invalid JSON in paired benchmark manifest: {manifest_path}") from exc

    root = _require_mapping(payload, path="manifest")
    _require_exact_keys(root, _PAIRED_ROOT_KEYS, path="manifest")
    if _require_string(root["producer"], path="manifest.producer") != PRODUCER:
        raise BenchmarkJsonError("Unsupported producer in paired benchmark manifest.")
    if _require_string(root["kind"], path="manifest.kind") != PAIRED_RUN_GROUP_KIND:
        raise BenchmarkJsonError("Unsupported paired benchmark manifest kind.")
    schema_version = _require_int(root["schema_version"], path="manifest.schema_version")
    if schema_version not in PAIRED_RUN_GROUP_SCHEMA_READ_VERSIONS:
        raise BenchmarkJsonError("Unsupported paired benchmark manifest schema version.")

    created_at = _require_string(root["created_at"], path="manifest.created_at")
    _validate_manifest_timestamp(created_at, path="manifest.created_at")
    baseline_command = tuple(
        _require_string_list(root["baseline_command"], path="manifest.baseline_command", non_empty=True)
    )
    candidate_command = tuple(
        _require_string_list(root["candidate_command"], path="manifest.candidate_command", non_empty=True)
    )
    baseline_cwd = Path(_require_non_empty_string(root["baseline_cwd"], path="manifest.baseline_cwd"))
    candidate_cwd = Path(_require_non_empty_string(root["candidate_cwd"], path="manifest.candidate_cwd"))
    baseline_commit = _require_optional_string(root["baseline_commit"], path="manifest.baseline_commit")
    candidate_commit = _require_optional_string(root["candidate_commit"], path="manifest.candidate_commit")
    baseline_fingerprint = _parse_optional_manifest_fingerprint(
        root["baseline_environment_fingerprint"],
        path="manifest.baseline_environment_fingerprint",
    )
    candidate_fingerprint = _parse_optional_manifest_fingerprint(
        root["candidate_environment_fingerprint"],
        path="manifest.candidate_environment_fingerprint",
    )
    requested_pairs = _require_positive_int(root["requested_pairs"], path="manifest.requested_pairs")
    automatic_pairs_value = root["automatic_pairs"]
    if not isinstance(automatic_pairs_value, bool):
        raise BenchmarkJsonError("Expected boolean at manifest.automatic_pairs.")
    random_seed = _require_non_negative_int(root["random_seed"], path="manifest.random_seed")
    expected_cells = _parse_cells(root["expected_cells"])

    records: list[BenchmarkPairedRunRecord] = []
    runs: list[BenchmarkRun] = []
    record_paths: set[Path] = set()
    for position, record_payload in enumerate(_require_list(root["runs"], path="manifest.runs")):
        record_path = f"manifest.runs[{position}]"
        record_mapping = _require_mapping(record_payload, path=record_path)
        _require_exact_keys(record_mapping, _PAIRED_RECORD_KEYS, path=record_path)
        record = _parse_paired_record(record_mapping, manifest_path=manifest_path, path=record_path)
        canonical_record_path = record.path.resolve()
        if canonical_record_path in record_paths:
            raise BenchmarkJsonError("Paired manifest run paths must be unique.")
        record_paths.add(canonical_record_path)
        records.append(record)
        if record.status == "succeeded":
            run = load_benchmark_run(record.path)
            _validate_loaded_paired_run(
                run,
                record=record,
                expected_cells=expected_cells,
                random_seed=random_seed,
            )
            runs.append(run)

    if runs:
        compatibility = compare_benchmark_run_groups(
            (runs[0],),
            tuple(runs[1:]) or (runs[0],),
            compatibility_policy=RunCompatibilityPolicy(mode="permissive"),
            evidence_policy=EvidencePolicy(minimum_runs=1, minimum_samples_per_run=0),
            inference_policy=InferencePolicy(method="legacy_consistency"),
        ).compatibility
        if compatibility.blocking:
            fields = ", ".join(finding.field for finding in compatibility.blocking)
            raise BenchmarkJsonError(f"Paired manifest contains incompatible successful environments: {fields}.")

    try:
        return BenchmarkPairedRunGroup(
            runs=tuple(runs),
            records=tuple(records),
            baseline_command=baseline_command,
            candidate_command=candidate_command,
            created_at=created_at,
            baseline_cwd=baseline_cwd,
            candidate_cwd=candidate_cwd,
            baseline_commit=baseline_commit,
            candidate_commit=candidate_commit,
            baseline_environment_fingerprint=baseline_fingerprint,
            candidate_environment_fingerprint=candidate_fingerprint,
            expected_cells=expected_cells,
            requested_pairs=requested_pairs,
            random_seed=random_seed,
            manifest_path=manifest_path,
            automatic_pairs=automatic_pairs_value,
        )
    except (TypeError, ValueError) as exc:
        raise BenchmarkJsonError(f"Invalid paired benchmark manifest: {exc}") from exc

make_paired_ab_ba_schedule

make_paired_ab_ba_schedule(
    pair_count: int,
    *,
    random_seed: int = 0,
    cell_count: int | None = None,
) -> tuple[BenchmarkPairSchedule, ...]

Return a deterministic joint AB/BA and balanced-row block schedule.

The seed chooses whether the first block is AB or BA. Later blocks alternate, so the counts differ by at most one for odd pair_count. Both members of a block use the same balanced cell-order row. When the matrix size is known, each row occurs once with each orientation over a complete joint supercycle. Omitting cell_count provides an orientation-only, single-row schedule for compatibility and collection before the matrix has been learned.

Parameters:

Name	Type	Description	Default
pair_count	int	Number of target baseline/candidate pairs.	required
random_seed	int	Non-negative deterministic schedule seed.	0
cell_count	int | None	Positive number of cells in the benchmark matrix, when known.	None

Returns:

Type	Description
tuple[BenchmarkPairSchedule, ...]	One schedule entry per requested pair.

Source code in src/benchmatrix/bench_collection.py

def make_paired_ab_ba_schedule(
    pair_count: int,
    *,
    random_seed: int = 0,
    cell_count: int | None = None,
) -> tuple[BenchmarkPairSchedule, ...]:
    """Return a deterministic joint AB/BA and balanced-row block schedule.

    The seed chooses whether the first block is AB or BA. Later blocks
    alternate, so the counts differ by at most one for odd ``pair_count``.
    Both members of a block use the same balanced cell-order row. When the
    matrix size is known, each row occurs once with each orientation over a
    complete joint supercycle. Omitting ``cell_count`` provides an
    orientation-only, single-row schedule for compatibility and collection
    before the matrix has been learned.

    Args:
        pair_count: Number of target baseline/candidate pairs.
        random_seed: Non-negative deterministic schedule seed.
        cell_count: Positive number of cells in the benchmark matrix, when
            known.

    Returns:
        One schedule entry per requested pair.
    """
    if isinstance(pair_count, bool) or not isinstance(pair_count, int) or pair_count <= 0:
        raise ValueError("pair_count must be a positive integer.")
    if isinstance(random_seed, bool) or not isinstance(random_seed, int):
        raise TypeError("random_seed must be an integer.")
    if random_seed < 0:
        raise ValueError("random_seed must be non-negative.")
    if cell_count is not None:
        if isinstance(cell_count, bool) or not isinstance(cell_count, int):
            raise TypeError("cell_count must be an integer or None.")
        if cell_count <= 0:
            raise ValueError("cell_count must be positive when provided.")

    seed_digest = hashlib.sha256(str(random_seed).encode()).digest()
    starts_with_ba = bool(seed_digest[0] & 1)
    row_cycle_length = balanced_order_cycle_length(cell_count) if cell_count is not None else 1
    return tuple(
        BenchmarkPairSchedule(
            pair_index=pair_index,
            pair_order=("BA" if starts_with_ba == (pair_index % 2 == 1) else "AB"),
            cell_order_index=(pair_index - 1) // 2 % row_cycle_length + 1,
        )
        for pair_index in range(1, pair_count + 1)
    )

compare_benchmark_run_groups

compare_benchmark_run_groups(
    baselines: Sequence[BenchmarkRun],
    candidates: Sequence[BenchmarkRun],
    *,
    compatibility_policy: RunCompatibilityPolicy
    | None = None,
    regression_policy: RegressionPolicy | None = None,
    evidence_policy: EvidencePolicy | None = None,
    inference_policy: InferencePolicy | None = None,
    precision_policy: PrecisionPolicy | None = None,
) -> BenchmarkRunComparison

Compare repeated baseline and candidate runs as two evidence groups.

Each cell uses the median of its per-run metric values. By default, run-level BCa bootstrap intervals quantify uncertainty and a Bonferroni adjustment controls the matrix-wide family-wise error rate. Practical thresholds then distinguish improvements, regressions, equivalence, and inconclusive intervals.

Parameters:

Name	Type	Description	Default
baselines	Sequence[BenchmarkRun]	Repeated reference benchmark runs.	required
candidates	Sequence[BenchmarkRun]	Repeated candidate benchmark runs.	required
compatibility_policy	RunCompatibilityPolicy | None	Environment checks applied across every run.	None
regression_policy	RegressionPolicy | None	Percentage thresholds for classifying changes.	None
evidence_policy	EvidencePolicy | None	Minimum repeated-run and sample evidence.	None
inference_policy	InferencePolicy | None	Statistical inference and multiplicity controls.	None
precision_policy	PrecisionPolicy | None	Optional paired fixed-design planning target. It must remain disabled for independent groups.	None

Returns:

Type	Description
BenchmarkRunComparison	A matrix comparison with per-side trust diagnostics.

Raises:

Type	Description
ValueError	If either run group is empty.
TypeError	If a group contains a value other than BenchmarkRun.

Source code in src/benchmatrix/bench_compare.py

def compare_benchmark_run_groups(
    baselines: Sequence[BenchmarkRun],
    candidates: Sequence[BenchmarkRun],
    *,
    compatibility_policy: RunCompatibilityPolicy | None = None,
    regression_policy: RegressionPolicy | None = None,
    evidence_policy: EvidencePolicy | None = None,
    inference_policy: InferencePolicy | None = None,
    precision_policy: PrecisionPolicy | None = None,
) -> BenchmarkRunComparison:
    """Compare repeated baseline and candidate runs as two evidence groups.

    Each cell uses the median of its per-run metric values. By default,
    run-level BCa bootstrap intervals quantify uncertainty and a Bonferroni
    adjustment controls the matrix-wide family-wise error rate. Practical
    thresholds then distinguish improvements, regressions, equivalence, and
    inconclusive intervals.

    Args:
        baselines: Repeated reference benchmark runs.
        candidates: Repeated candidate benchmark runs.
        compatibility_policy: Environment checks applied across every run.
        regression_policy: Percentage thresholds for classifying changes.
        evidence_policy: Minimum repeated-run and sample evidence.
        inference_policy: Statistical inference and multiplicity controls.
        precision_policy: Optional paired fixed-design planning target. It
            must remain disabled for independent groups.

    Returns:
        A matrix comparison with per-side trust diagnostics.

    Raises:
        ValueError: If either run group is empty.
        TypeError: If a group contains a value other than ``BenchmarkRun``.
    """
    return _compare_benchmark_run_groups(
        baselines,
        candidates,
        compatibility_policy=compatibility_policy,
        regression_policy=regression_policy,
        evidence_policy=evidence_policy,
        inference_policy=inference_policy,
        precision_policy=precision_policy,
        design="independent",
    )

compare_benchmark_runs

compare_benchmark_runs(
    baseline: BenchmarkRun,
    candidate: BenchmarkRun,
    *,
    compatibility_policy: RunCompatibilityPolicy
    | None = None,
    regression_policy: RegressionPolicy | None = None,
    inference_policy: InferencePolicy | None = None,
    precision_policy: PrecisionPolicy | None = None,
) -> BenchmarkRunComparison

Compare two runs across the union of their benchmark matrix cells.

Comparisons use mean latency for single_call_latency, mean throughput for batch_throughput, and p95 latency for tail_latency. Missing cells and changed case or unit metadata are retained as explicit results rather than silently dropped.

Parameters:

Name	Type	Description	Default
baseline	BenchmarkRun	Reference benchmark run.	required
candidate	BenchmarkRun	Benchmark run being evaluated.	required
compatibility_policy	RunCompatibilityPolicy | None	Environment checks to apply. Defaults to permissive compatibility.	None
regression_policy	RegressionPolicy | None	Thresholds used to classify cell changes. Defaults to a five-percent threshold.	None
inference_policy	InferencePolicy | None	Run-level uncertainty analysis to apply. A single run cannot produce a default bootstrap interval and is therefore inconclusive unless the legacy method is selected explicitly.	None
precision_policy	PrecisionPolicy | None	Optional precision planning. Planning requires an explicitly paired design and is rejected for this single-run API.	None

Returns:

Type	Description
BenchmarkRunComparison	A deterministic comparison across both run matrices.

Source code in src/benchmatrix/bench_compare.py

def compare_benchmark_runs(
    baseline: BenchmarkRun,
    candidate: BenchmarkRun,
    *,
    compatibility_policy: RunCompatibilityPolicy | None = None,
    regression_policy: RegressionPolicy | None = None,
    inference_policy: InferencePolicy | None = None,
    precision_policy: PrecisionPolicy | None = None,
) -> BenchmarkRunComparison:
    """Compare two runs across the union of their benchmark matrix cells.

    Comparisons use mean latency for ``single_call_latency``, mean throughput
    for ``batch_throughput``, and p95 latency for ``tail_latency``. Missing
    cells and changed case or unit metadata are retained as explicit results
    rather than silently dropped.

    Args:
        baseline: Reference benchmark run.
        candidate: Benchmark run being evaluated.
        compatibility_policy: Environment checks to apply. Defaults to
            permissive compatibility.
        regression_policy: Thresholds used to classify cell changes. Defaults
            to a five-percent threshold.
        inference_policy: Run-level uncertainty analysis to apply. A single
            run cannot produce a default bootstrap interval and is therefore
            inconclusive unless the legacy method is selected explicitly.
        precision_policy: Optional precision planning. Planning requires an
            explicitly paired design and is rejected for this single-run API.

    Returns:
        A deterministic comparison across both run matrices.
    """
    return compare_benchmark_run_groups(
        (baseline,),
        (candidate,),
        compatibility_policy=compatibility_policy,
        regression_policy=regression_policy,
        inference_policy=inference_policy,
        precision_policy=precision_policy,
        evidence_policy=EvidencePolicy(
            minimum_runs=1,
            minimum_samples_per_run=0,
            minimum_rounds_per_run=0,
            require_rounds=False,
            require_iterations=False,
            require_raw_samples_for_inference=False,
            minimum_tail_samples_per_run=0,
            require_tail_iterations_one=False,
        ),
    )

compare_paired_benchmark_run_groups

compare_paired_benchmark_run_groups(
    baselines: Sequence[BenchmarkRun],
    candidates: Sequence[BenchmarkRun],
    *,
    pair_strata: Sequence[str] | None = None,
    precision_pair_count_multiple: int = 2,
    compatibility_policy: RunCompatibilityPolicy
    | None = None,
    regression_policy: RegressionPolicy | None = None,
    evidence_policy: EvidencePolicy | None = None,
    inference_policy: InferencePolicy | None = None,
    precision_policy: PrecisionPolicy | None = None,
) -> BenchmarkRunComparison

Compare explicitly matched baseline/candidate process-run pairs.

The values at each position must come from one adjacent collection block. Complete pairs, rather than individual run files, are the independent experimental units. Pairing is explicit in this API and is never inferred from filenames or timestamps.

Parameters:

Name	Type	Description	Default
baselines	Sequence[BenchmarkRun]	Baseline members in pair order.	required
candidates	Sequence[BenchmarkRun]	Candidate members in the same pair order.	required
pair_strata	Sequence[str] | None	Optional fixed-design stratum label for each pair, such as its recorded AB or BA command orientation. When given, paired resampling preserves the observed count in every stratum.	None
precision_pair_count_multiple	int	Divisibility constraint for a future confirmatory collection. Paired designs default to an even count; manifest-backed collections pass their complete joint design supercycle.	2
compatibility_policy	RunCompatibilityPolicy | None	Environment checks applied across every run.	None
regression_policy	RegressionPolicy | None	Percentage thresholds for classifying changes.	None
evidence_policy	EvidencePolicy | None	Minimum complete-pair and sample evidence.	None
inference_policy	InferencePolicy | None	Statistical inference and multiplicity controls.	None
precision_policy	PrecisionPolicy | None	Optional fixed-design precision target for a fresh future paired collection.	None

Returns:

Type	Description
BenchmarkRunComparison	A paired matrix comparison with per-side trust diagnostics.

Raises:

Type	Description
ValueError	If the sequences are empty or have different lengths.
TypeError	If either sequence contains a non-BenchmarkRun value.

Source code in src/benchmatrix/bench_compare.py

def compare_paired_benchmark_run_groups(
    baselines: Sequence[BenchmarkRun],
    candidates: Sequence[BenchmarkRun],
    *,
    pair_strata: Sequence[str] | None = None,
    precision_pair_count_multiple: int = 2,
    compatibility_policy: RunCompatibilityPolicy | None = None,
    regression_policy: RegressionPolicy | None = None,
    evidence_policy: EvidencePolicy | None = None,
    inference_policy: InferencePolicy | None = None,
    precision_policy: PrecisionPolicy | None = None,
) -> BenchmarkRunComparison:
    """Compare explicitly matched baseline/candidate process-run pairs.

    The values at each position must come from one adjacent collection block.
    Complete pairs, rather than individual run files, are the independent
    experimental units. Pairing is explicit in this API and is never inferred
    from filenames or timestamps.

    Args:
        baselines: Baseline members in pair order.
        candidates: Candidate members in the same pair order.
        pair_strata: Optional fixed-design stratum label for each pair, such
            as its recorded ``AB`` or ``BA`` command orientation. When given,
            paired resampling preserves the observed count in every stratum.
        precision_pair_count_multiple: Divisibility constraint for a future
            confirmatory collection. Paired designs default to an even count;
            manifest-backed collections pass their complete joint design
            supercycle.
        compatibility_policy: Environment checks applied across every run.
        regression_policy: Percentage thresholds for classifying changes.
        evidence_policy: Minimum complete-pair and sample evidence.
        inference_policy: Statistical inference and multiplicity controls.
        precision_policy: Optional fixed-design precision target for a fresh
            future paired collection.

    Returns:
        A paired matrix comparison with per-side trust diagnostics.

    Raises:
        ValueError: If the sequences are empty or have different lengths.
        TypeError: If either sequence contains a non-``BenchmarkRun`` value.
    """
    if len(baselines) != len(candidates):
        raise ValueError("Paired baseline and candidate groups must contain the same number of runs.")
    return _compare_benchmark_run_groups(
        baselines,
        candidates,
        compatibility_policy=compatibility_policy,
        regression_policy=regression_policy,
        evidence_policy=evidence_policy,
        inference_policy=inference_policy,
        precision_policy=precision_policy,
        design="paired",
        pair_strata=pair_strata,
        precision_pair_count_multiple=precision_pair_count_multiple,
    )

benchmark_batch_throughput

benchmark_batch_throughput(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord

Benchmark batch throughput for one implementation and case.

Parameters:

Name	Type	Description	Default
benchmark	BenchmarkFixture	Pytest-benchmark fixture instance.	required
implementation_name	str	Name of the implementation under test.	required
function	TargetFunction	Synchronous function implementation to benchmark. The function must complete the measured work before returning.	required
case_name	str	Name of the input case under test.	required
case	BenchmarkCase	Benchmark input case. If case.work_units is provided, throughput is later derived as work units per second; otherwise it is derived as calls per second.	required
config	BenchmarkConfig | None	Benchmark harness configuration. Defaults to BenchmarkConfig().	None
stream	TextIO | None	Stream used for progress output. Defaults to sys.stdout when progress output is enabled.	None

Returns:

Type	Description
BenchmarkInvocationRecord	A lightweight invocation record containing metadata attached to the
BenchmarkInvocationRecord	benchmark. This is not a timing result.

Raises:

Type	Description
TypeError	If function is an async function.
ValueError	If case.work_units is not positive or finite.

Warning

Throughput is derived from one synchronous target invocation. It does not model concurrency, saturation, queueing, or service request load. case.work_units must accurately describe work completed by each target call.

Source code in src/benchmatrix/bench_harness.py

def benchmark_batch_throughput(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord:
    """Benchmark batch throughput for one implementation and case.

    Args:
        benchmark: Pytest-benchmark fixture instance.
        implementation_name: Name of the implementation under test.
        function: Synchronous function implementation to benchmark. The
            function must complete the measured work before returning.
        case_name: Name of the input case under test.
        case: Benchmark input case. If ``case.work_units`` is provided,
            throughput is later derived as work units per second; otherwise it
            is derived as calls per second.
        config: Benchmark harness configuration. Defaults to
            ``BenchmarkConfig()``.
        stream: Stream used for progress output. Defaults to ``sys.stdout``
            when progress output is enabled.

    Returns:
        A lightweight invocation record containing metadata attached to the
        benchmark. This is not a timing result.

    Raises:
        TypeError: If ``function`` is an async function.
        ValueError: If ``case.work_units`` is not positive or finite.

    Warning:
        Throughput is derived from one synchronous target invocation. It does
        not model concurrency, saturation, queueing, or service request load.
        ``case.work_units`` must accurately describe work completed by each
        target call.
    """
    resolved_config = _resolve_config(config)
    metric_name = METRIC_BATCH_THROUGHPUT
    extra_info: dict[str, object] = _make_base_extra_info(
        metric_name,
        implementation_name,
        case_name,
        case,
    )
    work_unit_count = case.work_unit_count()

    if work_unit_count is None:
        extra_info[KEY_THROUGHPUT_UNIT] = THROUGHPUT_UNIT_CALLS_PER_SECOND
    else:
        extra_info[KEY_WORK_UNITS] = work_unit_count
        extra_info[KEY_WORK_UNIT_NAME] = case.work_unit_name
        extra_info[KEY_THROUGHPUT_UNIT] = THROUGHPUT_UNIT_WORK_UNITS_PER_SECOND

    final_extra_info = _set_extra_info(benchmark, extra_info)
    _ = _run_target_with_hooks(
        benchmark,
        metric_name,
        implementation_name,
        function,
        case_name,
        case,
        config=resolved_config,
        force_pedantic=False,
    )

    record = BenchmarkInvocationRecord(
        metric_name=metric_name,
        implementation_name=implementation_name,
        case_name=case_name,
        extra_info=final_extra_info,
    )
    _maybe_display_invocation_record(record, config=resolved_config, stream=stream)
    return record

benchmark_single_call_latency

benchmark_single_call_latency(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord

Benchmark single-call latency for one implementation and case.

Parameters:

Name	Type	Description	Default
benchmark	BenchmarkFixture	Pytest-benchmark fixture instance.	required
implementation_name	str	Name of the implementation under test.	required
function	TargetFunction	Synchronous function implementation to benchmark. The function must complete the measured work before returning.	required
case_name	str	Name of the input case under test.	required
case	BenchmarkCase	Benchmark input case.	required
config	BenchmarkConfig | None	Benchmark harness configuration. Defaults to BenchmarkConfig().	None
stream	TextIO | None	Stream used for progress output. Defaults to sys.stdout when progress output is enabled.	None

Returns:

Type	Description
BenchmarkInvocationRecord	A lightweight invocation record containing metadata attached to the
BenchmarkInvocationRecord	benchmark. This is not a timing result.

Raises:

Type	Description
TypeError	If function is an async function.

Warning

This measures completed target-function work only. Input construction, lazy-result consumption, and other setup are excluded unless they occur inside function.

Source code in src/benchmatrix/bench_harness.py

def benchmark_single_call_latency(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord:
    """Benchmark single-call latency for one implementation and case.

    Args:
        benchmark: Pytest-benchmark fixture instance.
        implementation_name: Name of the implementation under test.
        function: Synchronous function implementation to benchmark. The
            function must complete the measured work before returning.
        case_name: Name of the input case under test.
        case: Benchmark input case.
        config: Benchmark harness configuration. Defaults to
            ``BenchmarkConfig()``.
        stream: Stream used for progress output. Defaults to ``sys.stdout``
            when progress output is enabled.

    Returns:
        A lightweight invocation record containing metadata attached to the
        benchmark. This is not a timing result.

    Raises:
        TypeError: If ``function`` is an async function.

    Warning:
        This measures completed target-function work only. Input construction,
        lazy-result consumption, and other setup are excluded unless they occur
        inside ``function``.
    """
    resolved_config = _resolve_config(config)
    metric_name = METRIC_SINGLE_CALL_LATENCY
    extra_info: dict[str, object] = _make_base_extra_info(
        metric_name,
        implementation_name,
        case_name,
        case,
    )
    final_extra_info = _set_extra_info(benchmark, extra_info)
    _ = _run_target_with_hooks(
        benchmark,
        metric_name,
        implementation_name,
        function,
        case_name,
        case,
        config=resolved_config,
        force_pedantic=False,
    )

    record = BenchmarkInvocationRecord(
        metric_name=metric_name,
        implementation_name=implementation_name,
        case_name=case_name,
        extra_info=final_extra_info,
    )
    _maybe_display_invocation_record(record, config=resolved_config, stream=stream)
    return record

benchmark_tail_latency

benchmark_tail_latency(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord

Benchmark latency distribution for one implementation and case.

Parameters:

Name	Type	Description	Default
benchmark	BenchmarkFixture	Pytest-benchmark fixture instance.	required
implementation_name	str	Name of the implementation under test.	required
function	TargetFunction	Synchronous function implementation to benchmark. The function must complete the measured work before returning.	required
case_name	str	Name of the input case under test.	required
case	BenchmarkCase	Benchmark input case.	required
config	BenchmarkConfig | None	Benchmark harness configuration. Defaults to BenchmarkConfig().	None
stream	TextIO | None	Stream used for progress output. Defaults to sys.stdout when progress output is enabled.	None

Returns:

Type	Description
BenchmarkInvocationRecord	A lightweight invocation record containing metadata attached to the
BenchmarkInvocationRecord	benchmark. This is not a timing result.

Raises:

Type	Description
TypeError	If function is an async function.

Warning

This uses pedantic mode. Tail percentiles should be calculated from pytest-benchmark JSON data values. This is an implementation-comparison metric, not production p95/p99 latency under load.

If case.fresh_inputs is false and config.pedantic_iterations is greater than one, raw samples are per-round averages of multiple calls, not individual-call latency samples. The harness emits a runtime warning for that configuration.

Source code in src/benchmatrix/bench_harness.py

def benchmark_tail_latency(
    benchmark: BenchmarkFixture,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord:
    """Benchmark latency distribution for one implementation and case.

    Args:
        benchmark: Pytest-benchmark fixture instance.
        implementation_name: Name of the implementation under test.
        function: Synchronous function implementation to benchmark. The
            function must complete the measured work before returning.
        case_name: Name of the input case under test.
        case: Benchmark input case.
        config: Benchmark harness configuration. Defaults to
            ``BenchmarkConfig()``.
        stream: Stream used for progress output. Defaults to ``sys.stdout``
            when progress output is enabled.

    Returns:
        A lightweight invocation record containing metadata attached to the
        benchmark. This is not a timing result.

    Raises:
        TypeError: If ``function`` is an async function.

    Warning:
        This uses pedantic mode. Tail percentiles should be calculated from
        pytest-benchmark JSON ``data`` values. This is an
        implementation-comparison metric, not production p95/p99 latency under
        load.

        If ``case.fresh_inputs`` is false and ``config.pedantic_iterations`` is
        greater than one, raw samples are per-round averages of multiple calls,
        not individual-call latency samples. The harness emits a runtime
        warning for that configuration.
    """
    resolved_config = _resolve_config(config)
    metric_name = METRIC_TAIL_LATENCY
    _warn_for_tail_latency_iteration_semantics(case, resolved_config)

    extra_info: dict[str, object] = _make_base_extra_info(
        metric_name,
        implementation_name,
        case_name,
        case,
    )
    extra_info[KEY_TAIL_LATENCY_NOTE] = (
        "Use pytest-benchmark JSON data to compute p50/p90/p95/p99. "
        "This is not production p95/p99 under load. If pedantic_iterations is "
        "greater than one, samples are per-round averages of multiple calls."
    )
    extra_info[KEY_TAIL_PERCENTILES] = list(TAIL_PERCENTILES)

    final_extra_info = _set_extra_info(benchmark, extra_info)
    _ = _run_target_with_hooks(
        benchmark,
        metric_name,
        implementation_name,
        function,
        case_name,
        case,
        config=resolved_config,
        force_pedantic=True,
    )

    record = BenchmarkInvocationRecord(
        metric_name=metric_name,
        implementation_name=implementation_name,
        case_name=case_name,
        extra_info=final_extra_info,
    )
    _maybe_display_invocation_record(record, config=resolved_config, stream=stream)
    return record

deep_copy

deep_copy(value: object) -> object

Return a deep copy of value.

Parameters:

Name	Type	Description	Default
value	object	Value to copy.	required

Returns:

Type	Description
object	A deep copy of value.

Source code in src/benchmatrix/bench_harness.py

def deep_copy(value: object) -> object:
    """Return a deep copy of ``value``.

    Args:
        value: Value to copy.

    Returns:
        A deep copy of ``value``.
    """
    return copy.deepcopy(value)

make_benchmark_parameters

make_benchmark_parameters(
    implementations: Mapping[str, TargetFunction],
    cases: Mapping[str, BenchmarkCase]
    | Iterable[BenchmarkCase],
    *,
    metrics: Iterable[MetricName] | None = None,
) -> list[object]

Create pytest parameters for a metric-by-implementation-by-case matrix.

Parameters:

Name	Type	Description	Default
implementations	Mapping[str, TargetFunction]	Mapping from implementation name to target function.	required
cases	Mapping[str, BenchmarkCase] | Iterable[BenchmarkCase]	Mapping or iterable of benchmark input cases.	required
metrics	Iterable[MetricName] | None	Metrics to include in the parameter matrix. Defaults to all supported benchmatrix metrics.	None

Returns:

Type	Description
list[object]	A list of values suitable for pytest.mark.parametrize.

Source code in src/benchmatrix/bench_harness.py

def make_benchmark_parameters(
    implementations: Mapping[str, TargetFunction],
    cases: Mapping[str, BenchmarkCase] | Iterable[BenchmarkCase],
    *,
    metrics: Iterable[MetricName] | None = None,
) -> list[object]:
    """Create pytest parameters for a metric-by-implementation-by-case matrix.

    Args:
        implementations: Mapping from implementation name to target function.
        cases: Mapping or iterable of benchmark input cases.
        metrics: Metrics to include in the parameter matrix. Defaults to all
            supported benchmatrix metrics.

    Returns:
        A list of values suitable for ``pytest.mark.parametrize``.
    """
    resolved_metrics = _metric_items(metrics)
    implementation_items = _implementation_items(implementations)
    case_items = _case_items(cases)
    pytest = _load_pytest()
    entries: list[tuple[MetricName, str, TargetFunction, str, BenchmarkCase]] = []

    for metric_name in resolved_metrics:
        for implementation_name, function in implementation_items:
            for case_name, case in case_items:
                entries.append((metric_name, implementation_name, function, case_name, case))

    collection_order = _collection_order_environment()
    if collection_order is not None:
        random_seed, order_index = collection_order
        indices = balanced_order_indices(
            [
                (implementation_name, case_name, metric_name)
                for metric_name, implementation_name, _function, case_name, _case in entries
            ],
            order_index=order_index,
            random_seed=random_seed,
        )
        entries = [entries[index] for index in indices]

    parameters: list[object] = []
    for metric_name, implementation_name, function, case_name, case in entries:
        parameters.append(
            pytest.param(
                metric_name,
                implementation_name,
                function,
                case_name,
                case,
                id=f"{metric_name}::{implementation_name}::{case_name}",
            )
        )

    return parameters

make_benchmark_test

make_benchmark_test(
    implementations: Mapping[str, TargetFunction],
    cases: Mapping[str, BenchmarkCase]
    | Iterable[BenchmarkCase],
    *,
    metrics: Iterable[MetricName] | None = None,
    config: BenchmarkConfig | None = None,
) -> Callable[..., None]

Create a pytest test function for a complete benchmark matrix.

Assign the returned function to a module-level name beginning with test_ so pytest collects it.

Parameters:

Name	Type	Description	Default
implementations	Mapping[str, TargetFunction]	Mapping from implementation name to target function.	required
cases	Mapping[str, BenchmarkCase] | Iterable[BenchmarkCase]	Mapping or iterable of benchmark input cases.	required
metrics	Iterable[MetricName] | None	Metrics to include in the parameter matrix. Defaults to all supported benchmatrix metrics.	None
config	BenchmarkConfig | None	Benchmark harness configuration. Defaults to BenchmarkConfig().	None

Returns:

Type	Description
Callable[..., None]	A parametrized pytest test function ready for module-level assignment.

Source code in src/benchmatrix/bench_harness.py

def make_benchmark_test(
    implementations: Mapping[str, TargetFunction],
    cases: Mapping[str, BenchmarkCase] | Iterable[BenchmarkCase],
    *,
    metrics: Iterable[MetricName] | None = None,
    config: BenchmarkConfig | None = None,
) -> Callable[..., None]:
    """Create a pytest test function for a complete benchmark matrix.

    Assign the returned function to a module-level name beginning with
    ``test_`` so pytest collects it.

    Args:
        implementations: Mapping from implementation name to target function.
        cases: Mapping or iterable of benchmark input cases.
        metrics: Metrics to include in the parameter matrix. Defaults to all
            supported benchmatrix metrics.
        config: Benchmark harness configuration. Defaults to
            ``BenchmarkConfig()``.

    Returns:
        A parametrized pytest test function ready for module-level assignment.
    """
    resolved_config = _resolve_config(config)
    parameters = make_benchmark_parameters(implementations, cases, metrics=metrics)

    def benchmark_test(
        benchmark: BenchmarkFixture,
        metric_name: MetricName,
        implementation_name: str,
        function: TargetFunction,
        case_name: str,
        case: BenchmarkCase,
    ) -> None:
        """Run one entry in the generated benchmark matrix."""
        _ = run_benchmark_metric(
            benchmark,
            metric_name,
            implementation_name,
            function,
            case_name,
            case,
            config=resolved_config,
        )

    pytest = _load_pytest()
    return pytest.mark.parametrize(
        ("metric_name", "implementation_name", "function", "case_name", "case"),
        parameters,
    )(benchmark_test)

run_benchmark_metric

run_benchmark_metric(
    benchmark: BenchmarkFixture,
    metric_name: MetricName,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord

Run one benchmark metric for one implementation and case.

Parameters:

Name	Type	Description	Default
benchmark	BenchmarkFixture	Pytest-benchmark fixture instance.	required
metric_name	MetricName	Metric to benchmark.	required
implementation_name	str	Name of the implementation under test.	required
function	TargetFunction	Synchronous function implementation to benchmark.	required
case_name	str	Name of the input case under test.	required
case	BenchmarkCase	Benchmark input case.	required
config	BenchmarkConfig | None	Benchmark harness configuration. Defaults to BenchmarkConfig().	None
stream	TextIO | None	Stream used for progress output. Defaults to sys.stdout when progress output is enabled.	None

Returns:

Type	Description
BenchmarkInvocationRecord	A lightweight invocation record containing metadata attached to the
BenchmarkInvocationRecord	benchmark. This is not a timing result.

Raises:

Type	Description
TypeError	If function is an async function.
ValueError	If metric_name is unsupported.

Source code in src/benchmatrix/bench_harness.py

def run_benchmark_metric(
    benchmark: BenchmarkFixture,
    metric_name: MetricName,
    implementation_name: str,
    function: TargetFunction,
    case_name: str,
    case: BenchmarkCase,
    *,
    config: BenchmarkConfig | None = None,
    stream: TextIO | None = None,
) -> BenchmarkInvocationRecord:
    """Run one benchmark metric for one implementation and case.

    Args:
        benchmark: Pytest-benchmark fixture instance.
        metric_name: Metric to benchmark.
        implementation_name: Name of the implementation under test.
        function: Synchronous function implementation to benchmark.
        case_name: Name of the input case under test.
        case: Benchmark input case.
        config: Benchmark harness configuration. Defaults to
            ``BenchmarkConfig()``.
        stream: Stream used for progress output. Defaults to ``sys.stdout``
            when progress output is enabled.

    Returns:
        A lightweight invocation record containing metadata attached to the
        benchmark. This is not a timing result.

    Raises:
        TypeError: If ``function`` is an async function.
        ValueError: If ``metric_name`` is unsupported.
    """
    resolved_config = _resolve_config(config)
    resolved_metric_name = _validate_metric_name(metric_name)

    if resolved_metric_name == METRIC_SINGLE_CALL_LATENCY:
        return benchmark_single_call_latency(
            benchmark,
            implementation_name,
            function,
            case_name,
            case,
            config=resolved_config,
            stream=stream,
        )

    if resolved_metric_name == METRIC_BATCH_THROUGHPUT:
        return benchmark_batch_throughput(
            benchmark,
            implementation_name,
            function,
            case_name,
            case,
            config=resolved_config,
            stream=stream,
        )

    if resolved_metric_name == METRIC_TAIL_LATENCY:
        return benchmark_tail_latency(
            benchmark,
            implementation_name,
            function,
            case_name,
            case,
            config=resolved_config,
            stream=stream,
        )

    raise ValueError(f"Unsupported benchmark metric: {resolved_metric_name!r}")

shallow_copy

shallow_copy(value: object) -> object

Return a shallow copy of value.

Parameters:

Name	Type	Description	Default
value	object	Value to copy.	required

Returns:

Type	Description
object	A shallow copy of value.

Source code in src/benchmatrix/bench_harness.py

def shallow_copy(value: object) -> object:
    """Return a shallow copy of ``value``.

    Args:
        value: Value to copy.

    Returns:
        A shallow copy of ``value``.
    """
    return copy.copy(value)

default_benchmark_policy

default_benchmark_policy() -> BenchmarkPolicyConfig

Return benchmatrix's built-in comparison policies.

Source code in src/benchmatrix/bench_policy.py

def default_benchmark_policy() -> BenchmarkPolicyConfig:
    """Return benchmatrix's built-in comparison policies."""
    return BenchmarkPolicyConfig(
        compatibility=RunCompatibilityPolicy(),
        evidence=EvidencePolicy(),
        inference=InferencePolicy(),
        precision=PrecisionPolicy(),
        regression=RegressionPolicy(),
    )

load_benchmark_policy

load_benchmark_policy(
    path: str | Path | None = None,
    *,
    search_from: str | Path | None = None,
) -> BenchmarkPolicyConfig

Load tool.benchmatrix policy from TOML.

With an explicit path, the file must contain [tool.benchmatrix]. Otherwise the nearest pyproject.toml at or above search_from is inspected. Discovery stops at the first pyproject; a project without a benchmatrix table uses built-in defaults.

Parameters:

Name	Type	Description	Default
path	str | Path | None	Explicit TOML or pyproject path.	None
search_from	str | Path | None	File or directory from which to discover pyproject.toml. Defaults to the current working directory.	None

Returns:

Type	Description
BenchmarkPolicyConfig	Validated compatibility, evidence, inference, precision, and regression policies.

Raises:

Type	Description
BenchmarkPolicyError	If an explicit file is missing, TOML is invalid, or the benchmatrix configuration does not satisfy its schema.

Source code in src/benchmatrix/bench_policy.py

def load_benchmark_policy(
    path: str | Path | None = None,
    *,
    search_from: str | Path | None = None,
) -> BenchmarkPolicyConfig:
    """Load ``tool.benchmatrix`` policy from TOML.

    With an explicit ``path``, the file must contain ``[tool.benchmatrix]``.
    Otherwise the nearest ``pyproject.toml`` at or above ``search_from`` is
    inspected. Discovery stops at the first pyproject; a project without a
    benchmatrix table uses built-in defaults.

    Args:
        path: Explicit TOML or pyproject path.
        search_from: File or directory from which to discover pyproject.toml.
            Defaults to the current working directory.

    Returns:
        Validated compatibility, evidence, inference, precision, and regression policies.

    Raises:
        BenchmarkPolicyError: If an explicit file is missing, TOML is invalid,
            or the benchmatrix configuration does not satisfy its schema.
    """
    explicit = path is not None
    source = Path(path) if explicit else _discover_pyproject(search_from)
    if source is None:
        return default_benchmark_policy()
    source = source.resolve()

    try:
        with source.open("rb") as stream:
            payload = cast(object, tomllib.load(stream))
    except OSError as exc:
        raise BenchmarkPolicyError(f"Could not read benchmark policy configuration: {source}") from exc
    except tomllib.TOMLDecodeError as exc:
        raise BenchmarkPolicyError(f"Invalid TOML in benchmark policy configuration: {source}") from exc

    root = _mapping(payload, path="root")
    tool = root.get("tool")
    if tool is None:
        if explicit:
            raise BenchmarkPolicyError(f"Configuration does not contain [tool.benchmatrix]: {source}")
        return default_benchmark_policy()
    tool_mapping = _mapping(tool, path="tool")
    raw_config = tool_mapping.get("benchmatrix")
    if raw_config is None:
        if explicit:
            raise BenchmarkPolicyError(f"Configuration does not contain [tool.benchmatrix]: {source}")
        return default_benchmark_policy()

    config = _mapping(raw_config, path="tool.benchmatrix")
    _exact_keys(config, _TOOL_KEYS, path="tool.benchmatrix")
    try:
        compatibility, compatibility_fields = _parse_compatibility(config.get("compatibility"))
        evidence, evidence_fields = _parse_evidence(config.get("evidence"))
        inference, inference_fields = _parse_inference(config.get("inference"))
        precision, precision_fields = _parse_precision(config.get("precision"))
        regression, regression_fields = _parse_regression(config.get("regression"))
        return BenchmarkPolicyConfig(
            compatibility=compatibility,
            evidence=evidence,
            inference=inference,
            precision=precision,
            regression=regression,
            source=source,
            configured_fields=frozenset(
                (
                    *compatibility_fields,
                    *evidence_fields,
                    *inference_fields,
                    *precision_fields,
                    *regression_fields,
                )
            ),
        )
    except BenchmarkPolicyError:
        raise
    except (TypeError, ValueError) as exc:
        raise BenchmarkPolicyError(f"Invalid benchmark policy in {source}: {exc}") from exc

format_comparison_report_markdown

format_comparison_report_markdown(
    report: BenchmarkComparisonReport,
) -> str

Render a comparison report as deterministic GitHub-flavored Markdown.

Parameters:

Name	Type	Description	Default
report	BenchmarkComparisonReport	Portable comparison report to render.	required

Returns:

Type	Description
str	A complete Markdown document ending with a newline.

Raises:

Type	Description
TypeError	If report is not a BenchmarkComparisonReport.

Source code in src/benchmatrix/bench_report.py

def format_comparison_report_markdown(report: BenchmarkComparisonReport) -> str:
    """Render a comparison report as deterministic GitHub-flavored Markdown.

    Args:
        report: Portable comparison report to render.

    Returns:
        A complete Markdown document ending with a newline.

    Raises:
        TypeError: If ``report`` is not a ``BenchmarkComparisonReport``.
    """
    if not isinstance(report, BenchmarkComparisonReport):
        raise TypeError("report must be a BenchmarkComparisonReport.")

    overall = "PASS" if report.passed else "FAIL"
    comparison = "PASS" if report.comparison_passed else "FAIL"
    compatibility = "compatible" if report.compatibility.is_compatible else "blocked"
    lines = [
        "# Benchmark comparison",
        "",
        f"**Overall:** {overall}  ",
        f"**Comparison decision:** {comparison}  ",
        f"**Environment compatibility:** {compatibility}  ",
        f"**Comparison design:** `{report.design}`  ",
        f"**Report schema:** `{report.kind}` version {report.schema_version}",
        "",
        "## Inputs",
        "",
        "| Side | Sources | Collections |",
        "| --- | --- | ---: |",
        (f"| Baseline | {_markdown_sources(report.baselines)} | " + f"{len(report.baseline_collections)} |"),
        (f"| Candidate | {_markdown_sources(report.candidates)} | " + f"{len(report.candidate_collections)} |"),
        "",
    ]
    collections = (*report.baseline_collections, *report.candidate_collections)
    if collections:
        lines.extend(
            [
                "### Collection lifecycle",
                "",
                "| Manifest | Target | Attempts | Successful | Failed | Retries | Complete |",
                "| --- | ---: | ---: | ---: | ---: | ---: | --- |",
            ]
        )
        for collection in collections:
            lines.append(
                "| "
                + " | ".join(
                    (
                        _markdown_text(collection.manifest),
                        str(collection.requested_runs),
                        str(collection.attempted_runs),
                        str(collection.successful_runs),
                        str(collection.failed_runs),
                        str(collection.retry_attempts),
                        "yes" if collection.complete else "no",
                    )
                )
                + " |"
            )
        lines.append("")
    if report.paired_collections:
        lines.extend(
            [
                "### Paired AB/BA collection lifecycle",
                "",
                (
                    "| Manifest | Pair target | Command attempts | Complete pairs | "
                    + "Failed commands | Orphan successes | Complete |"
                ),
                "| --- | ---: | ---: | ---: | ---: | ---: | --- |",
            ]
        )
        for collection in report.paired_collections:
            lines.append(
                "| "
                + " | ".join(
                    (
                        _markdown_text(collection.manifest),
                        str(collection.requested_pairs),
                        str(collection.attempted_commands),
                        str(collection.complete_pair_count),
                        str(collection.failed_commands),
                        str(collection.orphan_success_count),
                        "yes" if collection.complete else "no",
                    )
                )
                + " |"
            )
        lines.append("")

    lines.extend(
        [
            "## Summary",
            "",
            "| Improved | Unchanged | Regressed | Inconclusive | Not comparable |",
            "| ---: | ---: | ---: | ---: | ---: |",
            (
                f"| {len(report.improved)} | {len(report.unchanged)} | "
                + f"{len(report.regressed)} | {len(report.inconclusive)} | "
                + f"{len(report.not_comparable)} |"
            ),
            "",
            "## Matrix results",
            "",
            (
                "| Implementation | Case | Metric | Result | Baseline | Candidate | "
                + "Improvement | Confidence interval | Observed pairwise range | Threshold | Evidence |"
            ),
            "| --- | --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | --- |",
        ]
    )
    for cell, threshold in zip(
        report.comparisons,
        report.threshold_provenance,
        strict=True,
    ):
        result = cell.regression if cell.status == "matched" else cell.status
        effect_range = _markdown_effect_range(
            cell.improvement_low_percent,
            cell.improvement_high_percent,
        )
        confidence_interval = _markdown_confidence_interval(cell.inference)
        evidence = _markdown_evidence_state(
            cell.baseline_evidence,
            cell.candidate_evidence,
        )
        lines.append(
            "| "
            + " | ".join(
                (
                    _markdown_text(cell.implementation_name),
                    _markdown_text(cell.case_name),
                    _markdown_text(cell.metric_name),
                    _markdown_text(result),
                    _markdown_number(cell.baseline_value),
                    _markdown_number(cell.candidate_value),
                    _markdown_percent(cell.improvement_percent),
                    confidence_interval,
                    effect_range,
                    (
                        f"{cell.threshold_percent:.2f}% "
                        + f"({_markdown_text(threshold.scope)}/{_markdown_text(threshold.origin)})"
                    ),
                    evidence,
                )
            )
            + " |"
        )
    lines.append("")

    lines.extend(
        [
            "## Environment compatibility",
            "",
            (
                f"{report.compatibility.pairs_checked} environment pair(s) checked; "
                + f"{len(report.compatibility.blocking)} blocking finding(s); "
                + f"{len(report.compatibility.warnings)} warning(s)."
            ),
            "",
        ]
    )
    if report.compatibility.findings:
        lines.extend(
            [
                "| Severity | Field | Baseline | Candidate | Reason |",
                "| --- | --- | --- | --- | --- |",
            ]
        )
        for finding in report.compatibility.findings:
            lines.append(
                "| "
                + " | ".join(
                    (
                        _markdown_text(finding.severity),
                        _markdown_text(finding.field),
                        _markdown_text(_compact_json(finding.baseline_value)),
                        _markdown_text(_compact_json(finding.candidate_value)),
                        _markdown_text(finding.reason),
                    )
                )
                + " |"
            )
        lines.append("")
    else:
        lines.extend(["No environment differences were reported.", ""])

    lines.extend(
        [
            "## Evidence and diagnostics",
            "",
        ]
    )
    for cell in report.comparisons:
        lines.extend(
            [
                (
                    "### "
                    + " / ".join(
                        (
                            _markdown_text(cell.implementation_name),
                            _markdown_text(cell.case_name),
                            _markdown_text(cell.metric_name),
                        )
                    )
                ),
                "",
                f"- Baseline: {_markdown_evidence_detail(cell.baseline_evidence)}",
                f"- Candidate: {_markdown_evidence_detail(cell.candidate_evidence)}",
                f"- Inference: {_markdown_inference_detail(cell.inference)}",
                f"- Precision planning: {_markdown_precision_detail(cell.precision)}",
            ]
        )
        if cell.reason is not None:
            lines.append(f"- Diagnostic: {_markdown_text(cell.reason)}")
        lines.append("")

    policy = report.regression_policy
    provenance = report.policy_provenance
    lines.extend(
        [
            "## Effective policy",
            "",
            f"- Selection: `{provenance.selection}`",
            (
                "- Configuration: "
                + (
                    _markdown_text(provenance.configuration_file)
                    if provenance.configuration_file is not None
                    else "built-in defaults"
                )
            ),
            (
                "- Configured fields: "
                + (
                    ", ".join(f"`{_markdown_code(field)}`" for field in provenance.configured_fields)
                    if provenance.configured_fields
                    else "none"
                )
            ),
            (
                "- CLI overrides: "
                + (
                    ", ".join(f"`{_markdown_code(field)}`" for field in provenance.cli_overrides)
                    if provenance.cli_overrides
                    else "none"
                )
            ),
            f"- Compatibility mode: `{report.compatibility.policy.mode}`",
            f"- Minimum runs: {report.evidence_policy.minimum_runs}",
            f"- Minimum samples per run: {report.evidence_policy.minimum_samples_per_run}",
            f"- Minimum rounds per run: {report.evidence_policy.minimum_rounds_per_run}",
            (
                "- Require raw samples for inference: "
                + ("yes" if report.evidence_policy.require_raw_samples_for_inference else "no")
            ),
            f"- Minimum tail samples per run: {report.evidence_policy.minimum_tail_samples_per_run}",
            (
                "- Require one iteration for tail latency: "
                + ("yes" if report.evidence_policy.require_tail_iterations_one else "no")
            ),
            f"- Inference method: `{report.inference_policy.method}`",
            f"- Confidence level: {report.inference_policy.confidence_level * 100.0:.2f}%",
            f"- Bootstrap resamples: {report.inference_policy.resamples}",
            f"- Random seed: {report.inference_policy.random_seed}",
            f"- Multiplicity correction: `{report.inference_policy.multiplicity}`",
            (
                "- Precision target: disabled"
                if report.precision_policy.target_half_width_percent is None
                else f"- Precision target half-width: {report.precision_policy.target_half_width_percent:.2f}%"
            ),
            f"- Default regression threshold: {policy.default_threshold_percent:.2f}%",
            (
                "- Selector thresholds: "
                + f"{len(policy.by_metric)} metric, "
                + f"{len(policy.by_implementation)} implementation, "
                + f"{len(policy.by_case)} case, {len(policy.by_cell)} exact cell"
            ),
            "",
        ]
    )
    return "\n".join(lines)

load_comparison_report

load_comparison_report(
    path: str | Path,
) -> BenchmarkComparisonReport

Load and strictly validate a versioned comparison report.

Parameters:

Name	Type	Description	Default
path	str | Path	JSON report written by benchmatrix compare --format json or :func:write_comparison_report.	required

Returns:

Type	Description
BenchmarkComparisonReport	A portable typed comparison report.

Raises:

Type	Description
BenchmarkJsonError	If the file is unreadable, malformed, unsupported, or inconsistent with the report schema.

Source code in src/benchmatrix/bench_report.py

def load_comparison_report(path: str | Path) -> BenchmarkComparisonReport:
    """Load and strictly validate a versioned comparison report.

    Args:
        path: JSON report written by ``benchmatrix compare --format json`` or
            :func:`write_comparison_report`.

    Returns:
        A portable typed comparison report.

    Raises:
        BenchmarkJsonError: If the file is unreadable, malformed, unsupported,
            or inconsistent with the report schema.
    """
    source = Path(path)
    try:
        payload = cast(object, json.loads(source.read_text(encoding="utf-8")))
    except OSError as exc:
        raise BenchmarkJsonError(f"Could not read benchmark comparison report: {source}") from exc
    except json.JSONDecodeError as exc:
        raise BenchmarkJsonError(f"Invalid JSON in benchmark comparison report: {source}") from exc
    return _parse_report(payload)

write_comparison_report

write_comparison_report(
    report: BenchmarkComparisonReport, path: str | Path
) -> None

Write a comparison report as deterministic strict JSON.

Parameters:

Name	Type	Description	Default
report	BenchmarkComparisonReport	Portable report to serialize.	required
path	str | Path	Destination JSON path.	required

Raises:

Type	Description
TypeError	If report is not a BenchmarkComparisonReport.
OSError	If the destination cannot be written.

Source code in src/benchmatrix/bench_report.py

def write_comparison_report(report: BenchmarkComparisonReport, path: str | Path) -> None:
    """Write a comparison report as deterministic strict JSON.

    Args:
        report: Portable report to serialize.
        path: Destination JSON path.

    Raises:
        TypeError: If ``report`` is not a ``BenchmarkComparisonReport``.
        OSError: If the destination cannot be written.
    """
    if not isinstance(report, BenchmarkComparisonReport):
        raise TypeError("report must be a BenchmarkComparisonReport.")
    destination = Path(path)
    destination.write_text(
        json.dumps(report.to_dict(), allow_nan=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )

write_comparison_report_markdown

write_comparison_report_markdown(
    report: BenchmarkComparisonReport, path: str | Path
) -> None

Write a comparison report as deterministic Markdown.

Parameters:

Name	Type	Description	Default
report	BenchmarkComparisonReport	Portable report to render.	required
path	str | Path	Destination Markdown path.	required

Raises:

Type	Description
TypeError	If report is not a BenchmarkComparisonReport.
OSError	If the destination cannot be written.

Source code in src/benchmatrix/bench_report.py

def write_comparison_report_markdown(
    report: BenchmarkComparisonReport,
    path: str | Path,
) -> None:
    """Write a comparison report as deterministic Markdown.

    Args:
        report: Portable report to render.
        path: Destination Markdown path.

    Raises:
        TypeError: If ``report`` is not a ``BenchmarkComparisonReport``.
        OSError: If the destination cannot be written.
    """
    Path(path).write_text(format_comparison_report_markdown(report), encoding="utf-8")

display_benchmark_row

display_benchmark_row(
    row: ParsedBenchmarkRow, stream: TextIO | None = None
) -> None

Print one metric-aware summary of a parsed benchmark row.

Parameters:

Name	Type	Description	Default
row	ParsedBenchmarkRow	Parsed benchmark row to display.	required
stream	TextIO | None	Output stream. Defaults to sys.stdout.	None

Source code in src/benchmatrix/bench_results.py

def display_benchmark_row(
    row: ParsedBenchmarkRow,
    stream: TextIO | None = None,
) -> None:
    """Print one metric-aware summary of a parsed benchmark row.

    Args:
        row: Parsed benchmark row to display.
        stream: Output stream. Defaults to ``sys.stdout``.
    """
    output = sys.stdout if stream is None else stream
    prefix = f"[{row.metric_name}] implementation={row.implementation_name} case={row.case_name}"

    if row.metric_name == METRIC_SINGLE_CALL_LATENCY:
        message = (
            f"{prefix} mean={_format_seconds(row.stats.get(STAT_MEAN))} "
            + f"median={_format_seconds(row.stats.get(STAT_MEDIAN))} "
            + f"min={_format_seconds(row.stats.get(STAT_MIN))}"
        )
        print(
            message,
            file=output,
        )
        return

    if row.metric_name == METRIC_BATCH_THROUGHPUT:
        message = (
            f"{prefix} "
            + f"throughput_mean={_format_rate(row.derived.get(DERIVED_THROUGHPUT_MEAN))} "
            + f"throughput_median={_format_rate(row.derived.get(DERIVED_THROUGHPUT_MEDIAN))} "
            + f"unit={row.derived.get(DERIVED_THROUGHPUT_UNIT_LABEL)}"
        )
        print(
            message,
            file=output,
        )
        return

    if row.metric_name == METRIC_TAIL_LATENCY:
        message = (
            f"{prefix} p50={_format_seconds(row.derived.get(DERIVED_P50))} "
            + f"p95={_format_seconds(row.derived.get(DERIVED_P95))} "
            + f"p99={_format_seconds(row.derived.get(DERIVED_P99))} "
            + f"max={_format_seconds(row.derived.get(DERIVED_MAX))}"
        )
        print(
            message,
            file=output,
        )
        return

    print(
        f"{prefix} mean={_format_seconds(row.stats.get(STAT_MEAN))}",
        file=output,
    )

display_benchmark_rows

display_benchmark_rows(
    rows: Iterable[ParsedBenchmarkRow],
    stream: TextIO | None = None,
) -> None

Print concise metric-aware summaries of parsed benchmark rows.

Parameters:

Name	Type	Description	Default
rows	Iterable[ParsedBenchmarkRow]	Parsed benchmark rows.	required
stream	TextIO | None	Output stream. Defaults to sys.stdout.	None

Source code in src/benchmatrix/bench_results.py

def display_benchmark_rows(
    rows: Iterable[ParsedBenchmarkRow],
    stream: TextIO | None = None,
) -> None:
    """Print concise metric-aware summaries of parsed benchmark rows.

    Args:
        rows: Parsed benchmark rows.
        stream: Output stream. Defaults to ``sys.stdout``.
    """
    for row in rows:
        display_benchmark_row(row, stream=stream)

load_benchmark_json

load_benchmark_json(
    path: str | Path,
) -> list[ParsedBenchmarkRow]

Load benchmatrix-tagged pytest-benchmark JSON and derive metric views.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a JSON file created with --benchmark-json.	required

Returns:

Type	Description
list[ParsedBenchmarkRow]	Benchmatrix-tagged rows with raw pytest-benchmark statistics and derived
list[ParsedBenchmarkRow]	metric-specific fields. Non-benchmatrix rows are rejected.

Raises:

Type	Description
BenchmarkJsonError	If the JSON does not have the expected pytest-benchmark and benchmatrix structure.

Source code in src/benchmatrix/bench_results.py

def load_benchmark_json(path: str | Path) -> list[ParsedBenchmarkRow]:
    """Load benchmatrix-tagged pytest-benchmark JSON and derive metric views.

    Args:
        path: Path to a JSON file created with ``--benchmark-json``.

    Returns:
        Benchmatrix-tagged rows with raw pytest-benchmark statistics and derived
        metric-specific fields. Non-benchmatrix rows are rejected.

    Raises:
        BenchmarkJsonError: If the JSON does not have the expected
            pytest-benchmark and benchmatrix structure.
    """
    return list(load_benchmark_run(path).rows)

load_benchmark_run

load_benchmark_run(path: str | Path) -> BenchmarkRun

Load a benchmatrix run from pytest-benchmark JSON.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a JSON file created with --benchmark-json.	required

Returns:

Type	Description
BenchmarkRun	A first-class run containing matrix rows and top-level run metadata.

Raises:

Type	Description
BenchmarkJsonError	If the JSON does not have the expected pytest-benchmark and benchmatrix structure.

Source code in src/benchmatrix/bench_results.py

def load_benchmark_run(path: str | Path) -> BenchmarkRun:
    """Load a benchmatrix run from pytest-benchmark JSON.

    Args:
        path: Path to a JSON file created with ``--benchmark-json``.

    Returns:
        A first-class run containing matrix rows and top-level run metadata.

    Raises:
        BenchmarkJsonError: If the JSON does not have the expected
            pytest-benchmark and benchmatrix structure.
    """
    path_obj = Path(path)

    try:
        payload = _load_json(path_obj)
    except OSError as exc:
        raise BenchmarkJsonError(f"Could not read benchmark JSON: {path_obj}") from exc
    except json.JSONDecodeError as exc:
        raise BenchmarkJsonError(f"Invalid JSON in benchmark file: {path_obj}") from exc

    payload_mapping = _require_mapping(payload, path="root")
    benchmarks = _require_list(
        payload_mapping.get(JSON_KEY_BENCHMARKS),
        path=f"root.{JSON_KEY_BENCHMARKS}",
    )

    rows: list[ParsedBenchmarkRow] = []
    for index, benchmark_entry in enumerate(benchmarks):
        entry_path = f"root.{JSON_KEY_BENCHMARKS}[{index}]"
        entry = _require_mapping(benchmark_entry, path=entry_path)
        extra_info = _require_mapping(
            entry.get(JSON_KEY_EXTRA_INFO),
            path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}",
        )
        _require_benchmatrix_schema(extra_info, path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}")
        _ = _require_bool(
            extra_info.get(KEY_CASE_FRESH_INPUTS),
            path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_CASE_FRESH_INPUTS}",
        )

        stats = _require_mapping(
            entry.get(JSON_KEY_STATS),
            path=f"{entry_path}.{JSON_KEY_STATS}",
        )

        metric_name = _require_metric_name(
            extra_info.get(KEY_METRIC_NAME),
            path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_METRIC_NAME}",
        )
        if metric_name == METRIC_TAIL_LATENCY:
            _validate_tail_metadata(
                extra_info,
                path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}",
            )
        data = _extract_benchmark_data(entry, stats, metric_name, path=entry_path)
        stats_path = f"{entry_path}.{JSON_KEY_STATS}"
        extra_info_path = f"{entry_path}.{JSON_KEY_EXTRA_INFO}"

        rows.append(
            ParsedBenchmarkRow(
                benchmark_name=_benchmark_name(entry, path=entry_path),
                metric_name=metric_name,
                implementation_name=_require_non_empty_string(
                    extra_info.get(KEY_IMPLEMENTATION_NAME),
                    path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_IMPLEMENTATION_NAME}",
                ),
                case_name=_require_non_empty_string(
                    extra_info.get(KEY_CASE_NAME),
                    path=f"{entry_path}.{JSON_KEY_EXTRA_INFO}.{KEY_CASE_NAME}",
                ),
                stats=stats,
                extra_info=extra_info,
                derived=_derive_stats(
                    metric_name,
                    stats,
                    extra_info,
                    data,
                    stats_path=stats_path,
                    extra_info_path=extra_info_path,
                ),
                samples=tuple(data),
            )
        )

    metadata = {key: value for key, value in payload_mapping.items() if key != JSON_KEY_BENCHMARKS}
    _validate_run_metadata(metadata)
    return BenchmarkRun(rows=tuple(rows), metadata=metadata, source=path_obj)

plan_paired_precision

plan_paired_precision(
    baseline_values: Sequence[float],
    candidate_values: Sequence[float],
    *,
    lower_is_better: bool,
    target_half_width_percent: float,
    confidence_level: float = 0.95,
    family_size: int = 1,
    multiplicity: MultiplicityCorrection = "bonferroni",
    strata: Sequence[str] | None = None,
    minimum_pairs: int = _MINIMUM_GROUP_SIZE,
    pair_count_multiple: int = 2,
) -> PrecisionPlan

Estimate a fixed confirmatory pair count from paired pilot runs.

The planning approximation uses the residual standard deviation of signed paired log ratios. Positive signed log ratios mean improvement, regardless of metric direction. When fixed collection-design strata such as AB and BA are supplied, a separate mean is fitted for each stratum so a fixed orientation effect is not counted as future random variation. Student-t degrees of freedom account for those fitted means.

The requested percentage half-width is converted to log1p(target / 100). This is a multiplicative mean-log-ratio proxy for the formal ratio-of-marginal-medians BCa estimand; the two targets are not identical. minimum_pairs and pair_count_multiple are then applied to the smallest unconstrained count. The default multiple of two keeps a direct AB/BA plan even. With bonferroni, confidence is adjusted across family_size cells before calculating the count.

required_pairs is the size of a fresh future confirmatory collection. additional_pairs is only its arithmetic difference from the pilot count, not a recommendation to append runs to the analyzed pilot. This calculation describes precision only: it does not estimate power, justify optional stopping, or update a confirmatory run count after results have been examined.

Source code in src/benchmatrix/bench_statistics.py

def plan_paired_precision(
    baseline_values: Sequence[float],
    candidate_values: Sequence[float],
    *,
    lower_is_better: bool,
    target_half_width_percent: float,
    confidence_level: float = 0.95,
    family_size: int = 1,
    multiplicity: MultiplicityCorrection = "bonferroni",
    strata: Sequence[str] | None = None,
    minimum_pairs: int = _MINIMUM_GROUP_SIZE,
    pair_count_multiple: int = 2,
) -> PrecisionPlan:
    """Estimate a fixed confirmatory pair count from paired pilot runs.

    The planning approximation uses the residual standard deviation of signed
    paired log ratios. Positive signed log ratios mean improvement, regardless
    of metric direction. When fixed collection-design ``strata`` such as AB and
    BA are supplied, a separate mean is fitted for each stratum so a fixed
    orientation effect is not counted as future random variation. Student-t
    degrees of freedom account for those fitted means.

    The requested percentage half-width is converted to
    ``log1p(target / 100)``. This is a multiplicative mean-log-ratio proxy for
    the formal ratio-of-marginal-medians BCa estimand; the two targets are not
    identical. ``minimum_pairs`` and ``pair_count_multiple`` are then applied
    to the smallest unconstrained count. The default multiple of two keeps a
    direct AB/BA plan even. With ``bonferroni``, confidence is adjusted across
    ``family_size`` cells before calculating the count.

    ``required_pairs`` is the size of a fresh future confirmatory collection.
    ``additional_pairs`` is only its arithmetic difference from the pilot
    count, not a recommendation to append runs to the analyzed pilot. This
    calculation describes precision only: it does not estimate power, justify
    optional stopping, or update a confirmatory run count after results have
    been examined.
    """
    target = _validate_positive_number(
        target_half_width_percent,
        field_name="target_half_width_percent",
    )
    confidence = _validate_open_probability(
        confidence_level,
        field_name="confidence_level",
    )
    if isinstance(family_size, bool) or not isinstance(family_size, int):
        raise TypeError("family_size must be an integer.")
    if family_size <= 0:
        raise ValueError("family_size must be a positive integer.")
    if multiplicity not in {"bonferroni", "none"}:
        raise ValueError(f"Unsupported multiplicity correction: {multiplicity!r}.")
    if isinstance(minimum_pairs, bool) or not isinstance(minimum_pairs, int):
        raise TypeError("minimum_pairs must be an integer.")
    if minimum_pairs < _MINIMUM_GROUP_SIZE:
        raise ValueError(f"minimum_pairs must be at least {_MINIMUM_GROUP_SIZE}.")
    if isinstance(pair_count_multiple, bool) or not isinstance(pair_count_multiple, int):
        raise TypeError("pair_count_multiple must be an integer.")
    if pair_count_multiple <= 0:
        raise ValueError("pair_count_multiple must be a positive integer.")
    adjusted_confidence = 1.0 - (1.0 - confidence) / family_size if multiplicity == "bonferroni" else confidence
    if adjusted_confidence >= 1.0:
        raise ValueError("family_size is too large to represent the Bonferroni-adjusted confidence level.")

    assumptions = (
        "Pilot pairs are representative of the future fixed-design paired collection.",
        "Pairs are independent, while baseline and candidate observations remain dependent within each pair.",
        "The target is a multiplicative half-width for the mean signed paired log ratio; this is a variance-based "
        "proxy and is not the formal ratio-of-marginal-medians BCa estimand.",
        "Pilot residual log-ratio variability is representative of future residual variability; the Student-t "
        "proxy does not guarantee a BCa interval width.",
        *(
            (
                "Future collection uses a prespecified fixed stratum allocation compatible with the pair-count "
                "multiple; fitted stratum effects are not treated as random variation.",
            )
            if strata is not None
            else ("No fixed orientation strata are modeled, so all paired log ratios are treated as exchangeable.",)
        ),
        "Required pairs describe a fresh confirmatory collection; the additional-pairs field is descriptive "
        "arithmetic and does not justify reusing the pilot.",
        "The confirmatory pair count is fixed before collection; this is not power analysis or a sequential "
        "stopping rule.",
    )
    try:
        baseline = tuple(float(value) for value in baseline_values)
        candidate = tuple(float(value) for value in candidate_values)
    except (OverflowError, TypeError, ValueError):
        return _failed_precision_plan(
            pilot_pairs=0,
            target_half_width_percent=target,
            confidence_level=confidence,
            adjusted_confidence_level=adjusted_confidence,
            multiplicity=multiplicity,
            family_size=family_size,
            minimum_pairs=minimum_pairs,
            pair_count_multiple=pair_count_multiple,
            strata_count=0 if strata is not None else 1,
            assumptions=assumptions,
            issues=("paired pilot statistics must be finite numeric values",),
        )

    pilot_pairs = min(len(baseline), len(candidate))
    normalized_strata, strata_issues = _normalize_strata(
        strata,
        expected_length=len(baseline),
    )
    strata_count = 1 if normalized_strata is None and not strata_issues else 0
    if normalized_strata is not None:
        strata_count = len(set(normalized_strata))
    issues = (*_paired_measurement_issues(baseline, candidate), *strata_issues)
    if issues:
        return _failed_precision_plan(
            pilot_pairs=pilot_pairs,
            target_half_width_percent=target,
            confidence_level=confidence,
            adjusted_confidence_level=adjusted_confidence,
            multiplicity=multiplicity,
            family_size=family_size,
            minimum_pairs=minimum_pairs,
            pair_count_multiple=pair_count_multiple,
            strata_count=strata_count,
            assumptions=assumptions,
            issues=issues,
        )

    direction = -1.0 if lower_is_better else 1.0
    log_ratios = tuple(
        direction * (math.log(candidate_value) - math.log(baseline_value))
        for baseline_value, candidate_value in zip(baseline, candidate, strict=True)
    )
    try:
        variability = _residual_log_ratio_standard_deviation(
            log_ratios,
            strata=normalized_strata,
        )
    except (OverflowError, statistics.StatisticsError):
        variability = math.inf
    residual_degrees_of_freedom = pilot_pairs - strata_count
    warnings = (
        "Precision planning targets a multiplicative mean-log-ratio proxy; formal inference targets a ratio of "
        "marginal medians.",
        "The pair-count estimate treats pilot residual variability as representative and does not account for "
        "uncertainty in that estimated variability.",
        *(
            (
                "No orientation strata were supplied; planning treats fixed AB/BA order effects as random "
                "paired variation.",
            )
            if normalized_strata is None
            else ()
        ),
        *(
            (
                f"The pilot contains fewer than {_MINIMUM_STABLE_PILOT_SIZE} pairs; "
                + "its variability estimate is unstable.",
            )
            if pilot_pairs < _MINIMUM_STABLE_PILOT_SIZE
            else ()
        ),
        *(
            (
                f"The pilot has only {residual_degrees_of_freedom} residual degree(s) of freedom after fitting "
                f"{strata_count} stratum mean(s); its variability estimate is unstable.",
            )
            if normalized_strata is not None and residual_degrees_of_freedom < _MINIMUM_STABLE_PILOT_SIZE
            else ()
        ),
    )
    if not math.isfinite(variability):
        return _failed_precision_plan(
            pilot_pairs=pilot_pairs,
            target_half_width_percent=target,
            confidence_level=confidence,
            adjusted_confidence_level=adjusted_confidence,
            multiplicity=multiplicity,
            family_size=family_size,
            minimum_pairs=minimum_pairs,
            pair_count_multiple=pair_count_multiple,
            strata_count=strata_count,
            assumptions=assumptions,
            warnings=warnings,
            issues=("paired pilot log-ratio variability is not finite",),
        )
    if variability == 0.0:
        return _failed_precision_plan(
            pilot_pairs=pilot_pairs,
            target_half_width_percent=target,
            confidence_level=confidence,
            adjusted_confidence_level=adjusted_confidence,
            multiplicity=multiplicity,
            family_size=family_size,
            minimum_pairs=minimum_pairs,
            pair_count_multiple=pair_count_multiple,
            strata_count=strata_count,
            assumptions=assumptions,
            variability=variability,
            warnings=warnings,
            issues=("paired pilot residual log ratios have zero variability; required precision cannot be estimated",),
        )

    target_log_half_width = math.log1p(target / 100.0)
    unconstrained_required = _required_pairs_for_precision(
        variability,
        target_log_half_width=target_log_half_width,
        confidence_level=adjusted_confidence,
        strata_count=strata_count,
    )
    if unconstrained_required is None:
        return _failed_precision_plan(
            pilot_pairs=pilot_pairs,
            target_half_width_percent=target,
            confidence_level=confidence,
            adjusted_confidence_level=adjusted_confidence,
            multiplicity=multiplicity,
            family_size=family_size,
            minimum_pairs=minimum_pairs,
            pair_count_multiple=pair_count_multiple,
            strata_count=strata_count,
            assumptions=assumptions,
            variability=variability,
            warnings=warnings,
            issues=(f"estimated required pair count exceeds {_MAXIMUM_PLANNED_PAIRS:,}",),
        )
    required = _round_up_to_multiple(
        max(unconstrained_required, minimum_pairs),
        pair_count_multiple,
    )
    if required > _MAXIMUM_PLANNED_PAIRS:
        return _failed_precision_plan(
            pilot_pairs=pilot_pairs,
            target_half_width_percent=target,
            confidence_level=confidence,
            adjusted_confidence_level=adjusted_confidence,
            multiplicity=multiplicity,
            family_size=family_size,
            minimum_pairs=minimum_pairs,
            pair_count_multiple=pair_count_multiple,
            strata_count=strata_count,
            assumptions=assumptions,
            variability=variability,
            warnings=warnings,
            issues=(f"design-constrained required pair count exceeds {_MAXIMUM_PLANNED_PAIRS:,}",),
        )
    critical_value = _student_t_critical(
        adjusted_confidence,
        degrees_of_freedom=required - strata_count,
    )
    return PrecisionPlan(
        method="paired_log_ratio_t",
        pilot_pairs=pilot_pairs,
        target_half_width_percent=target,
        confidence_level=confidence,
        adjusted_confidence_level=adjusted_confidence,
        multiplicity=multiplicity,
        family_size=family_size,
        pilot_log_ratio_standard_deviation=variability,
        critical_value=critical_value,
        required_pairs=required,
        additional_pairs=max(0, required - pilot_pairs),
        assumptions=assumptions,
        minimum_pairs=minimum_pairs,
        pair_count_multiple=pair_count_multiple,
        unconstrained_required_pairs=unconstrained_required,
        strata_count=strata_count,
        warnings=warnings,
    )