feat(detection-core): replay admission evidence

This commit is contained in:
Marius Ciepluch
2026-11-01 11:34:23 +01:00
parent 6a1bc2023a
commit e351aca041
9 changed files with 1595 additions and 11 deletions
+1 -1
View File
@@ -280,7 +280,7 @@
- Done when forged decisions, substituted contracts, invalid fact review, contaminated lineage, or case-count drift fail before metric production. - 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_ - _Requirements: 12.5, 12.6, 15.1, 15.2_
- _Depends: 8.7_ - _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. - 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. - 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. - 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.
+341
View File
@@ -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
+136
View File
@@ -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",
]
+4 -4
View File
@@ -24,7 +24,7 @@ from openultrasast.core.admission_case import (
from openultrasast.core.admission_context import AdmissionValidationContext from openultrasast.core.admission_context import AdmissionValidationContext
from openultrasast.core.admission_review import AdmissionManifestV2, ScopeCensus, validate_reviewed_admission_payloads from openultrasast.core.admission_review import AdmissionManifestV2, ScopeCensus, validate_reviewed_admission_payloads
from openultrasast.core.admission_spend import AdmissionPreparation 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.coverage import CoverageFunnel, ReasonCount, validate_instrument
from openultrasast.core.custody import PopulationCustodyPort, PopulationUseReceipt from openultrasast.core.custody import PopulationCustodyPort, PopulationUseReceipt
from openultrasast.core.evidence import EvidenceBasis, EvidenceRecord, NeutralOutcome, derive_packet_digest 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 raise AdmissionMetricsError(f"instrument validation failed: {exc}") from exc
try: try:
check_budget(case.complexity, case.total_reply_ms, case.budget, resolver) check_budget(case.complexity, case.total_reply_ms, case.budget, resolver)
except BudgetGateFailure:
pass
except (TypeError, ValueError) as exc: except (TypeError, ValueError) as exc:
raise AdmissionMetricsError(f"budget validation failed: {exc}") from exc raise AdmissionMetricsError(f"budget validation failed: {exc}") from exc
try: try:
@@ -206,9 +208,7 @@ def _validate_population(cases: tuple[_ValidatedCase, ...], rows: tuple[ScopeCen
for row in rows: for row in rows:
selected = tuple(item for item in cases if item.binding.scope == row.scope) selected = tuple(item for item in cases if item.binding.scope == row.scope)
roles = Counter(item.binding.role for item in selected) roles = Counter(item.binding.role for item in selected)
model_calls = sum( model_calls = sum(value.model_result is not None for item in selected for value in item.supplied.evidence_inputs)
value.model_result is not None for item in selected for value in item.supplied.evidence_inputs
)
actual = ( actual = (
len(selected), len(selected),
roles[AdmissionCaseRole.VULNERABLE], roles[AdmissionCaseRole.VULNERABLE],
+424
View File
@@ -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",
]
+1 -1
View File
@@ -20,7 +20,7 @@ from openultrasast.core.readiness import ReadinessEvidenceUse, ReadinessStage, S
MEASUREMENT_RECORD_SCHEMA = "openultrasast.detection-core.measurement-record.v2" MEASUREMENT_RECORD_SCHEMA = "openultrasast.detection-core.measurement-record.v2"
REPOSITORY_HOLDOUT_REPORT_SCHEMA = "openultrasast.detection-core.repository-holdout-report.v2" REPOSITORY_HOLDOUT_REPORT_SCHEMA = "openultrasast.detection-core.repository-holdout-report.v2"
_DIGEST = re.compile(r"[0-9a-f]{64}") _DIGEST = re.compile(r"[0-9a-f]{64}")
_PREREQUISITE_STAGES = tuple(ReadinessStage)[:5] _PREREQUISITE_STAGES = tuple(ReadinessStage)[:6]
class MeasurementRecordError(ValueError): class MeasurementRecordError(ValueError):
+16 -3
View File
@@ -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)) _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()) run = _run(_setup())
cases = run.setup.cases cases = run.setup.cases
forged_input = replace(cases[0].evidence_inputs[0], input_digest="0" * 64) 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(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)) over_budget = replace(over_budget, result_digest=derive_admission_case_result_digest(over_budget))
budget_cases = (over_budget, *cases[1:]) budget_cases = (over_budget, *cases[1:])
with pytest.raises(AdmissionMetricsError, match="budget"): result = _derive(run, cases=budget_cases, release=_release_for(run, budget_cases))
_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: def test_runner_case_digest_is_rederived() -> None:
+670
View File
@@ -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
+2 -2
View File
@@ -54,7 +54,7 @@ def _funnel(*, inputs_read: int = 1, findings: int = 1) -> CoverageFunnel:
def _readiness(*, result: StageResult = StageResult.PASSED) -> tuple[ReadinessReference, ...]: def _readiness(*, result: StageResult = StageResult.PASSED) -> tuple[ReadinessReference, ...]:
return tuple( 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: def test_later_measurement_carries_canonical_prerequisite_results_and_digests() -> None:
record = _record() 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"): with pytest.raises(MeasurementRecordError, match="prerequisite readiness"):
replace(record, prerequisite_readiness=record.prerequisite_readiness[:-1], record_digest="") replace(record, prerequisite_readiness=record.prerequisite_readiness[:-1], record_digest="")
with pytest.raises(MeasurementRecordError, match="readiness digest"): with pytest.raises(MeasurementRecordError, match="readiness digest"):