From e351aca041533866cd10b544133f91e65f4deca3 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 1 Nov 2026 11:34:23 +0100 Subject: [PATCH] feat(detection-core): replay admission evidence --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/_admission_gates.py | 341 ++++++++++ src/openultrasast/core/_admission_replay.py | 136 ++++ src/openultrasast/core/admission_metrics.py | 8 +- src/openultrasast/core/admission_report.py | 424 ++++++++++++ src/openultrasast/core/measurement_record.py | 2 +- tests/test_core_admission_metrics.py | 19 +- tests/test_core_admission_report.py | 670 +++++++++++++++++++ tests/test_core_measurement_record.py | 4 +- 9 files changed, 1595 insertions(+), 11 deletions(-) create mode 100644 src/openultrasast/core/_admission_gates.py create mode 100644 src/openultrasast/core/_admission_replay.py create mode 100644 src/openultrasast/core/admission_report.py create mode 100644 tests/test_core_admission_report.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 55d714d..18c66e7 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -280,7 +280,7 @@ - Done when forged decisions, substituted contracts, invalid fact review, contaminated lineage, or case-count drift fail before metric production. - _Requirements: 12.5, 12.6, 15.1, 15.2_ - _Depends: 8.7_ -- [ ] 8.9 Assemble evidence-backed gates and the final admission report +- [x] 8.9 Assemble evidence-backed gates and the final admission report - Emit instrument, budget, readiness, and quality gates with reasons and supporting identities. - Bind the report to the validation context, authorization, pre-label spend receipt, authenticated case release, census, and complete case-result set. - Require every mandatory metric and gate before allowing GO; make model success mandatory when a model is baseline-configured or validated claimed-scope calls occur. Refuse undeclared claimed-scope calls before report production; make configured zero-call, missing-threshold, and failed declared-call cases NO-GO while ignoring out-of-scope-only calls. diff --git a/src/openultrasast/core/_admission_gates.py b/src/openultrasast/core/_admission_gates.py new file mode 100644 index 0000000..d01a4c2 --- /dev/null +++ b/src/openultrasast/core/_admission_gates.py @@ -0,0 +1,341 @@ +"""Typed evidence and re-derived gates for final admission reports.""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass +from enum import Enum + +from openultrasast.core import canonical +from openultrasast.core.admission_case import ( + AdmissionCaseResult, + derive_admission_case_result_digest, + validate_admission_case_shape, +) +from openultrasast.core.admission_review import AdmissionThreshold, derive_threshold_set_digest +from openultrasast.core.coverage import validate_instrument +from openultrasast.core.measurement import ( + BaselineIdentity, + Comparator, + MetricId, + MetricResult, + Statistic, + metric_passes, +) +from openultrasast.core.measurement_record import MeasurementRecordV2, ReadinessReference +from openultrasast.core.readiness import StageResult + +BUDGET_EVALUATION_SCHEMA = "openultrasast.detection-core.admission-budget-evaluation.v2" +THRESHOLD_EVALUATION_SCHEMA = "openultrasast.detection-core.admission-threshold-evaluation.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") +_MANDATORY_METRICS = tuple(MetricId)[:6] + + +class AdmissionReportError(ValueError): + """Admission evidence cannot form a valid final report.""" + + +class GateKind(Enum): + INSTRUMENT = "instrument" + BUDGET = "budget" + QUALITY = "quality" + READINESS = "readiness" + + +@dataclass(frozen=True) +class GateResult: + kind: GateKind + gate_id: str + passed: bool + reasons: tuple[str, ...] + evidence_digests: tuple[str, ...] + + def __post_init__(self) -> None: + validate_gate(self) + + +@dataclass(frozen=True) +class CaseBudgetEvaluation: + case: AdmissionCaseResult + case_result_digest: str + budget_derivation_digest: str + total_reply_ms: int + budget_ms: int + evaluation_digest: str + + def __post_init__(self) -> None: + validate_budget_evaluation(self) + + +@dataclass(frozen=True) +class ThresholdEvaluation: + metric_id: MetricId + statistic: Statistic + comparator: Comparator + target: float + measurement_record_digest: str + passed: bool + evaluation_digest: str + + def __post_init__(self) -> None: + validate_threshold_evaluation(self) + + +def _digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise AdmissionReportError(f"{field} must be a full lowercase SHA-256 digest") + + +def _canonical_digests(value: object, field: str, *, allow_empty: bool = False) -> tuple[str, ...]: + if not isinstance(value, tuple) or not all(isinstance(item, str) for item in value): + raise AdmissionReportError(f"{field} must be a tuple of digests") + for item in value: + _digest(item, field.removesuffix("s")) + if value != tuple(sorted(set(value))): + raise AdmissionReportError(f"{field} must be unique and canonical") + if not allow_empty and not value: + raise AdmissionReportError(f"{field} must not be empty") + return value + + +def validate_gate(value: GateResult) -> None: + if type(value) is not GateResult: + raise AdmissionReportError("gate must be an exact GateResult") + if not isinstance(value.kind, GateKind) or not isinstance(value.gate_id, str) or not value.gate_id: + raise AdmissionReportError("gate kind and gate_id must use closed nonempty values") + if not isinstance(value.passed, bool): + raise AdmissionReportError("gate passed must be boolean") + if not isinstance(value.reasons, tuple) or not all(isinstance(item, str) and item for item in value.reasons): + raise AdmissionReportError("gate reasons must be a tuple of nonempty text") + if value.reasons != tuple(sorted(set(value.reasons))): + raise AdmissionReportError("gate reasons must be unique and canonical") + if value.passed == bool(value.reasons): + raise AdmissionReportError("passed gates have no reasons and failed gates require reasons") + _canonical_digests(value.evidence_digests, "gate evidence digests") + + +def gate_value(value: GateResult) -> dict[str, canonical.JsonValue]: + validate_gate(value) + return { + "kind": value.kind.value, + "gate_id": value.gate_id, + "passed": value.passed, + "reasons": value.reasons, + "evidence_digests": value.evidence_digests, + } + + +def _make_gate(kind: GateKind, gate_id: str, reasons: tuple[str, ...], evidence: tuple[str, ...]) -> GateResult: + canonical_reasons = tuple(sorted(set(reasons))) + canonical_evidence = tuple(sorted(set(evidence))) + return GateResult(kind, gate_id, not canonical_reasons, canonical_reasons, canonical_evidence) + + +def budget_evaluation_value(value: CaseBudgetEvaluation) -> dict[str, canonical.JsonValue]: + return { + "schema_version": BUDGET_EVALUATION_SCHEMA, + "case_result_digest": value.case_result_digest, + "budget_derivation_digest": value.budget_derivation_digest, + "total_reply_ms": value.total_reply_ms, + "budget_ms": value.budget_ms, + } + + +def validate_budget_evaluation(value: CaseBudgetEvaluation) -> None: + if type(value) is not CaseBudgetEvaluation: + raise AdmissionReportError("budget evaluation must use the exact type") + if type(value.case) is not AdmissionCaseResult: + raise AdmissionReportError("budget evaluation must carry the exact admission case") + validate_admission_case_shape(value.case) + if value.case.result_digest != derive_admission_case_result_digest(value.case): + raise AdmissionReportError("budget evaluation case result digest cannot be re-derived") + _digest(value.case_result_digest, "case result digest") + _digest(value.budget_derivation_digest, "budget derivation digest") + for item, field in ((value.total_reply_ms, "total reply milliseconds"), (value.budget_ms, "budget milliseconds")): + if isinstance(item, bool) or not isinstance(item, int) or item <= 0: + raise AdmissionReportError(f"{field} must be a positive integer") + _digest(value.evaluation_digest, "budget evaluation digest") + expected = ( + value.case.result_digest, + value.case.budget.derivation_digest, + value.case.total_reply_ms, + value.case.budget.budget_ms, + ) + actual = (value.case_result_digest, value.budget_derivation_digest, value.total_reply_ms, value.budget_ms) + if actual != expected: + raise AdmissionReportError("budget evaluation does not match the exact case-to-budget derivation") + if value.evaluation_digest != canonical.content_digest(budget_evaluation_value(value)): + raise AdmissionReportError("budget evaluation digest does not match content") + + +def make_budget_evaluations(cases: tuple[AdmissionCaseResult, ...]) -> tuple[CaseBudgetEvaluation, ...]: + values = [] + for case in cases: + fields = ( + case, + case.result_digest, + case.budget.derivation_digest, + case.total_reply_ms, + case.budget.budget_ms, + ) + draft = object.__new__(CaseBudgetEvaluation) + for field, item in zip(CaseBudgetEvaluation.__dataclass_fields__, (*fields, "0" * 64), strict=True): + object.__setattr__(draft, field, item) + digest = canonical.content_digest(budget_evaluation_value(draft)) + values.append(CaseBudgetEvaluation(*fields, digest)) + return tuple(sorted(values, key=lambda item: item.case_result_digest)) + + +def threshold_evaluation_value(value: ThresholdEvaluation) -> dict[str, canonical.JsonValue]: + return { + "schema_version": THRESHOLD_EVALUATION_SCHEMA, + "metric_id": value.metric_id.value, + "statistic": value.statistic.value, + "comparator": value.comparator.value, + "target": value.target, + "measurement_record_digest": value.measurement_record_digest, + "passed": value.passed, + } + + +def validate_threshold_evaluation(value: ThresholdEvaluation) -> None: + if type(value) is not ThresholdEvaluation: + raise AdmissionReportError("threshold evaluation must use the exact type") + if not isinstance(value.metric_id, MetricId) or not isinstance(value.statistic, Statistic): + raise AdmissionReportError("threshold evaluation metric and statistic must use closed enums") + if not isinstance(value.comparator, Comparator): + raise AdmissionReportError("threshold evaluation comparator must use the closed enum") + if isinstance(value.target, bool) or not isinstance(value.target, float) or not math.isfinite(value.target): + raise AdmissionReportError("threshold evaluation target must be a finite float") + _digest(value.measurement_record_digest, "threshold measurement record digest") + if not isinstance(value.passed, bool): + raise AdmissionReportError("threshold evaluation passed must be boolean") + _digest(value.evaluation_digest, "threshold evaluation digest") + if value.evaluation_digest != canonical.content_digest(threshold_evaluation_value(value)): + raise AdmissionReportError("threshold evaluation digest does not match content") + + +def make_threshold_evaluations( + thresholds: tuple[AdmissionThreshold, ...], + measurement: MeasurementRecordV2, +) -> tuple[ThresholdEvaluation, ...]: + by_metric = {item.metric_id: item for item in measurement.metrics} + values = [] + for threshold in thresholds: + passed = metric_passes(by_metric[threshold.metric_id], threshold.statistic, threshold.comparator, threshold.target) + fields = ( + threshold.metric_id, + threshold.statistic, + threshold.comparator, + threshold.target, + measurement.record_digest, + passed, + ) + draft = object.__new__(ThresholdEvaluation) + for field, item in zip(ThresholdEvaluation.__dataclass_fields__, (*fields, "0" * 64), strict=True): + object.__setattr__(draft, field, item) + digest = canonical.content_digest(threshold_evaluation_value(draft)) + values.append(ThresholdEvaluation(*fields, digest)) + return tuple(values) + + +def _thresholds(evaluations: tuple[ThresholdEvaluation, ...]) -> tuple[AdmissionThreshold, ...]: + return tuple(AdmissionThreshold(item.metric_id, item.statistic, item.comparator, item.target) for item in evaluations) + + +def instrument_gate( + case_result_digests: tuple[str, ...], + cases: tuple[AdmissionCaseResult, ...], +) -> GateResult: + evidence = _canonical_digests(case_result_digests, "case result digests") + identities = tuple(sorted(item.result_digest for item in cases)) + if identities != evidence or len(identities) != len(cases): + raise AdmissionReportError("instrument cases must cover the exact complete case-result set") + try: + for case in cases: + validate_instrument(case.input_manifest, case.input_receipts) + except (TypeError, ValueError) as exc: + raise AdmissionReportError(f"instrument gate evidence is invalid: {exc}") from exc + return _make_gate(GateKind.INSTRUMENT, "instrument:all_cases", (), evidence) + + +def budget_gate( + case_result_digests: tuple[str, ...], + evaluations: tuple[CaseBudgetEvaluation, ...], +) -> GateResult: + if not isinstance(evaluations, tuple) or not all(type(item) is CaseBudgetEvaluation for item in evaluations): + raise AdmissionReportError("budget evaluations must use the exact typed tuple") + for item in evaluations: + validate_budget_evaluation(item) + identities = tuple(item.case_result_digest for item in evaluations) + if identities != case_result_digests or len(set(identities)) != len(identities): + raise AdmissionReportError("budget evaluations must cover the exact complete case-result set") + reasons = tuple(f"budget_exceeded:{item.case_result_digest}" for item in evaluations if item.total_reply_ms > item.budget_ms) + evidence = tuple(item.evaluation_digest for item in evaluations) + return _make_gate(GateKind.BUDGET, "budget:all_cases", reasons, evidence) + + +def readiness_gate(references: tuple[ReadinessReference, ...]) -> GateResult: + reasons = tuple(f"readiness_failed:{item.stage.value}" for item in references if item.result is not StageResult.PASSED) + evidence = tuple(item.result_digest for item in references) + return _make_gate(GateKind.READINESS, "readiness:prerequisites", reasons, evidence) + + +def quality_gate( + identity: BaselineIdentity, + measurement: MeasurementRecordV2, + evaluations: tuple[ThresholdEvaluation, ...], +) -> GateResult: + if not isinstance(evaluations, tuple) or not all(type(item) is ThresholdEvaluation for item in evaluations): + raise AdmissionReportError("quality threshold evaluations must use the exact typed tuple") + by_metric = {item.metric_id: item for item in measurement.metrics} + reasons = _mandatory_metric_reasons(by_metric) + model = by_metric[MetricId.MODEL_SUCCESS_RATE] + reasons.extend(_model_reasons(identity, model, evaluations)) + for item in evaluations: + validate_threshold_evaluation(item) + if item.measurement_record_digest != measurement.record_digest: + raise AdmissionReportError("threshold evaluation does not bind the exact measurement") + expected = metric_passes(by_metric[item.metric_id], item.statistic, item.comparator, item.target) + if item.passed is not expected: + raise AdmissionReportError("threshold evaluation outcome does not match the exact metric") + if not expected: + reasons.append(f"threshold_failed:{item.metric_id.value}:{item.statistic.value}") + thresholds = _thresholds(evaluations) + try: + threshold_set_digest = derive_threshold_set_digest(thresholds) + except ValueError as exc: + raise AdmissionReportError(f"quality threshold evaluations are invalid: {exc}") from exc + if threshold_set_digest != identity.threshold_set_digest: + raise AdmissionReportError("threshold evaluations do not match the exact threshold set") + evidence = (identity.threshold_set_digest, measurement.record_digest, *(item.evaluation_digest for item in evaluations)) + return _make_gate(GateKind.QUALITY, "quality:threshold_set", tuple(reasons), evidence) + + +def _model_reasons( + identity: BaselineIdentity, + model: MetricResult, + evaluations: tuple[ThresholdEvaluation, ...], +) -> list[str]: + reasons: list[str] = [] + model_configured = bool(identity.model_program_digests) + model_executed = model.denominator > 0 + model_applicable = model_configured or model_executed + model_threshold_count = sum(item.metric_id is MetricId.MODEL_SUCCESS_RATE for item in evaluations) + if model_applicable and model_threshold_count == 0: + reasons.append("mandatory_model_success_threshold_missing") + if model_configured and not model_executed: + reasons.append("mandatory_model_success_has_no_attempted_calls") + if model_executed and model.numerator < model.denominator: + reasons.append("mandatory_model_call_failed") + return reasons + + +def _mandatory_metric_reasons(by_metric: dict[MetricId, MetricResult]) -> list[str]: + reasons = [] + for metric in _MANDATORY_METRICS: + result = by_metric.get(metric) + if result is None or result.denominator == 0: + reasons.append(f"mandatory_metric_missing_or_zero:{metric.value}") + return reasons diff --git a/src/openultrasast/core/_admission_replay.py b/src/openultrasast/core/_admission_replay.py new file mode 100644 index 0000000..14a19d1 --- /dev/null +++ b/src/openultrasast/core/_admission_replay.py @@ -0,0 +1,136 @@ +"""Immutable transport identity for authoritative admission replay inputs.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from openultrasast.core import canonical +from openultrasast.core.admission_case import ( + AdmissionCaseRelease, + AdmissionCaseResult, + derive_admission_case_release_digest, + derive_admission_case_result_digest, +) +from openultrasast.core.admission_review import AdmissionManifestV2, derive_manifest_digest +from openultrasast.core.admission_spend import AdmissionPreparation, derive_admission_preparation_digest +from openultrasast.core.custody import ( + PopulationUseReceipt, + derive_population_use_receipt_digest, +) +from openultrasast.core.measurement import DenominatorPolicy, derive_denominator_policy_digest +from openultrasast.core.readiness import ReadinessResult, validate_readiness_result + +if TYPE_CHECKING: + from openultrasast.core.admission_report import AdmissionReportV2 + +ADMISSION_REPLAY_SCHEMA = "openultrasast.detection-core.admission-replay.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +@dataclass(frozen=True) +class AdmissionReplay: + manifest: AdmissionManifestV2 + denominator: DenominatorPolicy + readiness: tuple[ReadinessResult, ...] + preparation: AdmissionPreparation + label_release_receipt: PopulationUseReceipt + case_release: AdmissionCaseRelease + cases: tuple[AdmissionCaseResult, ...] + report: AdmissionReportV2 + replay_digest: str + + def __post_init__(self) -> None: + validate_admission_replay_structure(self) + + +def _replay_value(value: AdmissionReplay) -> canonical.JsonValue: + manifest_digest = derive_manifest_digest(value.manifest) + denominator_digest = derive_denominator_policy_digest(value.denominator) + preparation_digest = derive_admission_preparation_digest(value.preparation) + receipt_digest = derive_population_use_receipt_digest(value.label_release_receipt) + release_digest = derive_admission_case_release_digest(value.case_release) + for item in value.readiness: + validate_readiness_result(item) + case_digests = tuple(derive_admission_case_result_digest(item) for item in value.cases) + claimed = ( + value.manifest.manifest_digest, + value.denominator.policy_digest, + value.preparation.preparation_digest, + value.label_release_receipt.receipt_digest, + value.case_release.release_digest, + tuple(item.result_digest for item in value.cases), + ) + derived = (manifest_digest, denominator_digest, preparation_digest, receipt_digest, release_digest, case_digests) + if claimed != derived: + raise ValueError("replay component identity cannot be re-derived") + return { + "schema_version": ADMISSION_REPLAY_SCHEMA, + "manifest_digest": manifest_digest, + "denominator_policy_digest": denominator_digest, + "readiness_result_digests": tuple(item.result_digest for item in value.readiness), + "preparation_digest": preparation_digest, + "label_release_receipt_digest": receipt_digest, + "case_release_digest": release_digest, + "case_result_digests": case_digests, + "report_digest": value.report.report_digest, + } + + +def _validate_components(value: AdmissionReplay) -> None: + exact_components = ( + (value.manifest, AdmissionManifestV2, "manifest"), + (value.denominator, DenominatorPolicy, "denominator"), + (value.preparation, AdmissionPreparation, "preparation"), + (value.label_release_receipt, PopulationUseReceipt, "label release receipt"), + (value.case_release, AdmissionCaseRelease, "case release"), + ) + for item, expected, field in exact_components: + if type(item) is not expected: + raise ValueError(f"replay {field} must use the exact contract type") + if not all(type(item) is ReadinessResult for item in value.readiness): + raise ValueError("replay readiness must contain exact ReadinessResult values") + if not all(type(item) is AdmissionCaseResult for item in value.cases): + raise ValueError("replay cases must contain exact AdmissionCaseResult values") + case_ids = tuple(item.case_id_digest for item in value.cases) + if case_ids != tuple(sorted(set(case_ids))): + raise ValueError("replay cases must use unique canonical case identity order") + + +def validate_admission_replay_structure(value: AdmissionReplay) -> None: + from openultrasast.core.admission_report import AdmissionReportError, AdmissionReportV2, _validate_report_structure + + if type(value) is not AdmissionReplay: + raise AdmissionReportError("replay must be an exact AdmissionReplay") + if type(value.readiness) is not tuple or not value.readiness: + raise AdmissionReportError("replay readiness must be an exact nonempty tuple") + if type(value.cases) is not tuple or not value.cases: + raise AdmissionReportError("replay cases must be an exact nonempty tuple") + try: + _validate_components(value) + except (AttributeError, TypeError, ValueError) as exc: + raise AdmissionReportError(str(exc)) from exc + if type(value.report) is not AdmissionReportV2: + raise AdmissionReportError("replay report must be an exact AdmissionReportV2") + _validate_report_structure(value.report) + if not isinstance(value.replay_digest, str) or _DIGEST.fullmatch(value.replay_digest) is None: + raise AdmissionReportError("replay digest must be a full lowercase SHA-256 digest") + try: + expected = canonical.content_digest(_replay_value(value)) + except (AttributeError, TypeError, ValueError) as exc: + raise AdmissionReportError(str(exc)) from exc + if value.replay_digest != expected: + raise AdmissionReportError("replay digest does not match authoritative input identities") + + +def admission_replay_value(value: AdmissionReplay) -> canonical.JsonValue: + validate_admission_replay_structure(value) + return _replay_value(value) + + +__all__ = [ + "ADMISSION_REPLAY_SCHEMA", + "AdmissionReplay", + "admission_replay_value", +] diff --git a/src/openultrasast/core/admission_metrics.py b/src/openultrasast/core/admission_metrics.py index f04d079..a643916 100644 --- a/src/openultrasast/core/admission_metrics.py +++ b/src/openultrasast/core/admission_metrics.py @@ -24,7 +24,7 @@ from openultrasast.core.admission_case import ( from openultrasast.core.admission_context import AdmissionValidationContext from openultrasast.core.admission_review import AdmissionManifestV2, ScopeCensus, validate_reviewed_admission_payloads from openultrasast.core.admission_spend import AdmissionPreparation -from openultrasast.core.budget import check_budget +from openultrasast.core.budget import BudgetGateFailure, check_budget from openultrasast.core.coverage import CoverageFunnel, ReasonCount, validate_instrument from openultrasast.core.custody import PopulationCustodyPort, PopulationUseReceipt from openultrasast.core.evidence import EvidenceBasis, EvidenceRecord, NeutralOutcome, derive_packet_digest @@ -139,6 +139,8 @@ def _validated_records( raise AdmissionMetricsError(f"instrument validation failed: {exc}") from exc try: check_budget(case.complexity, case.total_reply_ms, case.budget, resolver) + except BudgetGateFailure: + pass except (TypeError, ValueError) as exc: raise AdmissionMetricsError(f"budget validation failed: {exc}") from exc try: @@ -206,9 +208,7 @@ def _validate_population(cases: tuple[_ValidatedCase, ...], rows: tuple[ScopeCen for row in rows: selected = tuple(item for item in cases if item.binding.scope == row.scope) roles = Counter(item.binding.role for item in selected) - model_calls = sum( - value.model_result is not None for item in selected for value in item.supplied.evidence_inputs - ) + model_calls = sum(value.model_result is not None for item in selected for value in item.supplied.evidence_inputs) actual = ( len(selected), roles[AdmissionCaseRole.VULNERABLE], diff --git a/src/openultrasast/core/admission_report.py b/src/openultrasast/core/admission_report.py new file mode 100644 index 0000000..1cfdf2f --- /dev/null +++ b/src/openultrasast/core/admission_report.py @@ -0,0 +1,424 @@ +"""Evidence-backed admission gates and canonical final reports.""" + +from __future__ import annotations + +import re +from collections.abc import Sequence +from dataclasses import dataclass +from enum import Enum + +from openultrasast.core import canonical +from openultrasast.core._admission_gates import ( + AdmissionReportError, + CaseBudgetEvaluation, + GateKind, + GateResult, + ThresholdEvaluation, + budget_evaluation_value, + budget_gate, + gate_value, + instrument_gate, + make_budget_evaluations, + make_threshold_evaluations, + quality_gate, + readiness_gate, + threshold_evaluation_value, +) +from openultrasast.core._admission_replay import ( + ADMISSION_REPLAY_SCHEMA, + AdmissionReplay, + _replay_value, + admission_replay_value, + validate_admission_replay_structure, +) +from openultrasast.core.admission_case import AdmissionCaseRelease, AdmissionCaseResult +from openultrasast.core.admission_context import AdmissionValidationContext +from openultrasast.core.admission_metrics import AdmissionMetricEvidence, derive_admission_metrics +from openultrasast.core.admission_review import AdmissionManifestV2 +from openultrasast.core.admission_spend import AdmissionPreparation +from openultrasast.core.custody import PopulationCustodyPort, PopulationUseReceipt +from openultrasast.core.measurement import ( + BaselineIdentity, + DenominatorPolicy, + derive_baseline_identity_digest, +) +from openultrasast.core.measurement_factory import create_measurement_record +from openultrasast.core.measurement_record import MeasurementRecordV2, ReadinessReference, measurement_record_value +from openultrasast.core.readiness import ReadinessResult, ReadinessStage + +ADMISSION_REPORT_SCHEMA = "openultrasast.detection-core.admission-report.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") +_PREREQUISITE_STAGES = tuple(ReadinessStage)[:6] + + +class AdmissionDecision(Enum): + GO = "go" + NO_GO = "no_go" + + +@dataclass(frozen=True) +class AdmissionReportV2: + identity: BaselineIdentity + census_digest: str + validation_context_digest: str + authorization_digest: str + label_release_receipt_digest: str + case_release_digest: str + case_result_digests: tuple[str, ...] + in_scope_case_count: int + whole_population_case_count: int + budget_evaluations: tuple[CaseBudgetEvaluation, ...] + quality_evaluations: tuple[ThresholdEvaluation, ...] + measurement: MeasurementRecordV2 + instrument_results: tuple[GateResult, ...] + budget_results: tuple[GateResult, ...] + readiness_results: tuple[GateResult, ...] + quality_results: tuple[GateResult, ...] + decision: AdmissionDecision + report_digest: str + + def __post_init__(self) -> None: + _validate_report_structure(self) + + +def _digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise AdmissionReportError(f"{field} must be a full lowercase SHA-256 digest") + + +def _case_digests(value: object, count: int) -> tuple[str, ...]: + if not isinstance(value, tuple) or not all(isinstance(item, str) for item in value): + raise AdmissionReportError("case result digests must be a tuple of digests") + for item in value: + _digest(item, "case result digest") + if value != tuple(sorted(set(value))) or len(value) != count: + raise AdmissionReportError("case result digests must identify every case exactly once in canonical order") + return value + + +def _report_value(value: AdmissionReportV2) -> canonical.JsonValue: + return { + "schema_version": ADMISSION_REPORT_SCHEMA, + "baseline_identity_digest": derive_baseline_identity_digest(value.identity), + "census_digest": value.census_digest, + "validation_context_digest": value.validation_context_digest, + "authorization_digest": value.authorization_digest, + "label_release_receipt_digest": value.label_release_receipt_digest, + "case_release_digest": value.case_release_digest, + "case_result_digests": value.case_result_digests, + "in_scope_case_count": value.in_scope_case_count, + "whole_population_case_count": value.whole_population_case_count, + "budget_evaluations": tuple( + {**budget_evaluation_value(item), "evaluation_digest": item.evaluation_digest} for item in value.budget_evaluations + ), + "quality_evaluations": tuple( + {**threshold_evaluation_value(item), "evaluation_digest": item.evaluation_digest} for item in value.quality_evaluations + ), + "measurement_record_digest": value.measurement.record_digest, + "instrument_results": tuple(gate_value(item) for item in value.instrument_results), + "budget_results": tuple(gate_value(item) for item in value.budget_results), + "readiness_results": tuple(gate_value(item) for item in value.readiness_results), + "quality_results": tuple(gate_value(item) for item in value.quality_results), + "decision": value.decision.value, + } + + +def _validate_report_bindings(value: AdmissionReportV2) -> None: + baseline_digest = derive_baseline_identity_digest(value.identity) + if baseline_digest != derive_baseline_identity_digest(value.measurement.identity): + raise AdmissionReportError("measurement baseline identity does not match report") + measurement_record_value(value.measurement) + for item, field in ( + (value.census_digest, "census digest"), + (value.validation_context_digest, "validation context digest"), + (value.authorization_digest, "authorization digest"), + (value.label_release_receipt_digest, "label release receipt digest"), + (value.case_release_digest, "case release digest"), + ): + _digest(item, field) + if isinstance(value.in_scope_case_count, bool) or not isinstance(value.in_scope_case_count, int): + raise AdmissionReportError("in-scope case count must be an integer") + if isinstance(value.whole_population_case_count, bool) or not isinstance(value.whole_population_case_count, int): + raise AdmissionReportError("whole-population case count must be an integer") + if not 0 < value.in_scope_case_count <= value.whole_population_case_count: + raise AdmissionReportError("case counts must be positive and in-scope cannot exceed whole-population") + _case_digests(value.case_result_digests, value.whole_population_case_count) + + +def _require_exact_gate(value: object, expected: GateResult) -> tuple[GateResult, ...]: + if not isinstance(value, tuple) or value != (expected,): + raise AdmissionReportError(f"{expected.kind.value} gate does not match its derived evidence and outcome") + return value + + +def _derived_gates(value: AdmissionReportV2) -> tuple[GateResult, GateResult, GateResult, GateResult]: + cases = tuple(item.case for item in value.budget_evaluations) + return ( + instrument_gate(value.case_result_digests, cases), + budget_gate(value.case_result_digests, value.budget_evaluations), + readiness_gate(value.measurement.prerequisite_readiness), + quality_gate(value.identity, value.measurement, value.quality_evaluations), + ) + + +def _validate_report_fields(value: AdmissionReportV2) -> None: + if type(value) is not AdmissionReportV2: + raise AdmissionReportError("report must be an exact AdmissionReportV2") + _validate_report_bindings(value) + instrument, budget, readiness, quality = _derived_gates(value) + actual = ( + _require_exact_gate(value.instrument_results, instrument), + _require_exact_gate(value.budget_results, budget), + _require_exact_gate(value.readiness_results, readiness), + _require_exact_gate(value.quality_results, quality), + ) + if not isinstance(value.decision, AdmissionDecision): + raise AdmissionReportError("decision must use AdmissionDecision") + all_passed = all(item.passed for group in actual for item in group) + expected = AdmissionDecision.GO if all_passed else AdmissionDecision.NO_GO + if value.decision is not expected: + raise AdmissionReportError("decision does not match every re-derived mandatory gate") + + +def _validate_report_structure(value: AdmissionReportV2) -> None: + _validate_report_fields(value) + _digest(value.report_digest, "report digest") + if value.report_digest != canonical.content_digest(_report_value(value)): + raise AdmissionReportError("report digest does not match admission content") + + +def admission_report_value(value: AdmissionReportV2) -> canonical.JsonValue: + _validate_report_structure(value) + return _report_value(value) + + +def _readiness_references(readiness: tuple[ReadinessResult, ...]) -> tuple[ReadinessReference, ...]: + references = tuple(ReadinessReference(item.stage, item.result, item.result_digest) for item in readiness) + if tuple(item.stage for item in references) != _PREREQUISITE_STAGES: + raise AdmissionReportError("admission measurement requires instrument through development readiness") + return references + + +def _measurement( + manifest: AdmissionManifestV2, + evidence: AdmissionMetricEvidence, + readiness: tuple[ReadinessResult, ...], +) -> MeasurementRecordV2: + return create_measurement_record( + identity=manifest.identity, + family_decision_vocabulary=evidence.family_decision_vocabulary, + family_decision_counts=evidence.family_decision_counts, + neutral_outcome_counts=evidence.neutral_outcome_counts, + evidence_basis_counts=evidence.evidence_basis_counts, + metrics=evidence.metrics, + stage_funnel=evidence.stage_funnel, + instrument_failures=(), + model_outcome_counts=evidence.model_outcome_counts, + model_failures=evidence.model_failures, + runtime_ms=evidence.runtime_ms, + prerequisite_readiness=_readiness_references(readiness), + ) + + +def _quality_gate(manifest: AdmissionManifestV2, measurement: MeasurementRecordV2) -> GateResult: + evaluations = make_threshold_evaluations(manifest.thresholds, measurement) + return quality_gate(manifest.identity, measurement, evaluations) + + +def _make_report( + manifest: AdmissionManifestV2, + preparation: AdmissionPreparation, + receipt: PopulationUseReceipt, + release: AdmissionCaseRelease, + cases: tuple[AdmissionCaseResult, ...], + evidence: AdmissionMetricEvidence, + measurement: MeasurementRecordV2, +) -> AdmissionReportV2: + budgets = make_budget_evaluations(cases) + quality = make_threshold_evaluations(manifest.thresholds, measurement) + gates = ( + instrument_gate(evidence.case_result_digests, cases), + budget_gate(evidence.case_result_digests, budgets), + readiness_gate(measurement.prerequisite_readiness), + quality_gate(manifest.identity, measurement, quality), + ) + decision = AdmissionDecision.GO if all(item.passed for item in gates) else AdmissionDecision.NO_GO + values = ( + manifest.identity, + manifest.census.census_digest, + preparation.authorization.validation_context_digest, + preparation.authorization.authorization_digest, + receipt.receipt_digest, + release.release_digest, + evidence.case_result_digests, + evidence.in_scope_case_count, + evidence.whole_population_case_count, + budgets, + quality, + measurement, + (gates[0],), + (gates[1],), + (gates[2],), + (gates[3],), + decision, + ) + draft = object.__new__(AdmissionReportV2) + fields = tuple(AdmissionReportV2.__dataclass_fields__) + for field, item in zip(fields, (*values, "0" * 64), strict=True): + object.__setattr__(draft, field, item) + _validate_report_fields(draft) + return AdmissionReportV2(*values, canonical.content_digest(_report_value(draft))) + + +def score_admission( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + readiness: Sequence[ReadinessResult], + preparation: AdmissionPreparation, + label_release_receipt: PopulationUseReceipt, + case_release: AdmissionCaseRelease, + cases: Sequence[AdmissionCaseResult], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> AdmissionReplay: + """Revalidate exact admission evidence and freeze its authoritative replay.""" + frozen_readiness = tuple(readiness) + frozen_cases = tuple(sorted(tuple(cases), key=lambda item: item.case_id_digest)) + report = _derive_report( + manifest, + denominator, + frozen_readiness, + preparation, + label_release_receipt, + case_release, + frozen_cases, + context, + custody, + resolver, + ) + return _make_replay( + manifest, + denominator, + frozen_readiness, + preparation, + label_release_receipt, + case_release, + frozen_cases, + report, + ) + + +def _derive_report( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + readiness: tuple[ReadinessResult, ...], + preparation: AdmissionPreparation, + label_release_receipt: PopulationUseReceipt, + case_release: AdmissionCaseRelease, + cases: tuple[AdmissionCaseResult, ...], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> AdmissionReportV2: + evidence = derive_admission_metrics( + manifest, + denominator, + readiness, + preparation, + label_release_receipt, + case_release, + cases, + context, + custody, + resolver, + ) + measurement = _measurement(manifest, evidence, readiness) + return _make_report( + manifest, + preparation, + label_release_receipt, + case_release, + cases, + evidence, + measurement, + ) + + +def _make_replay( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + readiness: tuple[ReadinessResult, ...], + preparation: AdmissionPreparation, + label_release_receipt: PopulationUseReceipt, + case_release: AdmissionCaseRelease, + cases: tuple[AdmissionCaseResult, ...], + report: AdmissionReportV2, +) -> AdmissionReplay: + values = (manifest, denominator, readiness, preparation, label_release_receipt, case_release, cases, report) + draft = object.__new__(AdmissionReplay) + for field, item in zip(AdmissionReplay.__dataclass_fields__, (*values, "0" * 64), strict=True): + object.__setattr__(draft, field, item) + return AdmissionReplay(*values, canonical.content_digest(_replay_value(draft))) + + +def validate_admission_replay( + replay: AdmissionReplay, + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> AdmissionReportV2: + """Re-run every carried input against independently supplied current trust state.""" + validate_admission_replay_structure(replay) + try: + expected = _derive_report( + replay.manifest, + replay.denominator, + replay.readiness, + replay.preparation, + replay.label_release_receipt, + replay.case_release, + replay.cases, + context, + custody, + resolver, + ) + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise AdmissionReportError(f"authoritative admission replay refused: {exc}") from exc + actual_bytes = canonical.canonical_bytes(_report_value(replay.report)) + expected_bytes = canonical.canonical_bytes(_report_value(expected)) + if replay.report != expected or actual_bytes != expected_bytes: + raise AdmissionReportError("report does not match the authoritative admission inputs") + expected_replay = _make_replay( + replay.manifest, + replay.denominator, + replay.readiness, + replay.preparation, + replay.label_release_receipt, + replay.case_release, + replay.cases, + expected, + ) + if replay != expected_replay or canonical.canonical_bytes(_replay_value(replay)) != canonical.canonical_bytes( + _replay_value(expected_replay) + ): + raise AdmissionReportError("replay does not match the authoritative admission inputs") + return expected + + +__all__ = [ + "ADMISSION_REPLAY_SCHEMA", + "ADMISSION_REPORT_SCHEMA", + "AdmissionReplay", + "AdmissionDecision", + "AdmissionReportError", + "AdmissionReportV2", + "CaseBudgetEvaluation", + "GateKind", + "GateResult", + "ThresholdEvaluation", + "admission_replay_value", + "admission_report_value", + "score_admission", + "validate_admission_replay", +] diff --git a/src/openultrasast/core/measurement_record.py b/src/openultrasast/core/measurement_record.py index a844d3c..015654f 100644 --- a/src/openultrasast/core/measurement_record.py +++ b/src/openultrasast/core/measurement_record.py @@ -20,7 +20,7 @@ from openultrasast.core.readiness import ReadinessEvidenceUse, ReadinessStage, S MEASUREMENT_RECORD_SCHEMA = "openultrasast.detection-core.measurement-record.v2" REPOSITORY_HOLDOUT_REPORT_SCHEMA = "openultrasast.detection-core.repository-holdout-report.v2" _DIGEST = re.compile(r"[0-9a-f]{64}") -_PREREQUISITE_STAGES = tuple(ReadinessStage)[:5] +_PREREQUISITE_STAGES = tuple(ReadinessStage)[:6] class MeasurementRecordError(ValueError): diff --git a/tests/test_core_admission_metrics.py b/tests/test_core_admission_metrics.py index 5a1d83f..35c121b 100644 --- a/tests/test_core_admission_metrics.py +++ b/tests/test_core_admission_metrics.py @@ -405,7 +405,7 @@ def test_supplied_verdict_and_instrument_are_not_trusted() -> None: _derive(run, cases=broken_cases, release=_release_for(run, broken_cases)) -def test_forged_evidence_and_budget_fail_before_metric_production() -> None: +def test_forged_evidence_fails_but_valid_budget_overrun_reaches_metrics() -> None: run = _run(_setup()) cases = run.setup.cases forged_input = replace(cases[0].evidence_inputs[0], input_digest="0" * 64) @@ -420,8 +420,21 @@ def test_forged_evidence_and_budget_fail_before_metric_production() -> None: over_budget = replace(cases[0], total_reply_ms=cases[0].budget.budget_ms + 1) over_budget = replace(over_budget, result_digest=derive_admission_case_result_digest(over_budget)) budget_cases = (over_budget, *cases[1:]) - with pytest.raises(AdmissionMetricsError, match="budget"): - _derive(run, cases=budget_cases, release=_release_for(run, budget_cases)) + result = _derive(run, cases=budget_cases, release=_release_for(run, budget_cases)) + + assert result.runtime_ms == sum(item.total_reply_ms for item in budget_cases) + + +def test_invalid_budget_measurement_still_fails_before_metric_production() -> None: + run = _run(_setup()) + cases = run.setup.cases + complexity = replace(cases[0].complexity, changed_bytes=cases[0].complexity.changed_bytes + 1) + forged = replace(cases[0], complexity=complexity) + forged = replace(forged, result_digest=derive_admission_case_result_digest(forged)) + forged_cases = (forged, *cases[1:]) + + with pytest.raises(AdmissionMetricsError, match="budget validation.*measurement"): + _derive(run, cases=forged_cases, release=_release_for(run, forged_cases)) def test_runner_case_digest_is_rederived() -> None: diff --git a/tests/test_core_admission_report.py b/tests/test_core_admission_report.py new file mode 100644 index 0000000..0d025a6 --- /dev/null +++ b/tests/test_core_admission_report.py @@ -0,0 +1,670 @@ +"""Evidence-backed final admission reports.""" + +from __future__ import annotations + +from collections.abc import Iterator, Sequence +from dataclasses import replace +from typing import NamedTuple, TypeVar, overload + +import pytest +from test_core_admission_context import _resign_manifest +from test_core_admission_metrics import _MetricRun, _MetricSetup, _release_for, _run, _setup, _with_out_of_scope_case +from test_core_readiness import _chain + +from openultrasast.core import _admission_gates as admission_gates_module +from openultrasast.core import admission_context, canonical, validation +from openultrasast.core import admission_report as admission_report_module +from openultrasast.core.admission_case import ( + AdmissionCaseRelease, + AdmissionCaseResult, + AdmissionCaseRole, + derive_admission_case_result_digest, +) +from openultrasast.core.admission_report import ( + AdmissionDecision, + AdmissionReplay, + AdmissionReportError, + AdmissionReportV2, + GateKind, + GateResult, + admission_replay_value, + admission_report_value, + score_admission, + validate_admission_replay, +) +from openultrasast.core.admission_review import AdmissionThreshold, derive_threshold_set_digest +from openultrasast.core.coverage import ReasonCount +from openultrasast.core.evidence import EvidenceBasis +from openultrasast.core.measurement import Comparator, MetricId, Statistic, evaluate_metric +from openultrasast.core.measurement_factory import create_measurement_record +from openultrasast.core.readiness import ReadinessResult, ReadinessStage + +_T = TypeVar("_T") + + +class _ScoredAdmission(NamedTuple): + replay: AdmissionReplay + setup: _MetricSetup + run: _MetricRun + cases: tuple[AdmissionCaseResult, ...] + readiness: tuple[ReadinessResult, ...] + release: AdmissionCaseRelease + + @property + def report(self) -> AdmissionReportV2: + return self.replay.report + + +class _SinglePassSequence(Sequence[_T]): + def __init__(self, values: tuple[_T, ...]) -> None: + self.values = values + self.traversals = 0 + + def __iter__(self) -> Iterator[_T]: + self.traversals += 1 + if self.traversals > 1: + raise AssertionError("caller-controlled sequence traversed twice") + return iter(self.values) + + def __len__(self) -> int: + return len(self.values) + + @overload + def __getitem__(self, index: int) -> _T: ... + + @overload + def __getitem__(self, index: slice) -> Sequence[_T]: ... + + def __getitem__(self, index: int | slice) -> _T | Sequence[_T]: + return self.values[index] + + +def _without_models(setup: _MetricSetup) -> _MetricSetup: + evidence = setup.context.evidence + model_lineages = {item.lineage_manifest_digest for item in evidence.model_programs} + model_free_evidence = validation.make_evidence_validation_context( + family_catalog=evidence.family_catalog, + fact_packs=evidence.fact_packs, + model_programs=(), + lineages=tuple(item for item in evidence.lineages if item.manifest_digest not in model_lineages), + fact_review_policy=evidence.fact_review_policy, + provenance_trust_policy=evidence.provenance_trust_policy, + custody_policy=evidence.custody_policy, + ) + context = admission_context.make_admission_validation_context( + evidence=model_free_evidence, + admission_review_policy=setup.context.admission_review_policy, + proof_trust_policy=setup.context.proof_trust_policy, + ) + identity = replace( + setup.manifest.identity, + model_program_digests=(), + model_program_set_digest=None, + ) + scope = identity.claimed_scopes[0] + analysis = admission_context.derive_analysis_identity(identity, evidence.family_catalog, scope, None) + readiness = _chain(identity=analysis)[:6] + manifest = _resign_manifest( + replace( + setup.manifest, + identity=identity, + readiness_digests=tuple(sorted(item.result_digest for item in readiness)), + validation_context_digest=context.context_digest, + ), + setup.manifest.review.conflict_disclosure, + ) + return setup._replace(manifest=manifest, context=context, readiness=readiness) + + +def _score( + setup: _MetricSetup, + *, + cases: Sequence[AdmissionCaseResult] | None = None, + readiness: Sequence[ReadinessResult] | None = None, +) -> AdmissionReportV2: + return _score_bundle(setup, cases=cases, readiness=readiness).report + + +def _score_bundle( + setup: _MetricSetup, + *, + cases: Sequence[AdmissionCaseResult] | None = None, + readiness: Sequence[ReadinessResult] | None = None, +) -> _ScoredAdmission: + run = _run(setup) + selected_cases = tuple(setup.cases if cases is None else cases) + selected_readiness = tuple(setup.readiness if readiness is None else readiness) + replay = score_admission( + setup.manifest, + setup.denominator, + selected_readiness, + run.preparation, + run.receipt, + run.release, + selected_cases, + setup.context, + setup.custody, + setup.resolver, + ) + return _ScoredAdmission(replay, setup, run, replay.cases, replay.readiness, run.release) + + +def _validate_authoritatively(replay: AdmissionReplay, scored: _ScoredAdmission) -> None: + validate_admission_replay( + replay, + scored.setup.context, + scored.setup.custody, + scored.setup.resolver, + ) + + +def _rehashed(report: AdmissionReportV2, **changes: object) -> AdmissionReportV2: + draft = object.__new__(AdmissionReportV2) + for field in AdmissionReportV2.__dataclass_fields__: + object.__setattr__(draft, field, changes.get(field, getattr(report, field))) + object.__setattr__(draft, "report_digest", "0" * 64) + digest = canonical.content_digest(admission_report_module._report_value(draft)) + object.__setattr__(draft, "report_digest", digest) + return draft + + +def _rehashed_replay(replay: AdmissionReplay, **changes: object) -> AdmissionReplay: + draft = object.__new__(AdmissionReplay) + for field in AdmissionReplay.__dataclass_fields__: + object.__setattr__(draft, field, changes.get(field, getattr(replay, field))) + object.__setattr__(draft, "replay_digest", "0" * 64) + digest = canonical.content_digest(admission_report_module._replay_value(draft)) + object.__setattr__(draft, "replay_digest", digest) + return draft + + +def _mutated(value: _T, **changes: object) -> _T: + draft = object.__new__(type(value)) + for field in value.__dataclass_fields__: # type: ignore[attr-defined] + object.__setattr__(draft, field, changes.get(field, getattr(value, field))) + return draft + + +def _failed_quality_bundle() -> _ScoredAdmission: + setup = _setup() + vulnerable, fixed, ordinary = setup.cases + vulnerable = replace(vulnerable, role=AdmissionCaseRole.FIXED, result_digest="0" * 64) + vulnerable = replace(vulnerable, result_digest=derive_admission_case_result_digest(vulnerable)) + fixed = replace(fixed, role=AdmissionCaseRole.VULNERABLE, result_digest="0" * 64) + fixed = replace(fixed, result_digest=derive_admission_case_result_digest(fixed)) + cases = (vulnerable, fixed, ordinary) + run = _run(setup) + release = _release_for(run, cases) + replay = score_admission( + setup.manifest, + setup.denominator, + setup.readiness, + run.preparation, + run.receipt, + release, + cases, + setup.context, + setup.custody, + setup.resolver, + ) + return _ScoredAdmission(replay, setup, run, replay.cases, replay.readiness, release) + + +def _failed_quality_report() -> AdmissionReportV2: + return _failed_quality_bundle().report + + +def test_complete_admission_is_reproducible_and_bound_to_exact_evidence() -> None: + setup = _without_models(_setup()) + run = _run(setup) + + first = score_admission( + setup.manifest, + setup.denominator, + setup.readiness, + run.preparation, + run.receipt, + run.release, + setup.cases, + setup.context, + setup.custody, + setup.resolver, + ) + second = _score_bundle(_without_models(_setup())).replay + + assert first == second + assert canonical.canonical_bytes(admission_replay_value(first)) == canonical.canonical_bytes(admission_replay_value(second)) + assert first.replay_digest == canonical.content_digest(admission_replay_value(first)) + report = first.report + assert report.decision is AdmissionDecision.GO + assert report.report_digest == canonical.content_digest(admission_report_value(report)) + assert report.census_digest == setup.manifest.census.census_digest + assert report.validation_context_digest == setup.context.context_digest + assert report.authorization_digest == run.preparation.authorization.authorization_digest + assert report.label_release_receipt_digest == run.receipt.receipt_digest + assert report.case_release_digest == run.release.release_digest + assert report.case_result_digests == tuple(sorted(case.result_digest for case in setup.cases)) + assert tuple(item.basis for item in report.measurement.evidence_basis_counts) == tuple(EvidenceBasis) + assert tuple(item.kind for item in report.instrument_results) == (GateKind.INSTRUMENT,) + assert tuple(item.kind for item in report.budget_results) == (GateKind.BUDGET,) + assert tuple(item.kind for item in report.readiness_results) == (GateKind.READINESS,) + assert tuple(item.kind for item in report.quality_results) == (GateKind.QUALITY,) + assert all( + gate.passed + for group in ( + report.instrument_results, + report.budget_results, + report.readiness_results, + ) + for gate in group + ) + + +def test_score_snapshots_caller_sequences_once_and_records_development_readiness() -> None: + setup = _setup() + cases = _SinglePassSequence(setup.cases) + readiness = _SinglePassSequence(setup.readiness) + + report = _score(setup, cases=cases, readiness=readiness) + + assert (cases.traversals, readiness.traversals) == (1, 1) + assert tuple(item.stage for item in report.measurement.prerequisite_readiness) == tuple(ReadinessStage)[:6] + + +def test_score_canonicalizes_case_permutations_after_one_sequence_traversal() -> None: + setup = _setup() + run = _run(setup) + cases = _SinglePassSequence(tuple(reversed(setup.cases))) + + replay = score_admission( + setup.manifest, + setup.denominator, + setup.readiness, + run.preparation, + run.receipt, + run.release, + cases, + setup.context, + setup.custody, + setup.resolver, + ) + canonical_replay = _score_bundle(_setup()).replay + + assert cases.traversals == 1 + assert replay == canonical_replay + assert replay.replay_digest == canonical_replay.replay_digest + assert tuple(item.case_id_digest for item in replay.cases) == tuple(sorted(item.case_id_digest for item in setup.cases)) + + +def test_contextual_validator_snapshots_caller_sequences_once() -> None: + scored = _score_bundle(_setup()) + cases = _SinglePassSequence(scored.cases) + readiness = _SinglePassSequence(scored.readiness) + + replay = _rehashed_replay(scored.replay, readiness=tuple(readiness), cases=tuple(cases)) + validate_admission_replay( + replay, + scored.setup.context, + scored.setup.custody, + scored.setup.resolver, + ) + + assert (cases.traversals, readiness.traversals) == (1, 1) + + +def test_raw_report_is_not_an_authorization_replay() -> None: + scored = _score_bundle(_setup()) + + with pytest.raises(AdmissionReportError, match="exact AdmissionReplay"): + validate_admission_replay( # type: ignore[arg-type] + scored.report, + scored.setup.context, + scored.setup.custody, + scored.setup.resolver, + ) + + +def test_rehashed_replay_cannot_substitute_readiness_order() -> None: + scored = _score_bundle(_setup()) + forged = _rehashed_replay(scored.replay, readiness=tuple(reversed(scored.readiness))) + + with pytest.raises(AdmissionReportError, match="readiness"): + _validate_authoritatively(forged, scored) + + +def test_replay_digest_mutation_is_refused_before_authoritative_replay() -> None: + replay = _score_bundle(_setup()).replay + + with pytest.raises(AdmissionReportError, match="replay digest"): + replace(replay, replay_digest=canonical.content_digest("forged-replay")) + + +def test_rehashed_replay_cannot_substitute_a_case_payload() -> None: + scored = _score_bundle(_setup()) + original = scored.replay.cases[0] + changed = replace(original, total_reply_ms=original.total_reply_ms + 1, result_digest="0" * 64) + changed = replace(changed, result_digest=derive_admission_case_result_digest(changed)) + cases = tuple(sorted((changed, *scored.replay.cases[1:]), key=lambda item: item.case_id_digest)) + forged = _rehashed_replay(scored.replay, cases=cases) + + with pytest.raises(AdmissionReportError, match="case|release|binding"): + _validate_authoritatively(forged, scored) + + +def test_failed_registered_quality_threshold_is_a_valid_no_go_report() -> None: + report = _failed_quality_report() + + recall = next(item for item in report.measurement.metrics if item.metric_id is MetricId.VULNERABLE_RECALL) + assert recall.numerator == 0 + assert report.decision is AdmissionDecision.NO_GO + assert report.quality_results[0].passed is False + assert "threshold_failed:vulnerable_recall:count" in report.quality_results[0].reasons + + +def test_validated_budget_overrun_produces_failed_gate_and_no_go() -> None: + setup = _setup() + run = _run(setup) + original = setup.cases[0] + changed = replace(original, total_reply_ms=original.budget.budget_ms + 1, result_digest="0" * 64) + changed = replace(changed, result_digest=derive_admission_case_result_digest(changed)) + cases = tuple(sorted((changed, *setup.cases[1:]), key=lambda item: item.case_id_digest)) + release = _release_for(run, cases) + + replay = score_admission( + setup.manifest, + setup.denominator, + setup.readiness, + run.preparation, + run.receipt, + release, + cases, + setup.context, + setup.custody, + setup.resolver, + ) + + report = replay.report + assert report.decision is AdmissionDecision.NO_GO + assert not report.budget_results[0].passed + assert report.budget_results[0].reasons == (f"budget_exceeded:{changed.result_digest}",) + assert validate_admission_replay(replay, setup.context, setup.custody, setup.resolver) == report + + +def test_no_go_cannot_be_replaced_by_go_or_lose_a_mandatory_gate() -> None: + report = _score(_setup()) + with pytest.raises(AdmissionReportError, match="quality gate"): + replace(report, quality_results=(), report_digest="0" * 64) + with pytest.raises(AdmissionReportError, match="report digest"): + replace(report, report_digest=canonical.content_digest("forged")) + + +def test_configured_model_with_zero_calls_is_canonical_null_and_no_go() -> None: + scored = _score_bundle(_setup()) + report = scored.report + model = next(item for item in report.measurement.metrics if item.metric_id is MetricId.MODEL_SUCCESS_RATE) + + assert (model.numerator, model.denominator, model.value) == (0, 0, None) + assert report.decision is AdmissionDecision.NO_GO + assert "mandatory_model_success_threshold_missing" in report.quality_results[0].reasons + assert "mandatory_model_success_has_no_attempted_calls" in report.quality_results[0].reasons + _validate_authoritatively(scored.replay, scored) + + +def test_rehashed_failed_quality_gate_cannot_be_promoted_to_go() -> None: + scored = _failed_quality_bundle() + report = scored.report + forged_gate = replace(report.quality_results[0], passed=True, reasons=()) + forged = _rehashed(report, quality_results=(forged_gate,), decision=AdmissionDecision.GO) + + with pytest.raises(AdmissionReportError, match="quality gate"): + _validate_authoritatively(_rehashed_replay(scored.replay, report=forged), scored) + + +@pytest.mark.parametrize("category", ("instrument", "budget", "readiness")) +def test_rehashed_gate_evidence_substitution_is_refused(category: str) -> None: + scored = _score_bundle(_setup()) + report = scored.report + field = f"{category}_results" + gate = getattr(report, field)[0] + forged_gate = replace(gate, evidence_digests=(canonical.content_digest(f"forged:{category}"),)) + forged = _rehashed(report, **{field: (forged_gate,)}) + + with pytest.raises(AdmissionReportError, match=rf"{category} gate"): + _validate_authoritatively(_rehashed_replay(scored.replay, report=forged), scored) + + +def test_rehashed_measurement_readiness_divergence_is_refused() -> None: + scored = _score_bundle(_setup()) + report = scored.report + measurement = report.measurement + changed_reference = replace( + measurement.prerequisite_readiness[0], + result_digest=canonical.content_digest("different-readiness-result"), + ) + changed_measurement = create_measurement_record( + identity=measurement.identity, + family_decision_vocabulary=measurement.family_decision_vocabulary, + family_decision_counts=measurement.family_decision_counts, + neutral_outcome_counts=measurement.neutral_outcome_counts, + evidence_basis_counts=measurement.evidence_basis_counts, + metrics=measurement.metrics, + stage_funnel=measurement.stage_funnel, + instrument_failures=measurement.instrument_failures, + model_outcome_counts=measurement.model_outcome_counts, + model_failures=measurement.model_failures, + runtime_ms=measurement.runtime_ms, + prerequisite_readiness=(changed_reference, *measurement.prerequisite_readiness[1:]), + ) + evaluations = tuple(_mutated(item, measurement_record_digest=changed_measurement.record_digest) for item in report.quality_evaluations) + evaluations = tuple( + _mutated( + item, + evaluation_digest=canonical.content_digest(admission_gates_module.threshold_evaluation_value(item)), + ) + for item in evaluations + ) + quality = admission_gates_module.quality_gate(report.identity, changed_measurement, evaluations) + forged = _rehashed( + report, + measurement=changed_measurement, + quality_evaluations=evaluations, + quality_results=(quality,), + ) + + with pytest.raises(AdmissionReportError, match="readiness gate"): + _validate_authoritatively(_rehashed_replay(scored.replay, report=forged), scored) + + +def test_rehashed_threshold_evaluation_mutations_are_refused() -> None: + scored = _score_bundle(_setup()) + report = scored.report + evaluations = report.quality_evaluations + changed = _mutated(evaluations[0], target=evaluations[0].target + 0.25) + changed = _mutated( + changed, + evaluation_digest=canonical.content_digest(admission_gates_module.threshold_evaluation_value(changed)), + ) + + for forged_evaluations in ((), (evaluations[0], evaluations[0]), (changed,)): + forged = _rehashed(report, quality_evaluations=forged_evaluations) + with pytest.raises(AdmissionReportError, match="threshold|quality"): + _validate_authoritatively(_rehashed_replay(scored.replay, report=forged), scored) + + +def test_rehashed_typed_case_budget_substitution_is_refused() -> None: + scored = _score_bundle(_setup()) + report = scored.report + original = report.budget_evaluations[0] + changed = _mutated(original, budget_derivation_digest=canonical.content_digest("forged-budget")) + changed = _mutated( + changed, + evaluation_digest=canonical.content_digest(admission_gates_module.budget_evaluation_value(changed)), + ) + forged = _rehashed(report, budget_evaluations=(changed, *report.budget_evaluations[1:])) + + with pytest.raises(AdmissionReportError, match="budget evaluation"): + _validate_authoritatively(_rehashed_replay(scored.replay, report=forged), scored) + + +def test_contextual_validation_refuses_coordinated_threshold_rehash() -> None: + scored = _failed_quality_bundle() + report = scored.report + threshold = AdmissionThreshold(MetricId.VULNERABLE_RECALL, Statistic.COUNT, Comparator.AT_LEAST, 0.0) + threshold_digest = derive_threshold_set_digest((threshold,)) + identity = replace( + report.identity, + model_program_digests=(), + model_program_set_digest=None, + threshold_set_digest=threshold_digest, + ) + measurement = create_measurement_record( + identity=identity, + family_decision_vocabulary=report.measurement.family_decision_vocabulary, + family_decision_counts=report.measurement.family_decision_counts, + neutral_outcome_counts=report.measurement.neutral_outcome_counts, + evidence_basis_counts=report.measurement.evidence_basis_counts, + metrics=report.measurement.metrics, + stage_funnel=report.measurement.stage_funnel, + instrument_failures=report.measurement.instrument_failures, + model_outcome_counts=report.measurement.model_outcome_counts, + model_failures=report.measurement.model_failures, + runtime_ms=report.measurement.runtime_ms, + prerequisite_readiness=report.measurement.prerequisite_readiness, + ) + evaluations = admission_gates_module.make_threshold_evaluations((threshold,), measurement) + quality = admission_gates_module.quality_gate(identity, measurement, evaluations) + forged = _rehashed( + report, + identity=identity, + measurement=measurement, + quality_evaluations=evaluations, + quality_results=(quality,), + decision=AdmissionDecision.GO, + ) + + admission_report_module._validate_report_structure(forged) + assert forged.authorization_digest == report.authorization_digest + with pytest.raises(AdmissionReportError, match="authoritative admission inputs"): + _validate_authoritatively(_rehashed_replay(scored.replay, report=forged), scored) + + +def test_contextual_validation_refuses_coordinated_readiness_rehash() -> None: + scored = _score_bundle(_setup()) + report = scored.report + references = tuple( + replace(item, result_digest=canonical.content_digest(f"forged-readiness:{item.stage.value}")) + for item in report.measurement.prerequisite_readiness + ) + measurement = create_measurement_record( + identity=report.identity, + family_decision_vocabulary=report.measurement.family_decision_vocabulary, + family_decision_counts=report.measurement.family_decision_counts, + neutral_outcome_counts=report.measurement.neutral_outcome_counts, + evidence_basis_counts=report.measurement.evidence_basis_counts, + metrics=report.measurement.metrics, + stage_funnel=report.measurement.stage_funnel, + instrument_failures=report.measurement.instrument_failures, + model_outcome_counts=report.measurement.model_outcome_counts, + model_failures=report.measurement.model_failures, + runtime_ms=report.measurement.runtime_ms, + prerequisite_readiness=references, + ) + thresholds = tuple( + AdmissionThreshold(item.metric_id, item.statistic, item.comparator, item.target) for item in report.quality_evaluations + ) + evaluations = admission_gates_module.make_threshold_evaluations(thresholds, measurement) + quality = admission_gates_module.quality_gate(report.identity, measurement, evaluations) + readiness = admission_gates_module.readiness_gate(references) + forged = _rehashed( + report, + measurement=measurement, + quality_evaluations=evaluations, + quality_results=(quality,), + readiness_results=(readiness,), + ) + + admission_report_module._validate_report_structure(forged) + with pytest.raises(AdmissionReportError, match="authoritative admission inputs"): + _validate_authoritatively(_rehashed_replay(scored.replay, report=forged), scored) + + +def test_out_of_scope_only_model_capacity_does_not_enter_claimed_scope_metric() -> None: + setup = _with_out_of_scope_case(_setup()) + report = _score(setup) + claimed = set(setup.manifest.identity.claimed_scopes) + rows = tuple(replace(row, model_call_capacity=1) if row.scope not in claimed else row for row in setup.manifest.census.rows) + changed_manifest = replace(setup.manifest, census=replace(setup.manifest.census, rows=rows)) + + gate = admission_report_module._quality_gate(changed_manifest, report.measurement) + model = next(item for item in report.measurement.metrics if item.metric_id is MetricId.MODEL_SUCCESS_RATE) + + assert (model.numerator, model.denominator) == (0, 0) + assert not gate.passed + assert report.decision is AdmissionDecision.NO_GO + + +def _model_quality_gate( + *, + numerator: int, + denominator: int = 1, + thresholds: tuple[AdmissionThreshold, ...], +) -> GateResult: + setup = _setup() + report = _score(setup) + identity = replace(report.identity, threshold_set_digest=derive_threshold_set_digest(thresholds)) + rule = next(item for item in setup.denominator.rules if item.metric_id is MetricId.MODEL_SUCCESS_RATE) + model_metric = evaluate_metric(rule, numerator=numerator, denominator=denominator) + metrics = tuple(model_metric if item.metric_id is MetricId.MODEL_SUCCESS_RATE else item for item in report.measurement.metrics) + model_outcomes = tuple( + ReasonCount(name, count) for name, count in (("failure", denominator - numerator), ("success", numerator)) if count + ) + measurement = create_measurement_record( + identity=identity, + family_decision_vocabulary=report.measurement.family_decision_vocabulary, + family_decision_counts=report.measurement.family_decision_counts, + neutral_outcome_counts=report.measurement.neutral_outcome_counts, + evidence_basis_counts=report.measurement.evidence_basis_counts, + metrics=metrics, + stage_funnel=report.measurement.stage_funnel, + instrument_failures=report.measurement.instrument_failures, + model_outcome_counts=model_outcomes, + model_failures=() if numerator == denominator else ("provider_timeout",), + runtime_ms=report.measurement.runtime_ms, + prerequisite_readiness=report.measurement.prerequisite_readiness, + ) + evaluations = admission_gates_module.make_threshold_evaluations(thresholds, measurement) + return admission_gates_module.quality_gate(identity, measurement, evaluations) + + +def test_applicable_passing_model_requires_and_passes_registered_threshold() -> None: + base = _setup().manifest.thresholds[0] + model = AdmissionThreshold(MetricId.MODEL_SUCCESS_RATE, Statistic.COUNT, Comparator.AT_LEAST, 1.0) + thresholds = tuple(sorted((base, model), key=lambda item: (item.metric_id.value, item.statistic.value))) + + gate = _model_quality_gate(numerator=1, thresholds=thresholds) + + assert gate.passed + + +def test_applicable_failed_model_or_missing_model_threshold_cannot_go() -> None: + base = _setup().manifest.thresholds[0] + missing = _model_quality_gate(numerator=0, thresholds=(base,)) + model = AdmissionThreshold(MetricId.MODEL_SUCCESS_RATE, Statistic.COUNT, Comparator.AT_LEAST, 1.0) + thresholds = tuple(sorted((base, model), key=lambda item: (item.metric_id.value, item.statistic.value))) + failed = _model_quality_gate(numerator=0, thresholds=thresholds) + + assert not missing.passed + assert "mandatory_model_success_threshold_missing" in missing.reasons + assert not failed.passed + assert "mandatory_model_call_failed" in failed.reasons + + +def test_any_failed_declared_model_call_forces_no_go_even_when_threshold_passes() -> None: + base = _setup().manifest.thresholds[0] + model = AdmissionThreshold(MetricId.MODEL_SUCCESS_RATE, Statistic.COUNT, Comparator.AT_LEAST, 1.0) + thresholds = tuple(sorted((base, model), key=lambda item: (item.metric_id.value, item.statistic.value))) + + gate = _model_quality_gate(numerator=1, denominator=2, thresholds=thresholds) + + assert not gate.passed + assert "mandatory_model_call_failed" in gate.reasons diff --git a/tests/test_core_measurement_record.py b/tests/test_core_measurement_record.py index 000b2b2..b150a04 100644 --- a/tests/test_core_measurement_record.py +++ b/tests/test_core_measurement_record.py @@ -54,7 +54,7 @@ def _funnel(*, inputs_read: int = 1, findings: int = 1) -> CoverageFunnel: def _readiness(*, result: StageResult = StageResult.PASSED) -> tuple[ReadinessReference, ...]: return tuple( - ReadinessReference(stage, result, _digest(f"readiness:{stage.value}:{result.value}")) for stage in tuple(ReadinessStage)[:5] + ReadinessReference(stage, result, _digest(f"readiness:{stage.value}:{result.value}")) for stage in tuple(ReadinessStage)[:6] ) @@ -163,7 +163,7 @@ def test_decision_vocabulary_counts_and_funnel_totals_must_agree() -> None: def test_later_measurement_carries_canonical_prerequisite_results_and_digests() -> None: record = _record() - assert tuple(item.stage for item in record.prerequisite_readiness) == tuple(ReadinessStage)[:5] + assert tuple(item.stage for item in record.prerequisite_readiness) == tuple(ReadinessStage)[:6] with pytest.raises(MeasurementRecordError, match="prerequisite readiness"): replace(record, prerequisite_readiness=record.prerequisite_readiness[:-1], record_digest="") with pytest.raises(MeasurementRecordError, match="readiness digest"):