From 9d266b2f11ce9ecab472b0b814b19060e04aaae7 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 18:26:24 +0100 Subject: [PATCH] feat(detection-core): define fixed measurement algebra --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/measurement.py | 481 ++++++++++++++++++++++++++ tests/test_core_measurement.py | 252 ++++++++++++++ 3 files changed, 734 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/core/measurement.py create mode 100644 tests/test_core_measurement.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 6bec39a..e17b81e 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -219,7 +219,7 @@ - _Depends: 4.2, 7.2, 7.4_ - [ ] 8. Implement fixed measurement and leakage-safe admission -- [ ] 8.1 (P) Define comparable baseline identities and fixed metric algebra +- [x] 8.1 (P) Define comparable baseline identities and fixed metric algebra - Bind arm, population/freeze, claimed scopes, exact contract/fact/model sets, acquisition, analyzer, configuration, denominator, threshold set, and outcome mapping. - Define closed numerator/denominator rules for pair correctness, recall, fixed and ordinary false alarms, precision, semantic coverage, and model success. - Require Wilson 95 intervals with numerator and denominator; serialize zero denominators as null and prevent them from passing. diff --git a/src/openultrasast/core/measurement.py b/src/openultrasast/core/measurement.py new file mode 100644 index 0000000..8d4bbac --- /dev/null +++ b/src/openultrasast/core/measurement.py @@ -0,0 +1,481 @@ +"""Comparable baseline identities and the fixed detection metric algebra.""" + +from __future__ import annotations + +import math +import re +from collections.abc import Sequence +from dataclasses import dataclass +from enum import Enum + +from openultrasast.core import canonical + +BASELINE_IDENTITY_SCHEMA = "openultrasast.detection-core.baseline-identity.v2" +DENOMINATOR_POLICY_SCHEMA = "openultrasast.detection-core.denominator-policy.v2" +IDENTITY_SET_SCHEMA = "openultrasast.detection-core.identity-set.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") +_WILSON_Z_95 = 1.959963984540054 + + +class MeasurementContractError(ValueError): + """A baseline or metric violates the fixed measurement contract.""" + + +def _text(value: object, field: str) -> None: + if not isinstance(value, str) or not value: + raise MeasurementContractError(f"{field} must be nonempty text") + + +def _digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise MeasurementContractError(f"{field} must be a full lowercase SHA-256 digest") + + +def _count(value: object, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise MeasurementContractError(f"{field} must be a non-negative integer") + return value + + +def _finite(value: object, field: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): + raise MeasurementContractError(f"{field} must be finite") + return float(value) + + +@dataclass(frozen=True, order=True) +class ScopeKey: + family_id: str + ecosystem: str + + def __post_init__(self) -> None: + _text(self.family_id, "family_id") + _text(self.ecosystem, "ecosystem") + + +def _scope_value(value: ScopeKey) -> canonical.JsonValue: + return {"family_id": value.family_id, "ecosystem": value.ecosystem} + + +def _canonical_scopes(values: tuple[ScopeKey, ...], field: str) -> None: + if not isinstance(values, tuple) or not values: + raise MeasurementContractError(f"{field} must be a nonempty tuple") + if not all(isinstance(item, ScopeKey) for item in values): + raise MeasurementContractError(f"{field} must contain ScopeKey values") + if values != tuple(sorted(values)): + raise MeasurementContractError(f"{field} must use canonical order") + if len(set(values)) != len(values): + raise MeasurementContractError(f"{field} contains a duplicate scope") + + +def derive_identity_set_digest(kind: str, identities: tuple[str, ...]) -> str: + _text(kind, "identity set kind") + if not isinstance(identities, tuple): + raise MeasurementContractError("identities must be a tuple") + for identity in identities: + _digest(identity, f"{kind} identity") + if identities != tuple(sorted(identities)): + raise MeasurementContractError("identities must use canonical order") + if len(set(identities)) != len(identities): + raise MeasurementContractError("identity set contains a duplicate") + return canonical.content_digest({"schema_version": IDENTITY_SET_SCHEMA, "kind": kind, "identities": identities}) + + +@dataclass(frozen=True) +class BaselineIdentity: + arm_id: str + population_digest: str + freeze_digest: str + claimed_scopes: tuple[ScopeKey, ...] + family_contract_digests: tuple[str, ...] + family_contract_set_digest: str + fact_schema_digest: str + fact_set_digest: str + acquisition_record_digest: str + analyzer_revision: str + configuration_digest: str + model_program_digests: tuple[str, ...] + model_program_set_digest: str | None + denominator_policy_digest: str + threshold_set_digest: str + outcome_mapping_digest: str + + def __post_init__(self) -> None: + _validate_baseline(self) + + +def _validate_identity_tuple(values: tuple[str, ...], field: str, *, allow_empty: bool) -> None: + if not isinstance(values, tuple) or (not values and not allow_empty): + qualifier = "a tuple" if allow_empty else "a nonempty tuple" + raise MeasurementContractError(f"{field} must be {qualifier}") + for value in values: + _digest(value, field) + if values != tuple(sorted(values)): + raise MeasurementContractError(f"{field} must use canonical order") + if len(set(values)) != len(values): + raise MeasurementContractError(f"{field} contains a duplicate identity") + + +def _validate_baseline(value: BaselineIdentity) -> None: + _text(value.arm_id, "arm_id") + _text(value.analyzer_revision, "analyzer_revision") + for item, field in ( + (value.population_digest, "population_digest"), + (value.freeze_digest, "freeze_digest"), + (value.fact_schema_digest, "fact_schema_digest"), + (value.fact_set_digest, "fact_set_digest"), + (value.acquisition_record_digest, "acquisition_record_digest"), + (value.configuration_digest, "configuration_digest"), + (value.denominator_policy_digest, "denominator_policy_digest"), + (value.threshold_set_digest, "threshold_set_digest"), + (value.outcome_mapping_digest, "outcome_mapping_digest"), + ): + _digest(item, field) + _canonical_scopes(value.claimed_scopes, "claimed_scopes") + _validate_identity_tuple(value.family_contract_digests, "family contract digests", allow_empty=False) + _digest(value.family_contract_set_digest, "family contract set digest") + # Association requires the actual catalog entries and is validated by the admission context. + _validate_identity_tuple(value.model_program_digests, "model program digests", allow_empty=True) + expected_models = derive_identity_set_digest("model_program", value.model_program_digests) if value.model_program_digests else None + if value.model_program_set_digest != expected_models: + raise MeasurementContractError("model program set digest does not match exact set") + + +def _baseline_value(value: BaselineIdentity) -> canonical.JsonValue: + return { + "schema_version": BASELINE_IDENTITY_SCHEMA, + "arm_id": value.arm_id, + "population_digest": value.population_digest, + "freeze_digest": value.freeze_digest, + "claimed_scopes": tuple(_scope_value(item) for item in value.claimed_scopes), + "family_contract_digests": value.family_contract_digests, + "family_contract_set_digest": value.family_contract_set_digest, + "fact_schema_digest": value.fact_schema_digest, + "fact_set_digest": value.fact_set_digest, + "acquisition_record_digest": value.acquisition_record_digest, + "analyzer_revision": value.analyzer_revision, + "configuration_digest": value.configuration_digest, + "model_program_digests": value.model_program_digests, + "model_program_set_digest": value.model_program_set_digest, + "denominator_policy_digest": value.denominator_policy_digest, + "threshold_set_digest": value.threshold_set_digest, + "outcome_mapping_digest": value.outcome_mapping_digest, + } + + +def derive_baseline_identity_digest(value: BaselineIdentity) -> str: + if not isinstance(value, BaselineIdentity): + raise MeasurementContractError("baseline must be a BaselineIdentity") + _validate_baseline(value) + return canonical.content_digest(_baseline_value(value)) + + +def require_comparable_baselines(values: Sequence[BaselineIdentity]) -> BaselineIdentity: + if not isinstance(values, Sequence) or not values: + raise MeasurementContractError("baselines must be a nonempty sequence") + first = values[0] + if not isinstance(first, BaselineIdentity): + raise MeasurementContractError("baselines must contain BaselineIdentity values") + expected = derive_baseline_identity_digest(first) + for value in values[1:]: + if not isinstance(value, BaselineIdentity) or derive_baseline_identity_digest(value) != expected: + raise MeasurementContractError("cannot aggregate a non-comparable baseline") + return first + + +class MetricId(Enum): + PAIRED_CORRECTNESS = "paired_correctness" + VULNERABLE_RECALL = "vulnerable_recall" + FIXED_FALSE_ALARM_RATE = "fixed_false_alarm_rate" + ORDINARY_FALSE_ALARMS_PER_100 = "ordinary_false_alarms_per_100" + PRECISION = "precision" + SEMANTIC_STAGE_COVERAGE = "semantic_stage_coverage" + MODEL_SUCCESS_RATE = "model_success_rate" + + +class NumeratorKind(Enum): + CORRECT_PAIRS = "correct_pairs" + VULNERABLE_VIOLATION_CASES = "vulnerable_violation_cases" + FIXED_DEFECT_CASES = "fixed_defect_cases" + ORDINARY_DEFECT_CASES = "ordinary_defect_cases" + TRUE_DEFECT_CASES = "true_defect_cases" + SITES_RECOGNIZED = "sites_recognized" + SUCCESSFUL_MODEL_CALLS = "successful_model_calls" + + +class DenominatorKind(Enum): + PAIRS = "pairs" + VULNERABLE_SIDES = "vulnerable_sides" + FIXED_SIDES = "fixed_sides" + ORDINARY_CASES = "ordinary_cases" + ALL_DEFECT_CASES = "all_defect_cases" + SITES_CONSIDERED = "sites_considered" + MODEL_CALLS = "model_calls" + + +class IntervalMethod(Enum): + WILSON_95 = "wilson_95" + + +class Statistic(Enum): + COUNT = "count" + RATE = "rate" + PER_100 = "per_100" + WILSON_LOWER = "wilson_lower" + WILSON_UPPER = "wilson_upper" + + +class Comparator(Enum): + AT_LEAST = "at_least" + AT_MOST = "at_most" + + +_ALGEBRA = { + MetricId.PAIRED_CORRECTNESS: (NumeratorKind.CORRECT_PAIRS, DenominatorKind.PAIRS, Comparator.AT_LEAST), + MetricId.VULNERABLE_RECALL: ( + NumeratorKind.VULNERABLE_VIOLATION_CASES, + DenominatorKind.VULNERABLE_SIDES, + Comparator.AT_LEAST, + ), + MetricId.FIXED_FALSE_ALARM_RATE: ( + NumeratorKind.FIXED_DEFECT_CASES, + DenominatorKind.FIXED_SIDES, + Comparator.AT_MOST, + ), + MetricId.ORDINARY_FALSE_ALARMS_PER_100: ( + NumeratorKind.ORDINARY_DEFECT_CASES, + DenominatorKind.ORDINARY_CASES, + Comparator.AT_MOST, + ), + MetricId.PRECISION: (NumeratorKind.TRUE_DEFECT_CASES, DenominatorKind.ALL_DEFECT_CASES, Comparator.AT_LEAST), + MetricId.SEMANTIC_STAGE_COVERAGE: ( + NumeratorKind.SITES_RECOGNIZED, + DenominatorKind.SITES_CONSIDERED, + Comparator.AT_LEAST, + ), + MetricId.MODEL_SUCCESS_RATE: ( + NumeratorKind.SUCCESSFUL_MODEL_CALLS, + DenominatorKind.MODEL_CALLS, + Comparator.AT_LEAST, + ), +} + + +@dataclass(frozen=True) +class MetricRule: + metric_id: MetricId + scopes: tuple[ScopeKey, ...] + numerator: NumeratorKind + denominator: DenominatorKind + + def __post_init__(self) -> None: + if not isinstance(self.metric_id, MetricId): + raise MeasurementContractError("metric_id must be a MetricId") + if not isinstance(self.numerator, NumeratorKind) or not isinstance(self.denominator, DenominatorKind): + raise MeasurementContractError("metric rule kinds must use the closed enums") + _canonical_scopes(self.scopes, "metric scopes") + + +def standard_metric_rules(scopes: tuple[ScopeKey, ...]) -> tuple[MetricRule, ...]: + _canonical_scopes(scopes, "metric scopes") + return tuple(MetricRule(metric, scopes, _ALGEBRA[metric][0], _ALGEBRA[metric][1]) for metric in MetricId) + + +def _rule_value(value: MetricRule) -> canonical.JsonValue: + return { + "metric_id": value.metric_id.value, + "scopes": tuple(_scope_value(scope) for scope in value.scopes), + "numerator": value.numerator.value, + "denominator": value.denominator.value, + } + + +@dataclass(frozen=True) +class DenominatorPolicy: + rules: tuple[MetricRule, ...] + interval_method: IntervalMethod + policy_digest: str + + def __post_init__(self) -> None: + _validate_policy(self) + + +def _validate_policy(value: DenominatorPolicy) -> None: + if not isinstance(value.rules, tuple) or not all(isinstance(item, MetricRule) for item in value.rules): + raise MeasurementContractError("rules must be a tuple of MetricRule values") + if tuple(rule.metric_id for rule in value.rules) != tuple(MetricId): + raise MeasurementContractError("each standard metric must occur exactly once in canonical order") + for rule in value.rules: + expected_numerator, expected_denominator, _ = _ALGEBRA[rule.metric_id] + if (rule.numerator, rule.denominator) != (expected_numerator, expected_denominator): + raise MeasurementContractError("metric rule violates the fixed algebra") + if value.interval_method is not IntervalMethod.WILSON_95: + raise MeasurementContractError("interval method must be Wilson 95%") + expected_digest = _policy_digest(value.rules, value.interval_method) + if value.policy_digest == "": + object.__setattr__(value, "policy_digest", expected_digest) + return + _digest(value.policy_digest, "denominator policy digest") + if value.policy_digest != expected_digest: + raise MeasurementContractError("denominator policy digest does not match policy") + + +def _policy_digest(rules: tuple[MetricRule, ...], method: IntervalMethod) -> str: + return canonical.content_digest( + { + "schema_version": DENOMINATOR_POLICY_SCHEMA, + "rules": tuple(_rule_value(rule) for rule in rules), + "interval_method": method.value, + } + ) + + +def derive_denominator_policy_digest(value: DenominatorPolicy) -> str: + if not isinstance(value, DenominatorPolicy): + raise MeasurementContractError("policy must be a DenominatorPolicy") + return _policy_digest(value.rules, value.interval_method) + + +@dataclass(frozen=True) +class MetricResult: + metric_id: MetricId + numerator: int + denominator: int + value: float | None + interval_method: IntervalMethod + interval_lower: float | None + interval_upper: float | None + + def __post_init__(self) -> None: + _validate_result(self) + + +def _validate_rule(rule: MetricRule) -> None: + if not isinstance(rule, MetricRule): + raise MeasurementContractError("rule must be a MetricRule") + expected = _ALGEBRA[rule.metric_id] + if (rule.numerator, rule.denominator) != expected[:2]: + raise MeasurementContractError("metric rule violates the fixed algebra") + + +def _wilson(numerator: int, denominator: int) -> tuple[float, float]: + proportion = numerator / denominator + z_squared = _WILSON_Z_95 * _WILSON_Z_95 + denominator_term = 1.0 + z_squared / denominator + center = (proportion + z_squared / (2.0 * denominator)) / denominator_term + margin = (_WILSON_Z_95 / denominator_term) * math.sqrt( + proportion * (1.0 - proportion) / denominator + z_squared / (4.0 * denominator * denominator) + ) + return center - margin, center + margin + + +def _metric_values(metric_id: MetricId, numerator: int, denominator: int) -> tuple[float | None, float | None, float | None]: + if denominator == 0: + return None, None, None + lower, upper = _wilson(numerator, denominator) + scale = 100.0 if metric_id is MetricId.ORDINARY_FALSE_ALARMS_PER_100 else 1.0 + return (numerator / denominator) * scale, lower * scale, upper * scale + + +def _canonical_metric_float(actual: object, expected: float | None) -> bool: + if expected is None: + return actual is None + return type(actual) is float and math.isfinite(actual) and actual.hex() == expected.hex() + + +def _validate_result(result: MetricResult) -> None: + if not isinstance(result.metric_id, MetricId): + raise MeasurementContractError("metric result id must use the closed enum") + numerator = _count(result.numerator, "numerator") + denominator = _count(result.denominator, "denominator") + if numerator > denominator: + raise MeasurementContractError("numerator must not exceed denominator") + if result.interval_method is not IntervalMethod.WILSON_95: + raise MeasurementContractError("interval method must be Wilson 95%") + actual = (result.value, result.interval_lower, result.interval_upper) + expected = _metric_values(result.metric_id, numerator, denominator) + if not all(_canonical_metric_float(item, expected[index]) for index, item in enumerate(actual)): + raise MeasurementContractError("metric values do not match the fixed Wilson algebra") + + +def evaluate_metric(rule: MetricRule, *, numerator: int, denominator: int) -> MetricResult: + _validate_rule(rule) + numerator = _count(numerator, "numerator") + denominator = _count(denominator, "denominator") + if numerator > denominator: + raise MeasurementContractError("numerator must not exceed denominator") + value, lower, upper = _metric_values(rule.metric_id, numerator, denominator) + return MetricResult(rule.metric_id, numerator, denominator, value, IntervalMethod.WILSON_95, lower, upper) + + +def _statistic_value(result: MetricResult, statistic: Statistic) -> float | None: + if statistic is Statistic.COUNT: + return float(result.numerator) + if statistic is Statistic.RATE: + if result.metric_id is MetricId.ORDINARY_FALSE_ALARMS_PER_100: + raise MeasurementContractError("ordinary false alarms use the per_100 statistic") + return result.value + if statistic is Statistic.PER_100: + if result.metric_id is not MetricId.ORDINARY_FALSE_ALARMS_PER_100: + raise MeasurementContractError("per_100 statistic is only valid for ordinary false alarms") + return result.value + if statistic is Statistic.WILSON_LOWER: + return result.interval_lower + if statistic is Statistic.WILSON_UPPER: + return result.interval_upper + raise MeasurementContractError("statistic must use the closed enum") + + +def metric_passes(result: MetricResult, statistic: Statistic, comparator: Comparator, target: float) -> bool: + if not isinstance(result, MetricResult): + raise MeasurementContractError("result must be a MetricResult") + _validate_result(result) + if not isinstance(statistic, Statistic): + raise MeasurementContractError("statistic must use the closed enum") + if not isinstance(comparator, Comparator): + raise MeasurementContractError("comparator must use the closed enum") + target = _finite(target, "target") + required = _ALGEBRA[result.metric_id][2] + if comparator is not required: + raise MeasurementContractError("comparator direction violates the fixed metric algebra") + value = _statistic_value(result, statistic) + if result.denominator == 0 or value is None: + return False + return value >= target if comparator is Comparator.AT_LEAST else value <= target + + +def metric_result_value(result: MetricResult) -> canonical.JsonValue: + _validate_result(result) + return { + "metric_id": result.metric_id.value, + "numerator": result.numerator, + "denominator": result.denominator, + "value": result.value, + "interval_method": result.interval_method.value, + "interval_lower": result.interval_lower, + "interval_upper": result.interval_upper, + } + + +__all__ = [ + "BaselineIdentity", + "Comparator", + "DenominatorKind", + "DenominatorPolicy", + "IntervalMethod", + "MeasurementContractError", + "MetricId", + "MetricResult", + "MetricRule", + "NumeratorKind", + "ScopeKey", + "Statistic", + "derive_baseline_identity_digest", + "derive_denominator_policy_digest", + "derive_identity_set_digest", + "evaluate_metric", + "metric_passes", + "metric_result_value", + "require_comparable_baselines", + "standard_metric_rules", +] diff --git a/tests/test_core_measurement.py b/tests/test_core_measurement.py new file mode 100644 index 0000000..c2f9636 --- /dev/null +++ b/tests/test_core_measurement.py @@ -0,0 +1,252 @@ +"""Fixed measurement algebra and baseline comparability contracts.""" + +from __future__ import annotations + +from dataclasses import replace + +import pytest +from test_core_family_catalog import _catalog as make_family_catalog +from test_core_family_catalog import _contract as make_family_contract + +from openultrasast.core import canonical +from openultrasast.core.measurement import ( + BaselineIdentity, + Comparator, + DenominatorKind, + DenominatorPolicy, + IntervalMethod, + MeasurementContractError, + MetricId, + NumeratorKind, + ScopeKey, + Statistic, + derive_baseline_identity_digest, + derive_denominator_policy_digest, + derive_identity_set_digest, + evaluate_metric, + metric_passes, + metric_result_value, + require_comparable_baselines, + standard_metric_rules, +) + + +def _digest(label: str) -> str: + return canonical.content_digest(label) + + +def _scopes() -> tuple[ScopeKey, ...]: + return (ScopeKey("access_control", "python"), ScopeKey("injection", "javascript")) + + +def _policy() -> DenominatorPolicy: + rules = standard_metric_rules(_scopes()) + return DenominatorPolicy(rules, IntervalMethod.WILSON_95, "") + + +def _baseline(*, arm_id: str = "arm-a", scopes: tuple[ScopeKey, ...] | None = None) -> BaselineIdentity: + contracts = (_digest("contract-a"), _digest("contract-b")) + models = (_digest("model-a"),) + return BaselineIdentity( + arm_id=arm_id, + population_digest=_digest("population"), + freeze_digest=_digest("freeze"), + claimed_scopes=scopes or _scopes(), + family_contract_digests=contracts, + family_contract_set_digest=_digest("family-catalog"), + fact_schema_digest=_digest("fact-schema"), + fact_set_digest=_digest("fact-set"), + acquisition_record_digest=_digest("acquisition"), + analyzer_revision="analyzer-v7", + configuration_digest=_digest("configuration"), + model_program_digests=models, + model_program_set_digest=derive_identity_set_digest("model_program", models), + denominator_policy_digest=_policy().policy_digest, + threshold_set_digest=_digest("thresholds"), + outcome_mapping_digest=_digest("outcome-mapping"), + ) + + +def test_baseline_identity_binds_every_comparability_field() -> None: + baseline = _baseline() + original = derive_baseline_identity_digest(baseline) + mutations = ( + replace(baseline, arm_id="arm-b"), + replace(baseline, population_digest=_digest("population-b")), + replace(baseline, freeze_digest=_digest("freeze-b")), + _baseline(scopes=(ScopeKey("access_control", "python"),)), + replace(baseline, family_contract_digests=(_digest("contract-c"),)), + replace(baseline, family_contract_set_digest=_digest("family-catalog-c")), + replace(baseline, fact_schema_digest=_digest("fact-schema-b")), + replace(baseline, fact_set_digest=_digest("fact-set-b")), + replace(baseline, acquisition_record_digest=_digest("acquisition-b")), + replace(baseline, analyzer_revision="analyzer-v8"), + replace(baseline, configuration_digest=_digest("configuration-b")), + replace( + baseline, + model_program_digests=(_digest("model-b"),), + model_program_set_digest=derive_identity_set_digest("model_program", (_digest("model-b"),)), + ), + replace(baseline, denominator_policy_digest=_digest("denominator-b")), + replace(baseline, threshold_set_digest=_digest("thresholds-b")), + replace(baseline, outcome_mapping_digest=_digest("outcome-mapping-b")), + ) + assert len(mutations) == 15 + assert all(derive_baseline_identity_digest(item) != original for item in mutations) + for mutation in mutations: + with pytest.raises(MeasurementContractError, match="non-comparable baseline"): + require_comparable_baselines((baseline, mutation)) + + +def test_baseline_accepts_authoritative_catalog_digest_and_exact_contract_set() -> None: + catalog = make_family_catalog(make_family_contract("alpha"), make_family_contract("zeta")) + contracts = tuple(sorted(entry.contract_digest for entry in catalog.entries)) + baseline = replace( + _baseline(), + family_contract_digests=contracts, + family_contract_set_digest=catalog.catalog_digest, + ) + assert baseline.family_contract_set_digest == catalog.catalog_digest + assert baseline.family_contract_digests == contracts + mutations = ( + replace(baseline, family_contract_set_digest=_digest("substituted catalog")), + replace(baseline, family_contract_digests=(_digest("substituted contract"),)), + ) + for mutation in mutations: + assert derive_baseline_identity_digest(mutation) != derive_baseline_identity_digest(baseline) + with pytest.raises(MeasurementContractError, match="non-comparable baseline"): + require_comparable_baselines((baseline, mutation)) + + +def test_baseline_refuses_noncanonical_or_inconsistent_sets() -> None: + baseline = _baseline() + with pytest.raises(MeasurementContractError, match="canonical order"): + replace(baseline, claimed_scopes=tuple(reversed(baseline.claimed_scopes))) + with pytest.raises(MeasurementContractError, match="duplicate"): + replace( + baseline, + family_contract_digests=(baseline.family_contract_digests[0],) * 2, + family_contract_set_digest=derive_identity_set_digest("family_contract", (baseline.family_contract_digests[0],) * 2), + ) + with pytest.raises(MeasurementContractError, match="model program set digest"): + replace(baseline, model_program_digests=(), model_program_set_digest=baseline.model_program_set_digest) + + +def test_two_arms_and_cross_scope_results_cannot_aggregate() -> None: + baseline = _baseline() + assert require_comparable_baselines((baseline, baseline)) == baseline + with pytest.raises(MeasurementContractError, match="non-comparable baseline"): + require_comparable_baselines((baseline, _baseline(arm_id="arm-b"))) + with pytest.raises(MeasurementContractError, match="non-comparable baseline"): + require_comparable_baselines((baseline, _baseline(scopes=(ScopeKey("access_control", "python"),)))) + with pytest.raises(MeasurementContractError, match="nonempty"): + require_comparable_baselines(()) + + +def test_standard_policy_is_closed_and_canonical() -> None: + policy = _policy() + assert tuple(rule.metric_id for rule in policy.rules) == tuple(MetricId) + assert policy.interval_method is IntervalMethod.WILSON_95 + assert policy.policy_digest == derive_denominator_policy_digest(policy) + expected = { + MetricId.PAIRED_CORRECTNESS: (NumeratorKind.CORRECT_PAIRS, DenominatorKind.PAIRS), + MetricId.VULNERABLE_RECALL: (NumeratorKind.VULNERABLE_VIOLATION_CASES, DenominatorKind.VULNERABLE_SIDES), + MetricId.FIXED_FALSE_ALARM_RATE: (NumeratorKind.FIXED_DEFECT_CASES, DenominatorKind.FIXED_SIDES), + MetricId.ORDINARY_FALSE_ALARMS_PER_100: (NumeratorKind.ORDINARY_DEFECT_CASES, DenominatorKind.ORDINARY_CASES), + MetricId.PRECISION: (NumeratorKind.TRUE_DEFECT_CASES, DenominatorKind.ALL_DEFECT_CASES), + MetricId.SEMANTIC_STAGE_COVERAGE: (NumeratorKind.SITES_RECOGNIZED, DenominatorKind.SITES_CONSIDERED), + MetricId.MODEL_SUCCESS_RATE: (NumeratorKind.SUCCESSFUL_MODEL_CALLS, DenominatorKind.MODEL_CALLS), + } + assert {rule.metric_id: (rule.numerator, rule.denominator) for rule in policy.rules} == expected + + +def test_policy_refuses_missing_duplicate_or_reversed_rules() -> None: + policy = _policy() + with pytest.raises(MeasurementContractError, match="exactly once"): + DenominatorPolicy(policy.rules[:-1], policy.interval_method, policy.policy_digest) + with pytest.raises(MeasurementContractError, match="exactly once"): + DenominatorPolicy(policy.rules[:-1] + (policy.rules[0],), policy.interval_method, policy.policy_digest) + wrong = replace(policy.rules[0], numerator=NumeratorKind.FIXED_DEFECT_CASES) + with pytest.raises(MeasurementContractError, match="fixed algebra"): + DenominatorPolicy((wrong,) + policy.rules[1:], policy.interval_method, policy.policy_digest) + with pytest.raises(MeasurementContractError, match="canonical order"): + DenominatorPolicy(tuple(reversed(policy.rules)), policy.interval_method, policy.policy_digest) + + +@pytest.mark.parametrize("metric_id", tuple(MetricId)) +def test_zero_denominator_is_null_and_cannot_pass(metric_id: MetricId) -> None: + rule = {item.metric_id: item for item in standard_metric_rules(_scopes())}[metric_id] + result = evaluate_metric(rule, numerator=0, denominator=0) + assert result.numerator == 0 + assert result.denominator == 0 + assert result.value is None + assert result.interval_lower is None + assert result.interval_upper is None + serialized = canonical.canonical_bytes(metric_result_value(result)) + assert b'"value":null' in serialized + assert b'"interval_lower":null' in serialized + assert b'"interval_upper":null' in serialized + false_alarm = metric_id in {MetricId.FIXED_FALSE_ALARM_RATE, MetricId.ORDINARY_FALSE_ALARMS_PER_100} + comparator = Comparator.AT_MOST if false_alarm else Comparator.AT_LEAST + statistic = Statistic.PER_100 if metric_id is MetricId.ORDINARY_FALSE_ALARMS_PER_100 else Statistic.RATE + assert not metric_passes(result, statistic, comparator, 0.0) + assert not metric_passes(result, Statistic.COUNT, comparator, 0.0) + + +def test_wilson_95_interval_and_metric_scales_are_fixed() -> None: + rules = {rule.metric_id: rule for rule in standard_metric_rules(_scopes())} + recall = evaluate_metric(rules[MetricId.VULNERABLE_RECALL], numerator=8, denominator=10) + assert recall.value == pytest.approx(0.8) + assert recall.interval_lower == pytest.approx(0.4901624715) + assert recall.interval_upper == pytest.approx(0.9433178485) + ordinary = evaluate_metric(rules[MetricId.ORDINARY_FALSE_ALARMS_PER_100], numerator=2, denominator=40) + assert ordinary.value == pytest.approx(5.0) + assert ordinary.interval_lower == pytest.approx(1.3820667386) + assert ordinary.interval_upper == pytest.approx(16.5038773691) + assert metric_passes(recall, Statistic.WILSON_LOWER, Comparator.AT_LEAST, 0.49) + assert metric_passes(ordinary, Statistic.PER_100, Comparator.AT_MOST, 5.0) + + +def test_metric_counts_and_threshold_inputs_are_strict() -> None: + rule = standard_metric_rules(_scopes())[0] + for numerator, denominator in ((-1, 1), (True, 1), (2, 1), (0, -1)): + with pytest.raises(MeasurementContractError): + evaluate_metric(rule, numerator=numerator, denominator=denominator) + result = evaluate_metric(rule, numerator=1, denominator=1) + with pytest.raises(MeasurementContractError, match="fixed Wilson algebra"): + replace(result, value=0.0) + with pytest.raises(MeasurementContractError, match="finite"): + metric_passes(result, Statistic.RATE, Comparator.AT_LEAST, float("nan")) + with pytest.raises(MeasurementContractError, match="statistic"): + metric_passes(result, Statistic.PER_100, Comparator.AT_LEAST, 1.0) + with pytest.raises(MeasurementContractError, match="comparator direction"): + metric_passes(result, Statistic.RATE, Comparator.AT_MOST, 1.0) + + +def test_metric_result_rejects_boolean_float_aliases() -> None: + rule = standard_metric_rules(_scopes())[0] + result = evaluate_metric(rule, numerator=1, denominator=1) + with pytest.raises(MeasurementContractError, match="fixed Wilson algebra"): + replace(result, value=True) + with pytest.raises(MeasurementContractError, match="fixed Wilson algebra"): + replace(result, interval_upper=True) + + +def test_metric_result_rejects_integer_float_aliases() -> None: + rule = standard_metric_rules(_scopes())[0] + result = evaluate_metric(rule, numerator=1, denominator=1) + assert replace(result, value=1.0, interval_upper=1.0) == result + with pytest.raises(MeasurementContractError, match="fixed Wilson algebra"): + replace(result, value=1) + with pytest.raises(MeasurementContractError, match="fixed Wilson algebra"): + replace(result, interval_upper=1) + + +def test_metric_result_rejects_negative_signed_zero_aliases() -> None: + rule = standard_metric_rules(_scopes())[0] + result = evaluate_metric(rule, numerator=0, denominator=1) + assert replace(result, value=0.0, interval_lower=0.0) == result + with pytest.raises(MeasurementContractError, match="fixed Wilson algebra"): + replace(result, value=-0.0) + with pytest.raises(MeasurementContractError, match="fixed Wilson algebra"): + replace(result, interval_lower=-0.0)