diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 5cbd7c6..b0d82cb 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -416,7 +416,7 @@ - Require current prerequisite readiness identities at every later stage. - Done when both authored repositories complete the path without target-specific core fields or a real admission population. - _Requirements: 3.5, 8.1, 8.2, 8.4, 8.6, 12.1, 14.1, 15.5_ -- [ ] 11.2 Inject one failure at every semantic and governance stage +- [x] 11.2 Inject one failure at every semantic and governance stage - Exercise unread/zero input, missing recognition, incomplete transition, missing constraint, pair mismatch, failed development, contaminated admission, and refused integration. - Verify each run stops at the first failing stage with a named correction lever and no later metric or clean result. - Mutate a semantic identity and verify transitive staleness through all dependent readiness and measurement records. diff --git a/tests/detection_core_failure_inputs.py b/tests/detection_core_failure_inputs.py new file mode 100644 index 0000000..b919154 --- /dev/null +++ b/tests/detection_core_failure_inputs.py @@ -0,0 +1,208 @@ +"""Mutated inputs that production detection-core validators must refuse.""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path + +from detection_core_end_to_end import _fixtures, _PreparedFamily +from test_core_admission_metrics import _setup + +from benchmarks.valid import readiness_contract +from openultrasast.core import canonical, evidence, family, semantics +from openultrasast.core._admission_release import _validate_release_shape +from openultrasast.core.admission_case import ( + ADMISSION_CASE_RELEASE_SCHEMA, + AdmissionCaseRelease, + AdmissionCaseRole, + ReleasedCaseBinding, + derive_released_case_binding_digest, +) +from openultrasast.core.coverage import ( + CoverageFunnel, + CoverageStage, + RelationStatusCount, + StageDiagnosis, + StageExpectation, + diagnose_first_zero, +) +from openultrasast.core.measurement import ScopeKey +from openultrasast.core.readiness import AnalysisIdentity, ReadinessResult, ReadinessStage +from openultrasast.core.semantics import RelationStatus + + +def _coverage_diagnosis(stage: ReadinessStage, record: evidence.EvidenceRecord) -> StageDiagnosis | None: + if stage not in (ReadinessStage.RECOGNITION, ReadinessStage.CONSTRAINTS): + return None + transitions = record.packet.transitions if stage is ReadinessStage.CONSTRAINTS else () + statuses = tuple(RelationStatusCount(status, sum(item.status is status for item in transitions)) for status in RelationStatus) + counts = (1, 1, 1, 0, 0, 0, 0, 0, 0) if stage is ReadinessStage.RECOGNITION else (1, 1, 1, 1, len(transitions), 1, 0, 0, 0) + funnel = CoverageFunnel(*counts, statuses, ()) + expected = tuple(StageExpectation(item, item is not CoverageStage.INPUTS, True) for item in CoverageStage) + return diagnose_first_zero(funnel, expected) + + +def _record_with_packet(record: evidence.EvidenceRecord, packet: evidence.EvidencePacket) -> evidence.EvidenceRecord: + decision = replace(record.decision, packet_digest=evidence.derive_packet_digest(packet)) + return evidence.EvidenceRecord(packet, decision, "") + + +def _recognition_failure(prepared: _PreparedFamily) -> tuple[str, StageDiagnosis, str]: + record = prepared.inputs[0].record + packet = replace(record.packet, security_object_id=canonical.content_digest("unrecognized-object")) + mutated = _record_with_packet(record, packet) + try: + family.validate_record(prepared.contract, mutated) + except ValueError as exc: + diagnosis = _coverage_diagnosis(ReadinessStage.RECOGNITION, mutated) + assert diagnosis is not None + return str(exc), diagnosis, "openultrasast.core.family.validate_record" + raise AssertionError("missing recognition was accepted") + + +def _transition_failure(prepared: _PreparedFamily) -> tuple[str, None, str]: + record = prepared.inputs[0].record + relation = record.packet.transitions[0] + incomplete = semantics.make_relation( + kind=relation.kind, + role=relation.role, + source=relation.source, + target=relation.target, + location=relation.location, + status=RelationStatus.INCOMPLETE, + conditions=relation.conditions, + reasons=("missing framework call relation",), + provenance=relation.provenance, + ) + packet = replace( + record.packet, + transitions=(incomplete, *record.packet.transitions[1:]), + completeness_reasons=("missing framework call relation",), + ) + try: + _record_with_packet(record, packet) + except ValueError as exc: + return str(exc), None, "openultrasast.core.evidence.EvidenceRecord" + raise AssertionError("incomplete transition was accepted") + + +def _constraint_failure(prepared: _PreparedFamily) -> tuple[str, StageDiagnosis, str]: + record = prepared.inputs[0].record + rule = next(item for item in prepared.contract.decisions if item.family_code == record.decision.family_code) + missing_role = rule.required_roles[-1] + packet = replace( + record.packet, + transitions=tuple(item for item in record.packet.transitions if item.role != missing_role), + observations=tuple(item for item in record.packet.observations if item.role != missing_role), + ) + mutated = _record_with_packet(record, packet) + try: + family.validate_record(prepared.contract, mutated) + except ValueError as exc: + diagnosis = _coverage_diagnosis(ReadinessStage.CONSTRAINTS, mutated) + assert diagnosis is not None + return str(exc), diagnosis, "openultrasast.core.family.validate_record" + raise AssertionError("missing constraint was accepted") + + +def _binding(scope: ScopeKey, role: AdmissionCaseRole, pair_id: str, name: str) -> ReleasedCaseBinding: + draft = ReleasedCaseBinding( + canonical.content_digest(f"{name}:case"), + scope, + role, + pair_id, + canonical.content_digest(f"{name}:result"), + (canonical.content_digest(f"{name}:evidence"),), + (canonical.content_digest(f"{name}:change"),), + canonical.content_digest(f"{name}:manifest"), + "0" * 64, + ) + return replace(draft, binding_digest=derive_released_case_binding_digest(draft)) + + +def _pair_failure(prepared: _PreparedFamily) -> tuple[str, None, str]: + scope = ScopeKey(prepared.contract.family_id, prepared.ecosystem) + bindings = tuple( + sorted( + ( + _binding(scope, AdmissionCaseRole.VULNERABLE, canonical.content_digest("pair-a"), "vulnerable"), + _binding(scope, AdmissionCaseRole.FIXED, canonical.content_digest("pair-b"), "fixed"), + ), + key=lambda item: item.case_id_digest, + ) + ) + digest = canonical.content_digest("release-header") + release = AdmissionCaseRelease( + ADMISSION_CASE_RELEASE_SCHEMA, + digest, + digest, + digest, + digest, + digest, + digest, + bindings, + digest, + _setup().manifest.review, + digest, + ) + try: + _validate_release_shape(release) + except ValueError as exc: + return str(exc), None, "openultrasast.core._admission_release._validate_release_shape" + raise AssertionError("mismatched case pair was accepted") + + +def _development_failure( + prepared: _PreparedFamily, + identity: AnalysisIdentity, + prior: tuple[ReadinessResult, ...], + scratch: Path, +) -> str: + zero_path = scratch / f"{prepared.contract.family_id}-zero-development-input" + zero_path.write_bytes(b"") + manifest, receipts = readiness_contract.read_tracked_inputs((("development-case", zero_path),)) + try: + readiness_contract.build_development_observation( + arm_id="reference-arm", + family_id=prepared.contract.family_id, + ecosystem=prepared.ecosystem, + identity=identity, + population_digest=canonical.content_digest("development-population"), + population_freeze_digest=canonical.content_digest("development-freeze"), + population_use_receipt_digests=(canonical.content_digest("development-use"),), + predecessor_digests=tuple(item.result_digest for item in prior), + fixture_results=_fixtures(prepared.inputs), + input_manifest=manifest, + input_read_receipts=receipts, + ) + except ValueError as exc: + return str(exc) + raise AssertionError("zero-input development observation was accepted") + + +def observe_semantic_failure( + prepared: _PreparedFamily, + stage: ReadinessStage, + identity: AnalysisIdentity, + prior: tuple[ReadinessResult, ...], + scratch: Path, +) -> tuple[str, StageDiagnosis | None, str]: + observers = { + ReadinessStage.RECOGNITION: _recognition_failure, + ReadinessStage.TRANSITIONS: _transition_failure, + ReadinessStage.CONSTRAINTS: _constraint_failure, + ReadinessStage.PAIRS: _pair_failure, + } + if stage is ReadinessStage.DEVELOPMENT: + return ( + _development_failure(prepared, identity, prior, scratch), + None, + "benchmarks.valid.readiness_contract.build_development_observation", + ) + observer = observers.get(stage) + if observer is None: + raise AssertionError(f"unsupported semantic failure stage: {stage.value}") + return observer(prepared) + + +__all__ = ["observe_semantic_failure"] diff --git a/tests/detection_core_failure_visibility.py b/tests/detection_core_failure_visibility.py new file mode 100644 index 0000000..64ac982 --- /dev/null +++ b/tests/detection_core_failure_visibility.py @@ -0,0 +1,366 @@ +"""Test-owned failure injection over the production readiness contracts.""" + +from __future__ import annotations + +import enum +import hashlib +import shutil +from dataclasses import dataclass, replace +from pathlib import Path + +from detection_core_end_to_end import ( + _admission_result, + _admission_setup, + _baseline, + _fixtures, + _measurement, + _prepare_family, + _PreparedFamily, + _run_admission, + run_reference_family_path, +) +from detection_core_failure_inputs import observe_semantic_failure +from test_core_admission_context import _resign_manifest +from test_core_admission_metrics import _resign_census, _setup +from test_core_consumer_v2 import _request +from test_core_reference_families import _contract +from test_core_validation_v2 import _catalog + +from benchmarks.valid import readiness_contract +from openultrasast.core import admission_context, canonical, consumer, family +from openultrasast.core.admission_report import _make_replay, validate_admission_replay +from openultrasast.core.admission_spend import authorize_admission +from openultrasast.core.coverage import ( + InstrumentFailure, + StageDiagnosis, + validate_instrument, +) +from openultrasast.core.measurement import DenominatorPolicy, IntervalMethod, ScopeKey, standard_metric_rules +from openultrasast.core.readiness import ( + AnalysisIdentity, + ReadinessResult, + ReadinessStage, + StageAuthorization, + StageResult, + authorize_stage, +) + + +class FailureKind(enum.Enum): + UNREAD_INPUT = "unread_input" + ZERO_INPUT = "zero_input" + MISSING_RECOGNITION = "missing_recognition" + INCOMPLETE_TRANSITION = "incomplete_transition" + MISSING_CONSTRAINT = "missing_constraint" + PAIR_MISMATCH = "pair_mismatch" + FAILED_DEVELOPMENT = "failed_development" + CONTAMINATED_ADMISSION = "contaminated_admission" + REFUSED_INTEGRATION = "refused_integration" + + +@dataclass(frozen=True) +class FailureTrace: + failed_stage: ReadinessStage + reasons: tuple[str, ...] + correction_lever: str + completed_stages: tuple[ReadinessStage, ...] + metric_stages: tuple[ReadinessStage, ...] + later_metric_stages: tuple[ReadinessStage, ...] + authorization: StageAuthorization | None + label_read_count: int + coverage_diagnosis: StageDiagnosis | None + source_validator: str + + +@dataclass(frozen=True) +class IdentityStaleness: + targets: tuple[ReadinessStage, ...] + authorizations: tuple[StageAuthorization, ...] + old_readiness_digests: tuple[str, ...] + rebuilt_readiness_digests: tuple[str, ...] + measurement_readiness_digests: tuple[str, ...] + manifest_readiness_digests: tuple[str, ...] + authoritative_refusal: str + + +def _stage_index(stage: ReadinessStage) -> int: + return tuple(ReadinessStage).index(stage) + + +def _trace( + stage: ReadinessStage, + reasons: tuple[str, ...], + correction: str, + receipts: tuple[ReadinessResult, ...], + authorization: StageAuthorization | None, + *, + labels: int = 0, + diagnosis: StageDiagnosis | None = None, + source_validator: str, +) -> FailureTrace: + metrics = tuple(item.stage for item in receipts) + completed = tuple(item.stage for item in receipts if item.result is StageResult.PASSED) + later = tuple(item for item in metrics if _stage_index(item) > _stage_index(stage)) + return FailureTrace(stage, reasons, correction, completed, metrics, later, authorization, labels, diagnosis, source_validator) + + +def _identity(prepared: _PreparedFamily) -> AnalysisIdentity: + base = _setup() + contract = prepared.contract + catalog = _catalog(contract) + scope = ScopeKey(contract.family_id, prepared.ecosystem) + denominator = DenominatorPolicy(standard_metric_rules((scope,)), IntervalMethod.WILSON_95, "") + baseline = _baseline(base, catalog, scope, denominator, prepared.authority.authority_digest) + return admission_context.derive_analysis_identity(baseline, catalog, scope, None) + + +def _instrument_identity(fixture_root: Path, family_file: str, authority_manifest: Path) -> AnalysisIdentity: + contract = _contract(_load_json(fixture_root / family_file)) + catalog = _catalog(contract) + authority = authority_manifest.read_bytes() + if not authority: + raise ValueError("source authority manifest is empty") + return AnalysisIdentity( + contract.schema_version, + admission_context.derive_semantic_vocabulary_digest(), + "reference-analyzer@v1", + family.family_contract_digest(contract), + canonical.content_digest("empty-fact-set"), + hashlib.sha256(authority).hexdigest(), + canonical.content_digest("configuration"), + None, + admission_context.derive_outcome_mapping_digest(catalog), + ) + + +def _load_json(path: Path) -> dict[str, object]: + import json + + value = json.loads(path.read_bytes()) + if not isinstance(value, dict): + raise ValueError("fixture must contain an object") + return value + + +def _instrument_failure( + fixture_root: Path, + family_file: str, + repository: str, + failure: FailureKind, + scratch: Path, +) -> FailureTrace: + repository_root = scratch / f"{repository}-{failure.value}" + shutil.copytree(fixture_root / repository, repository_root) + authority_manifest = repository_root / "source_authority.json" + manifest_value = _load_json(authority_manifest) + cases = manifest_value.get("cases") + if not isinstance(cases, list) or not cases or not isinstance(cases[0], dict): + raise ValueError("source authority must name an authored case") + input_id = str(cases[0]["case_id"]) + path = repository_root / str(cases[0]["path"]) + if failure is FailureKind.UNREAD_INPUT: + path.unlink() + if failure is FailureKind.ZERO_INPUT: + path.write_bytes(b"") + manifest, receipts = readiness_contract.read_tracked_inputs(((input_id, path),)) + try: + validate_instrument(manifest, receipts) + except InstrumentFailure as exc: + reason = str(exc) + else: + raise AssertionError("injected bad input was accepted") + identity = _instrument_identity(fixture_root, family_file, authority_manifest) + authorization = authorize_stage(ReadinessStage.RECOGNITION, (), identity) + return _trace( + ReadinessStage.INSTRUMENT, + (reason,), + authorization.correction_lever or "repair instrument", + (), + authorization, + source_validator="openultrasast.core.coverage.validate_instrument", + ) + + +def _semantic_failure(prepared: _PreparedFamily, failure: FailureKind, scratch: Path) -> FailureTrace: + stage = { + FailureKind.MISSING_RECOGNITION: ReadinessStage.RECOGNITION, + FailureKind.INCOMPLETE_TRANSITION: ReadinessStage.TRANSITIONS, + FailureKind.MISSING_CONSTRAINT: ReadinessStage.CONSTRAINTS, + FailureKind.PAIR_MISMATCH: ReadinessStage.PAIRS, + FailureKind.FAILED_DEVELOPMENT: ReadinessStage.DEVELOPMENT, + }[failure] + identity = _identity(prepared) + results: list[ReadinessResult] = [] + for current in tuple(ReadinessStage)[: _stage_index(stage)]: + if results: + readiness_contract.require_stage_authorized(current, tuple(results), identity) + results.append(_passed_result(prepared, identity, current, results)) + reason, diagnosis, validator = observe_semantic_failure(prepared, stage, identity, tuple(results), scratch) + target = tuple(ReadinessStage)[_stage_index(stage) + 1] + authorization = authorize_stage(target, tuple(results), identity) + correction = authorization.correction_lever or "repair readiness failure" + return _trace( + stage, + (reason,), + correction, + tuple(results), + authorization, + diagnosis=diagnosis, + source_validator=validator, + ) + + +def _passed_result( + prepared: _PreparedFamily, + identity: AnalysisIdentity, + stage: ReadinessStage, + prior: list[ReadinessResult], +) -> ReadinessResult: + return readiness_contract.build_readiness_result( + arm_id="reference-arm", + family_id=prepared.contract.family_id, + ecosystem=prepared.ecosystem, + stage=stage, + result=StageResult.PASSED, + identity=identity, + predecessor_digests=tuple(item.result_digest for item in prior), + fixture_results=_fixtures(prepared.inputs), + measurement_digest=_measurement(stage, prepared), + correction_lever=None, + population_digest=None, + population_freeze_digest=None, + population_use_receipt_digests=(), + ) + + +def _contaminated_admission(prepared: _PreparedFamily) -> FailureTrace: + setup = _admission_setup(prepared) + exposed = replace(setup.manifest.census, case_labels_exposed=True) + census = _resign_census(exposed, exposed.rows[0], exposed.population_digest) + manifest = _resign_manifest(replace(setup.manifest, census=census), census.review.conflict_disclosure) + try: + authorize_admission(manifest, setup.denominator, setup.readiness, setup.context, setup.custody, setup.resolver) + except ValueError as exc: + reason = str(exc) + else: + raise AssertionError("contaminated admission was accepted") + authorization = authorize_stage(ReadinessStage.INTEGRATION, setup.readiness, setup.readiness[0].identity) + return _trace( + ReadinessStage.ADMISSION, + (reason,), + authorization.correction_lever or "repair admission", + setup.readiness, + authorization, + source_validator="openultrasast.core.admission_spend.authorize_admission", + ) + + +def _refused_integration(prepared: _PreparedFamily, scratch: Path) -> FailureTrace: + setup = _admission_setup(prepared) + replay, _status, label_reads = _run_admission(setup, scratch) + admission = _admission_result(setup, replay, setup.readiness) + prior = (*setup.readiness, admission) + readiness_contract.require_stage_authorized(ReadinessStage.INTEGRATION, prior, prior[0].identity) + request, resolver = _request(prepared.inputs[0], setup.resolver) + try: + consumer.consume_records(request, (), setup.context.evidence, setup.custody, 1, resolver) + except ValueError as exc: + reason = str(exc) + else: + raise AssertionError("consumer accepted missing evidence") + return _trace( + ReadinessStage.INTEGRATION, + (reason,), + "repair consumer evidence binding", + prior, + None, + labels=label_reads, + source_validator="openultrasast.core.consumer.consume_records", + ) + + +def run_stage_failure( + *, + fixture_root: Path, + family_file: str, + cases_file: str, + repository: str, + failure: FailureKind, + scratch: Path, +) -> FailureTrace: + if failure in (FailureKind.UNREAD_INPUT, FailureKind.ZERO_INPUT): + return _instrument_failure(fixture_root, family_file, repository, failure, scratch) + prepared = _prepare_family(fixture_root, family_file, cases_file, repository) + if failure is FailureKind.CONTAMINATED_ADMISSION: + return _contaminated_admission(prepared) + if failure is FailureKind.REFUSED_INTEGRATION: + return _refused_integration(prepared, scratch) + return _semantic_failure(prepared, failure, scratch) + + +def _rebuild(readiness: tuple[ReadinessResult, ...], identity: AnalysisIdentity) -> tuple[ReadinessResult, ...]: + rebuilt: list[ReadinessResult] = [] + for item in readiness: + rebuilt.append( + readiness_contract.build_readiness_result( + arm_id=item.arm_id, + family_id=item.family_id, + ecosystem=item.ecosystem, + stage=item.stage, + result=item.result, + identity=identity, + predecessor_digests=tuple(value.result_digest for value in rebuilt), + fixture_results=item.fixture_results, + measurement_digest=item.measurement_digest, + correction_lever=item.correction_lever, + population_digest=item.population_digest, + population_freeze_digest=item.population_freeze_digest, + population_use_receipt_digests=item.population_use_receipt_digests, + ) + ) + return tuple(rebuilt) + + +def run_identity_staleness(*, fixture_root: Path, family_file: str, cases_file: str, repository: str, scratch: Path) -> IdentityStaleness: + result = run_reference_family_path( + fixture_root=fixture_root, + family_file=family_file, + cases_file=cases_file, + repository=repository, + scratch=scratch, + ) + current = result.readiness[0].identity + changed = replace(current, semantic_vocabulary_digest=canonical.content_digest("changed-semantic-vocabulary")) + targets = tuple(ReadinessStage)[1:] + authorizations = tuple(authorize_stage(target, result.readiness[: _stage_index(target)], changed) for target in targets) + rebuilt = _rebuild(result.readiness, changed) + measurement = result.admission.report.measurement + replay = _make_replay( + result.admission.manifest, + result.admission.denominator, + rebuilt[:6], + result.admission.preparation, + result.admission.label_release_receipt, + result.admission.case_release, + result.admission.cases, + result.admission.report, + ) + current_setup = _admission_setup(_prepare_family(fixture_root, family_file, cases_file, repository)) + try: + validate_admission_replay(replay, current_setup.context, current_setup.custody, current_setup.resolver) + except ValueError as exc: + refusal = str(exc) + else: + raise AssertionError("stale admission measurement was accepted against rebuilt readiness") + return IdentityStaleness( + targets, + authorizations, + tuple(item.result_digest for item in result.readiness), + tuple(item.result_digest for item in rebuilt), + tuple(item.result_digest for item in measurement.prerequisite_readiness), + result.admission.manifest.readiness_digests, + refusal, + ) + + +__all__ = ["FailureKind", "FailureTrace", "IdentityStaleness", "run_identity_staleness", "run_stage_failure"] diff --git a/tests/test_detection_core_failure_visibility.py b/tests/test_detection_core_failure_visibility.py new file mode 100644 index 0000000..f31fdf4 --- /dev/null +++ b/tests/test_detection_core_failure_visibility.py @@ -0,0 +1,182 @@ +"""Every semantic/governance failure stops at its first detection-core stage.""" + +from __future__ import annotations + +from pathlib import Path + +import detection_core_failure_visibility as visibility +import pytest +from detection_core_failure_visibility import FailureKind, run_identity_staleness, run_stage_failure + +from openultrasast.core.coverage import CoverageStage +from openultrasast.core.readiness import ReadinessStage + +FIXTURES = Path(__file__).parent / "fixtures" / "detection_core" + +FAMILIES = ( + ("obligation_family.json", "obligation_cases.json", "obligation_repository"), + ("taint_family.json", "taint_cases.json", "taint_repository"), +) + +FAILURES = ( + (FailureKind.UNREAD_INPUT, ReadinessStage.INSTRUMENT), + (FailureKind.ZERO_INPUT, ReadinessStage.INSTRUMENT), + (FailureKind.MISSING_RECOGNITION, ReadinessStage.RECOGNITION), + (FailureKind.INCOMPLETE_TRANSITION, ReadinessStage.TRANSITIONS), + (FailureKind.MISSING_CONSTRAINT, ReadinessStage.CONSTRAINTS), + (FailureKind.PAIR_MISMATCH, ReadinessStage.PAIRS), + (FailureKind.FAILED_DEVELOPMENT, ReadinessStage.DEVELOPMENT), + (FailureKind.CONTAMINATED_ADMISSION, ReadinessStage.ADMISSION), + (FailureKind.REFUSED_INTEGRATION, ReadinessStage.INTEGRATION), +) + +EXPECTED_FAILURE = { + FailureKind.UNREAD_INPUT: ("tracked input not opened", "produce instrument readiness evidence"), + FailureKind.ZERO_INPUT: ("zero-byte tracked input", "produce instrument readiness evidence"), + FailureKind.MISSING_RECOGNITION: ( + "security object identity does not match", + "produce recognition readiness evidence", + ), + FailureKind.INCOMPLETE_TRANSITION: ( + "non-complete semantics or completeness reasons forbid a determinate outcome", + "produce transitions readiness evidence", + ), + FailureKind.MISSING_CONSTRAINT: ("required role is missing", "produce constraints readiness evidence"), + FailureKind.PAIR_MISMATCH: ( + "every scoped release pair must contain one vulnerable and one fixed case", + "produce pairs readiness evidence", + ), + FailureKind.FAILED_DEVELOPMENT: ("zero-byte tracked input", "produce development readiness evidence"), + FailureKind.CONTAMINATED_ADMISSION: ("census must not expose case labels", "produce admission readiness evidence"), + FailureKind.REFUSED_INTEGRATION: ("push inputs must be nonempty", "repair consumer evidence binding"), +} + + +@pytest.mark.parametrize(("family_file", "cases_file", "repository"), FAMILIES) +@pytest.mark.parametrize(("failure", "failed_stage"), FAILURES) +def test_each_injected_failure_stops_at_its_first_stage( + tmp_path: Path, + family_file: str, + cases_file: str, + repository: str, + failure: FailureKind, + failed_stage: ReadinessStage, +) -> None: + trace = run_stage_failure( + fixture_root=FIXTURES, + family_file=family_file, + cases_file=cases_file, + repository=repository, + failure=failure, + scratch=tmp_path, + ) + + assert trace.failed_stage is failed_stage + reason, correction = EXPECTED_FAILURE[failure] + assert reason in trace.reasons[0] + assert trace.correction_lever == correction + assert trace.later_metric_stages == () + stage_index = tuple(ReadinessStage).index(failed_stage) + assert trace.completed_stages == tuple(ReadinessStage)[:stage_index] + assert trace.metric_stages == tuple(ReadinessStage)[:stage_index] + assert trace.label_read_count == (1 if failure is FailureKind.REFUSED_INTEGRATION else 0) + assert bool(tuple(tmp_path.glob("*-labels.json"))) is (failure is FailureKind.REFUSED_INTEGRATION) + if failure is FailureKind.REFUSED_INTEGRATION: + assert trace.authorization is None + else: + assert trace.authorization is not None + assert trace.authorization.authorized is False + assert trace.authorization.failing_stage is failed_stage + assert trace.authorization.correction_lever == correction + + if failure is FailureKind.MISSING_RECOGNITION: + assert trace.coverage_diagnosis is not None + assert trace.coverage_diagnosis.last_verified_nonzero_stage is CoverageStage.SITES_CONSIDERED + assert trace.coverage_diagnosis.first_failed_output_stage is CoverageStage.SITES_RECOGNIZED + elif failure is FailureKind.MISSING_CONSTRAINT: + assert trace.coverage_diagnosis is not None + assert trace.coverage_diagnosis.last_verified_nonzero_stage is CoverageStage.SECURITY_OBJECTS + assert trace.coverage_diagnosis.first_failed_output_stage is CoverageStage.CONSTRAINTS + + +@pytest.mark.parametrize(("family_file", "cases_file", "repository"), FAMILIES) +@pytest.mark.parametrize(("failure", "failed_stage"), FAILURES) +def test_failure_source_is_a_production_validator_and_later_work_is_not_invoked( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + family_file: str, + cases_file: str, + repository: str, + failure: FailureKind, + failed_stage: ReadinessStage, +) -> None: + calls: list[tuple[str, object]] = [] + + def record(name: str, original: object): + def wrapped(*args: object, **kwargs: object) -> object: + calls.append((name, args[0] if args else None)) + return original(*args, **kwargs) # type: ignore[operator] + + return wrapped + + monkeypatch.setattr(visibility, "_measurement", record("measurement", visibility._measurement)) + monkeypatch.setattr( + visibility, + "authorize_admission", + record("authorize_admission", visibility.authorize_admission), + ) + monkeypatch.setattr(visibility, "_run_admission", record("run_admission", visibility._run_admission)) + monkeypatch.setattr( + visibility.consumer, + "consume_records", + record("consume_records", visibility.consumer.consume_records), + ) + + trace = visibility.run_stage_failure( + fixture_root=FIXTURES, + family_file=family_file, + cases_file=cases_file, + repository=repository, + failure=failure, + scratch=tmp_path, + ) + + assert trace.source_validator.startswith("openultrasast.") or trace.source_validator.startswith("benchmarks.valid.") + measurement_stages = tuple(value for name, value in calls if name == "measurement") + expected_measurements = ( + tuple(ReadinessStage)[: tuple(ReadinessStage).index(failed_stage)] + if failed_stage.value in {"recognition", "transitions", "constraints", "pairs", "development"} + else () + ) + assert measurement_stages == expected_measurements + assert sum(name == "authorize_admission" for name, _value in calls) == int(failure is FailureKind.CONTAMINATED_ADMISSION) + assert sum(name == "run_admission" for name, _value in calls) == int(failure is FailureKind.REFUSED_INTEGRATION) + assert sum(name == "consume_records" for name, _value in calls) == int(failure is FailureKind.REFUSED_INTEGRATION) + + +@pytest.mark.parametrize(("family_file", "cases_file", "repository"), FAMILIES) +def test_semantic_identity_change_stales_all_later_readiness_and_measurement( + tmp_path: Path, + family_file: str, + cases_file: str, + repository: str, +) -> None: + result = run_identity_staleness( + fixture_root=FIXTURES, + family_file=family_file, + cases_file=cases_file, + repository=repository, + scratch=tmp_path, + ) + + assert result.targets == tuple(ReadinessStage)[1:] + assert all(not item.authorized for item in result.authorizations) + assert all(item.failing_stage is ReadinessStage.INSTRUMENT for item in result.authorizations) + assert all(item.correction_lever == "rerun instrument with current analysis identity" for item in result.authorizations) + assert all(old != new for old, new in zip(result.old_readiness_digests, result.rebuilt_readiness_digests, strict=True)) + assert result.measurement_readiness_digests == result.old_readiness_digests[:6] + assert result.measurement_readiness_digests != result.rebuilt_readiness_digests[:6] + assert result.manifest_readiness_digests == tuple(sorted(result.old_readiness_digests[:6])) + assert result.manifest_readiness_digests != tuple(sorted(result.rebuilt_readiness_digests[:6])) + assert "authoritative admission replay refused" in result.authoritative_refusal + assert "readiness" in result.authoritative_refusal