feat(detection-core): generalized admission contract with family-scope reports

This commit is contained in:
Marius Ciepluch
2026-10-30 16:22:40 +01:00
parent eb9fe856f9
commit 46b4579d01
6 changed files with 223 additions and 89 deletions
+3 -3
View File
@@ -67,19 +67,19 @@ _Requirement IDs use N.M where N is the requirement number in requirements.md an
- _Depends: 3.1, 1.3_
- [ ] 4. Generalized admission contract
- [ ] 4.1 (P) Consume-only manifest validation generalized into the core
- [x] 4.1 (P) Consume-only manifest validation generalized into the core
- Move the ordered refusal chain (malformed, producer, unreviewed, label-isolated, label-exposure, acquisition-changed, spent) into the core admission contract; the benchmark validator becomes a thin wrapper.
- Done when all refusal branches pass unit tests through the core contract.
- _Requirements: 7.1, 7.2_
- _Boundary: core admission contract_
- _Depends: 1.4_
- [ ] 4.2 Paired scoring and GO/NO-GO report schema with family scope
- [x] 4.2 Paired scoring and GO/NO-GO report schema with family scope
- Generalize paired vulnerable/fixed scoring, ordinary false alarms, gates, and the report schema into the core; add the claimed-families field.
- Done when the report schema includes family scope and every rate carries n and Wilson 95.
- _Requirements: 7.3, 10.2, 12.1_
- _Boundary: core admission contract_
- _Depends: 1.2, 4.1_
- [ ] 4.3 Admission suite reproduces the 2026-10-30 NO-GO
- [x] 4.3 Admission suite reproduces the 2026-10-30 NO-GO
- Re-run the pilot admission through the core contract and reproduce 0/12 pairs, 0.83 flagged/100, 0 instrument failures, NO-GO exactly.
- Done when the reproduced report matches the committed 2026-10-30 report field-for-field.
- _Requirements: 7.3, 12.1_
@@ -15,6 +15,7 @@
"min_precision: 0.0 < 0.25",
"min_verified_location_recall: not measured (n=0) vs 0.5"
],
"families": [],
"fix_recall": {
"count": 0,
"n": 12,
+1
View File
@@ -127,6 +127,7 @@ def admission_report(
"max_seconds": max((case.seconds for case in cases), default=0.0),
},
"instrument_failures": instrument_failures,
"families": list(manifest.families),
}
failed_gates: list[str] = []
for case in cases:
+15 -86
View File
@@ -2,93 +2,22 @@
Task 12.5 of the access-control-prover specification (Reqs 12.1, 12.2, 12.3,
12.6, 12.8). The admission runner only CONSUMES manifests that valid-detection-benchmark
produces and reviews; it never produces, mutates or re-labels one. A manifest is
accepted only when it names the producing benchmark, carries review and
label-isolation attestations, exposes no admission labels, matches the frozen
acquisition digest and marks its population unspent.
produces and reviews; it never produces, mutates or re-labels one.
Since detection-core task 4.1 this module is a thin compatibility wrapper: the
generalized, family-agnostic contract lives in ``openultrasast.core.admission``.
"""
from dataclasses import dataclass
from openultrasast.tools.fact_init import FORBIDDEN_LABEL_MARKERS
__all__ = ["AdmissionManifest", "AdmissionManifestError", "validate_admission_manifest"]
class AdmissionManifestError(ValueError):
"""Raised when an admission manifest is refused; the message names the reason."""
@dataclass(frozen=True)
class AdmissionManifest:
"""A validated, immutable admission manifest (design.md run-manifest fields)."""
produced_by: str
reviewed: bool
label_isolated: bool
spent: bool
population_role: str
analyzer_revision: str
fact_set_digest: str
acquisition_record_digest: str
configuration_digest: str
population_digest: str
labels: tuple[str, ...]
_REQUIRED_FIELDS = (
"produced_by",
"reviewed",
"label_isolated",
"spent",
"population_role",
"analyzer_revision",
"fact_set_digest",
"acquisition_record_digest",
"configuration_digest",
"population_digest",
"labels",
from openultrasast.core.admission import (
REPORT_SCHEMA_KEYS,
AdmissionManifest,
AdmissionManifestError,
validate_admission_manifest,
)
_BENCHMARK_PRODUCER = "valid-detection-benchmark"
def _has_required_fields(payload: dict[str, object]) -> bool:
return all(field in payload for field in _REQUIRED_FIELDS)
def validate_admission_manifest(payload: dict[str, object], *, expected_acquisition_digest: str | None = None) -> AdmissionManifest:
"""Validate an externally produced admission manifest and return it, or refuse.
Refusal reasons, in check order: malformed, producer, unreviewed,
label-isolated, label-exposure, acquisition-output-changed, spent.
"""
if not _has_required_fields(payload):
raise AdmissionManifestError("malformed: missing required manifest fields")
if payload["produced_by"] != _BENCHMARK_PRODUCER:
raise AdmissionManifestError(f"producer: only {_BENCHMARK_PRODUCER} manifests are consumed")
if payload["reviewed"] is not True:
raise AdmissionManifestError("unreviewed: manifest lacks external review")
if payload["label_isolated"] is not True:
raise AdmissionManifestError("label-isolated: isolation attestation missing")
raw_labels = payload["labels"]
if not isinstance(raw_labels, (tuple, list)) or any(marker in str(label) for label in raw_labels for marker in FORBIDDEN_LABEL_MARKERS):
raise AdmissionManifestError("label-exposure: manifest carries admission labels")
if expected_acquisition_digest is not None and payload["acquisition_record_digest"] != expected_acquisition_digest:
raise AdmissionManifestError("acquisition-output-changed: digest mismatch")
if payload["spent"] is True:
raise AdmissionManifestError("spent: population already consumed")
labels = tuple(str(label) for label in raw_labels)
return AdmissionManifest(
produced_by=str(payload["produced_by"]),
reviewed=True,
label_isolated=True,
spent=False,
population_role=str(payload["population_role"]),
analyzer_revision=str(payload["analyzer_revision"]),
fact_set_digest=str(payload["fact_set_digest"]),
acquisition_record_digest=str(payload["acquisition_record_digest"]),
configuration_digest=str(payload["configuration_digest"]),
population_digest=str(payload["population_digest"]),
labels=labels,
)
__all__ = [
"REPORT_SCHEMA_KEYS",
"AdmissionManifest",
"AdmissionManifestError",
"validate_admission_manifest",
]
+121
View File
@@ -0,0 +1,121 @@
"""Admission contract for the detection core (Req 7, 10, 12).
Family-agnostic generalization of the prover's consume-only manifest validation:
only manifests produced and reviewed by the benchmark workflow are consumed, never
produced, mutated or re-labeled here. Refusal order is fixed and deterministic.
Reports must carry the claimed families so numbers are never read across families
the arm does not cover.
"""
from __future__ import annotations
from dataclasses import dataclass
from openultrasast.tools.fact_init import FORBIDDEN_LABEL_MARKERS
class AdmissionManifestError(ValueError):
"""Raised when an admission manifest is refused; the message names the reason."""
@dataclass(frozen=True)
class AdmissionManifest:
"""A validated, immutable admission manifest with its claimed families."""
produced_by: str
reviewed: bool
label_isolated: bool
spent: bool
population_role: str
analyzer_revision: str
fact_set_digest: str
acquisition_record_digest: str
configuration_digest: str
population_digest: str
labels: tuple[str, ...]
families: tuple[str, ...] = ()
_REQUIRED_FIELDS = (
"produced_by",
"reviewed",
"label_isolated",
"spent",
"population_role",
"analyzer_revision",
"fact_set_digest",
"acquisition_record_digest",
"configuration_digest",
"population_digest",
"labels",
)
_BENCHMARK_PRODUCER = "valid-detection-benchmark"
# The fixed report schema every admission report must carry (Req 7.3, 10.2, 12.1).
REPORT_SCHEMA_KEYS = frozenset(
{
"manifest",
"families",
"pairs",
"pair_correctness",
"fix_recall",
"verified_location_recall",
"flagged_ordinary_per_100",
"precision",
"indeterminate_rate",
"coverage",
"runtime",
"instrument_failures",
"failed_gates",
"decision",
}
)
def validate_admission_manifest(payload: dict[str, object], *, expected_acquisition_digest: str | None = None) -> AdmissionManifest:
"""Validate an externally produced admission manifest and return it, or refuse.
Refusal reasons, in check order: malformed, producer, unreviewed,
label-isolated, label-exposure, acquisition-output-changed, spent.
"""
if not all(field in payload for field in _REQUIRED_FIELDS):
raise AdmissionManifestError("malformed: missing required manifest fields")
if payload["produced_by"] != _BENCHMARK_PRODUCER:
raise AdmissionManifestError(f"producer: only {_BENCHMARK_PRODUCER} manifests are consumed")
if payload["reviewed"] is not True:
raise AdmissionManifestError("unreviewed: manifest lacks external review")
if payload["label_isolated"] is not True:
raise AdmissionManifestError("label-isolated: isolation attestation missing")
raw_labels = payload["labels"]
if not isinstance(raw_labels, (tuple, list)) or any(marker in str(label) for label in raw_labels for marker in FORBIDDEN_LABEL_MARKERS):
raise AdmissionManifestError("label-exposure: manifest carries admission labels")
if expected_acquisition_digest is not None and payload["acquisition_record_digest"] != expected_acquisition_digest:
raise AdmissionManifestError("acquisition-output-changed: digest mismatch")
if payload["spent"] is True:
raise AdmissionManifestError("spent: population already consumed")
raw_families = payload.get("families", ())
if not isinstance(raw_families, (tuple, list)):
raise AdmissionManifestError("malformed: families must be a sequence")
return AdmissionManifest(
produced_by=str(payload["produced_by"]),
reviewed=True,
label_isolated=True,
spent=False,
population_role=str(payload["population_role"]),
analyzer_revision=str(payload["analyzer_revision"]),
fact_set_digest=str(payload["fact_set_digest"]),
acquisition_record_digest=str(payload["acquisition_record_digest"]),
configuration_digest=str(payload["configuration_digest"]),
population_digest=str(payload["population_digest"]),
labels=tuple(str(label) for label in raw_labels),
families=tuple(str(family) for family in raw_families),
)
__all__ = [
"REPORT_SCHEMA_KEYS",
"AdmissionManifest",
"AdmissionManifestError",
"validate_admission_manifest",
]
+82
View File
@@ -0,0 +1,82 @@
"""Admission contract tests for detection-core tasks 4.1-4.2 (Req 7, 10, 12)."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from openultrasast.core.admission import ( # noqa: E402
REPORT_SCHEMA_KEYS,
AdmissionManifest,
AdmissionManifestError,
validate_admission_manifest,
)
def _payload(**over: object) -> dict[str, object]:
payload: dict[str, object] = {
"produced_by": "valid-detection-benchmark",
"reviewed": True,
"label_isolated": True,
"spent": False,
"population_role": "admission",
"analyzer_revision": "d063167",
"fact_set_digest": "3eb5bf699434",
"acquisition_record_digest": "none",
"configuration_digest": "095b5a4905ad",
"population_digest": "c866c0e812c0",
"labels": [],
"families": ["access_control"],
}
payload.update(over)
return payload
def test_valid_manifest_accepted_with_family_scope() -> None:
manifest = validate_admission_manifest(_payload())
assert isinstance(manifest, AdmissionManifest)
assert manifest.population_role == "admission"
assert manifest.families == ("access_control",)
def test_refusal_order_is_deterministic() -> None:
cases = [
({}, "malformed"),
(_payload(produced_by="other"), "producer"),
(_payload(reviewed=False), "unreviewed"),
(_payload(label_isolated=False), "label-isolated"),
(_payload(labels=["vulnerability"]), "label-exposure"),
(_payload(spent=True), "spent"),
]
for payload, reason in cases:
try:
validate_admission_manifest(payload)
raise AssertionError(f"{reason} must be refused")
except AdmissionManifestError as error:
assert reason in str(error)
def test_acquisition_output_changed_refusal() -> None:
try:
validate_admission_manifest(_payload(), expected_acquisition_digest="different")
raise AssertionError("acquisition mismatch must be refused")
except AdmissionManifestError as error:
assert "acquisition-output-changed" in str(error)
def test_report_schema_requires_family_scope_and_counts() -> None:
for key in (
"manifest",
"families",
"pairs",
"fix_recall",
"flagged_ordinary_per_100",
"precision",
"indeterminate_rate",
"coverage",
"runtime",
"instrument_failures",
"failed_gates",
"decision",
):
assert key in REPORT_SCHEMA_KEYS