diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 36703a6..a562ab3 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -335,7 +335,7 @@ - _Requirements: 10.2, 10.4, 11.1, 11.2, 11.3, 11.4, 11.5, 13.4, 14.3_ - _Boundary: Fact Pack Adapter Conformance_ - _Depends: 5.4_ -- [ ] 9.5 (P) Produce readiness and development records through benchmark-owned adapters +- [x] 9.5 (P) Produce readiness and development records through benchmark-owned adapters - Construct canonical readiness results while leaving sequencing and persistence with benchmark workflows. - Emit development observations with v2 identities, evidence-use roles, instrument receipts, and prerequisite ancestry. - Stop a runner before population work when readiness is failed, absent, or stale. diff --git a/benchmarks/valid/development_runner.py b/benchmarks/valid/development_runner.py index f3f7947..b84fb25 100644 --- a/benchmarks/valid/development_runner.py +++ b/benchmarks/valid/development_runner.py @@ -14,18 +14,22 @@ from __future__ import annotations import hashlib import json import time +from collections.abc import Callable from dataclasses import asdict, dataclass from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, TypeVar if TYPE_CHECKING: from openultrasast.hunt.prover import ProverResult +from benchmarks.valid.readiness_contract import require_stage_authorized from benchmarks.valid.run_manifests import AnalyzerBundle +from openultrasast.core.readiness import AnalysisIdentity, ReadinessResult, ReadinessStage from openultrasast.hunt.producers import Context from openultrasast.hunt.prover import prove_change _STAGE_KEYS = ("entry_recognition", "resolution", "judgment") +_T = TypeVar("_T") @dataclass(frozen=True) @@ -118,6 +122,16 @@ def run_development( return tuple(records) +def run_development_when_ready( + readiness_receipts: tuple[ReadinessResult, ...], + current_identity: AnalysisIdentity, + population_work: Callable[[], _T], +) -> _T: + """Run caller-owned development work only after all prior stages pass.""" + require_stage_authorized(ReadinessStage.DEVELOPMENT, readiness_receipts, current_identity) + return population_work() + + class GateFailure(RuntimeError): """A measured change exceeded the hard runtime budget (Req 6.1).""" @@ -247,6 +261,7 @@ __all__ = [ "four_arm_report", "judgment_transition", "run_development", + "run_development_when_ready", "stage_funnel", "summarize_development", ] diff --git a/benchmarks/valid/readiness_contract.py b/benchmarks/valid/readiness_contract.py new file mode 100644 index 0000000..8b24c1c --- /dev/null +++ b/benchmarks/valid/readiness_contract.py @@ -0,0 +1,246 @@ +"""Benchmark-owned adapter for detection-core v2 readiness records. + +Core owns immutable contracts and pure validation. Benchmark workflows retain +sequencing, persistence, and population work; this module only constructs the +canonical values and enforces the readiness gate before that work starts. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from pathlib import Path + +from openultrasast.core.canonical import JsonValue, content_digest +from openultrasast.core.coverage import InputManifest, InputReadReceipt, derive_input_manifest_digest, validate_instrument +from openultrasast.core.readiness import ( + AnalysisIdentity, + FixtureResult, + ReadinessEvidenceUse, + ReadinessResult, + ReadinessStage, + ReadinessValidationError, + StageAuthorization, + StageResult, + authorize_stage, + derive_readiness_result_digest, + required_evidence_use, + validate_readiness_result, +) + + +class ReadinessBlocked(RuntimeError): + """Population work is not authorized by current readiness evidence.""" + + def __init__( + self, + failing_stage: ReadinessStage | None, + reasons: tuple[str, ...], + correction_lever: str | None, + ) -> None: + self.failing_stage = failing_stage + self.reasons = reasons + self.correction_lever = correction_lever + layer = "readiness contract" if failing_stage is None else failing_stage.value + correction = "" if correction_lever is None else f"; correction: {correction_lever}" + super().__init__(f"{layer}: {'; '.join(reasons)}{correction}") + + +def _read_input(input_id: str, path: Path) -> InputReadReceipt: + try: + payload = path.read_bytes() + except OSError: + return InputReadReceipt(input_id, str(path), False, 0, None) + return InputReadReceipt(input_id, str(path), True, len(payload), hashlib.sha256(payload).hexdigest()) + + +def read_tracked_inputs( + inputs: tuple[tuple[str, Path], ...], +) -> tuple[InputManifest, tuple[InputReadReceipt, ...]]: + """Open tracked inputs and emit receipts; validation later refuses bad reads.""" + input_ids = tuple(input_id for input_id, _path in inputs) + manifest = InputManifest(input_ids, derive_input_manifest_digest(input_ids)) + receipts = tuple(_read_input(input_id, path) for input_id, path in inputs) + return manifest, receipts + + +def _read_receipt_value(receipt: InputReadReceipt) -> JsonValue: + return { + "input_id": receipt.input_id, + "path": receipt.path, + "opened": receipt.opened, + "byte_count": receipt.byte_count, + "content_digest": receipt.content_digest, + } + + +def _instrument_value(manifest: InputManifest, receipts: tuple[InputReadReceipt, ...]) -> JsonValue: + return { + "tracked_input_ids": manifest.tracked_input_ids, + "manifest_digest": manifest.manifest_digest, + "receipts": tuple(_read_receipt_value(item) for item in receipts), + } + + +def derive_development_measurement_digest( + manifest: InputManifest, + receipts: tuple[InputReadReceipt, ...], +) -> str: + """Bind a development measurement to the exact verified instrument reads.""" + validate_instrument(manifest, receipts) + return content_digest(_instrument_value(manifest, receipts)) + + +def build_readiness_result( + *, + arm_id: str, + family_id: str, + ecosystem: str, + stage: ReadinessStage, + result: StageResult, + identity: AnalysisIdentity, + predecessor_digests: tuple[str, ...], + fixture_results: tuple[FixtureResult, ...], + measurement_digest: str, + correction_lever: str | None, + population_digest: str | None, + population_freeze_digest: str | None, + population_use_receipt_digests: tuple[str, ...], +) -> ReadinessResult: + evidence_use = required_evidence_use(stage) + digest = derive_readiness_result_digest( + arm_id=arm_id, + family_id=family_id, + ecosystem=ecosystem, + stage=stage, + evidence_use=evidence_use, + population_digest=population_digest, + population_freeze_digest=population_freeze_digest, + population_use_receipt_digests=population_use_receipt_digests, + result=result, + identity=identity, + predecessor_digests=predecessor_digests, + fixture_results=fixture_results, + measurement_digest=measurement_digest, + correction_lever=correction_lever, + ) + return ReadinessResult( + arm_id, + family_id, + ecosystem, + stage, + evidence_use, + population_digest, + population_freeze_digest, + population_use_receipt_digests, + result, + identity, + predecessor_digests, + fixture_results, + measurement_digest, + correction_lever, + digest, + ) + + +@dataclass(frozen=True) +class DevelopmentObservation: + """A v2 development result plus the instrument evidence that produced it.""" + + readiness_result: ReadinessResult + input_manifest: InputManifest + input_read_receipts: tuple[InputReadReceipt, ...] + observation_digest: str + + def __post_init__(self) -> None: + validate_development_observation(self) + + +def _observation_value(value: DevelopmentObservation) -> JsonValue: + return { + "readiness_result_digest": value.readiness_result.result_digest, + "instrument": _instrument_value(value.input_manifest, value.input_read_receipts), + } + + +def validate_development_observation(value: DevelopmentObservation) -> None: + """Revalidate the readiness result, instrument, and their exact binding.""" + validate_readiness_result(value.readiness_result) + if value.readiness_result.stage is not ReadinessStage.DEVELOPMENT: + raise ReadinessValidationError("development observation must use the development stage") + if value.readiness_result.evidence_use is not ReadinessEvidenceUse.DEVELOPMENT_POPULATION: + raise ReadinessValidationError("development observation must use development population evidence") + expected_measurement = derive_development_measurement_digest(value.input_manifest, value.input_read_receipts) + if value.readiness_result.measurement_digest != expected_measurement: + raise ReadinessValidationError("development measurement does not match instrument receipts") + if value.observation_digest != content_digest(_observation_value(value)): + raise ReadinessValidationError("development observation digest does not match content") + + +def build_development_observation( + *, + arm_id: str, + family_id: str, + ecosystem: str, + identity: AnalysisIdentity, + population_digest: str, + population_freeze_digest: str, + population_use_receipt_digests: tuple[str, ...], + predecessor_digests: tuple[str, ...], + fixture_results: tuple[FixtureResult, ...], + input_manifest: InputManifest, + input_read_receipts: tuple[InputReadReceipt, ...], + result: StageResult = StageResult.PASSED, + correction_lever: str | None = None, +) -> DevelopmentObservation: + """Create a development observation only from verified nonzero input reads.""" + measurement = derive_development_measurement_digest(input_manifest, input_read_receipts) + readiness_result = build_readiness_result( + arm_id=arm_id, + family_id=family_id, + ecosystem=ecosystem, + stage=ReadinessStage.DEVELOPMENT, + result=result, + identity=identity, + predecessor_digests=predecessor_digests, + fixture_results=fixture_results, + measurement_digest=measurement, + correction_lever=correction_lever, + population_digest=population_digest, + population_freeze_digest=population_freeze_digest, + population_use_receipt_digests=population_use_receipt_digests, + ) + probe = object.__new__(DevelopmentObservation) + object.__setattr__(probe, "readiness_result", readiness_result) + object.__setattr__(probe, "input_manifest", input_manifest) + object.__setattr__(probe, "input_read_receipts", input_read_receipts) + object.__setattr__(probe, "observation_digest", "") + digest = content_digest(_observation_value(probe)) + return DevelopmentObservation(readiness_result, input_manifest, input_read_receipts, digest) + + +def require_stage_authorized( + target: ReadinessStage, + receipts: tuple[ReadinessResult, ...], + current_identity: AnalysisIdentity, +) -> StageAuthorization: + """Refuse population work on missing, failed, stale, or malformed readiness.""" + try: + authorization = authorize_stage(target, receipts, current_identity) + except ReadinessValidationError as exc: + raise ReadinessBlocked(None, (str(exc),), None) from exc + if not authorization.authorized: + raise ReadinessBlocked(authorization.failing_stage, authorization.reasons, authorization.correction_lever) + return authorization + + +__all__ = [ + "DevelopmentObservation", + "ReadinessBlocked", + "build_development_observation", + "build_readiness_result", + "derive_development_measurement_digest", + "read_tracked_inputs", + "require_stage_authorized", + "validate_development_observation", +] diff --git a/tests/test_detection_core_readiness_adapter.py b/tests/test_detection_core_readiness_adapter.py new file mode 100644 index 0000000..fbc0d7c --- /dev/null +++ b/tests/test_detection_core_readiness_adapter.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import sys +from dataclasses import replace +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from benchmarks.valid import development_runner, readiness_contract +from openultrasast.core import canonical, coverage, readiness + + +def _digest(label: str) -> str: + return canonical.content_digest(label) + + +def _identity() -> readiness.AnalysisIdentity: + return readiness.AnalysisIdentity( + core_schema_version="v2", + semantic_vocabulary_digest=_digest("semantics"), + analyzer_revision="analyzer-v2", + family_contract_digest=_digest("family"), + fact_set_digest=_digest("facts"), + acquisition_record_digest=_digest("acquisition"), + configuration_digest=_digest("configuration"), + model_program_digest=None, + outcome_mapping_digest=_digest("mapping"), + ) + + +def _fixtures(stage: readiness.ReadinessStage) -> tuple[readiness.FixtureResult, ...]: + return tuple(readiness.FixtureResult(role, _digest(f"{stage.value}:{role.value}"), True, ()) for role in readiness.FixtureRole) + + +def _result( + stage: readiness.ReadinessStage, + identity: readiness.AnalysisIdentity, + predecessors: tuple[str, ...], + *, + result: readiness.StageResult = readiness.StageResult.PASSED, + correction_lever: str | None = None, +) -> readiness.ReadinessResult: + population = stage is readiness.ReadinessStage.DEVELOPMENT + fixtures = _fixtures(stage) + if result is readiness.StageResult.FAILED: + fixtures = (*fixtures[:2], replace(fixtures[2], passed=False, reasons=("stage failed",))) + return readiness_contract.build_readiness_result( + arm_id="reference-arm", + family_id="reference-family", + ecosystem="python", + stage=stage, + result=result, + identity=identity, + predecessor_digests=predecessors, + fixture_results=fixtures, + measurement_digest=_digest(f"measurement:{stage.value}"), + correction_lever=correction_lever, + population_digest=_digest("development-population") if population else None, + population_freeze_digest=_digest("development-freeze") if population else None, + population_use_receipt_digests=(_digest("development-use"),) if population else (), + ) + + +def _prior(identity: readiness.AnalysisIdentity) -> tuple[readiness.ReadinessResult, ...]: + results: list[readiness.ReadinessResult] = [] + for stage in tuple(readiness.ReadinessStage)[:5]: + results.append(_result(stage, identity, tuple(item.result_digest for item in results))) + return tuple(results) + + +def _instrument() -> tuple[coverage.InputManifest, tuple[coverage.InputReadReceipt, ...]]: + return readiness_contract.read_tracked_inputs((("fixture.py", Path(__file__)),)) + + +def test_instrument_receipt_is_derived_from_an_opened_nonzero_file() -> None: + manifest, receipts = _instrument() + assert manifest.tracked_input_ids == ("fixture.py",) + assert receipts[0].opened is True + assert receipts[0].byte_count == Path(__file__).stat().st_size + assert receipts[0].byte_count > 0 + + +def test_development_observation_carries_v2_readiness_and_instrument_evidence() -> None: + identity = _identity() + prior = _prior(identity) + manifest, receipts = _instrument() + observation = readiness_contract.build_development_observation( + arm_id="reference-arm", + family_id="reference-family", + ecosystem="python", + identity=identity, + population_digest=_digest("development-population"), + population_freeze_digest=_digest("development-freeze"), + population_use_receipt_digests=(_digest("development-use"),), + predecessor_digests=tuple(item.result_digest for item in prior), + fixture_results=_fixtures(readiness.ReadinessStage.DEVELOPMENT), + input_manifest=manifest, + input_read_receipts=receipts, + ) + assert observation.readiness_result.stage is readiness.ReadinessStage.DEVELOPMENT + assert observation.readiness_result.evidence_use is readiness.ReadinessEvidenceUse.DEVELOPMENT_POPULATION + assert observation.readiness_result.identity == identity + assert observation.readiness_result.predecessor_digests == tuple(item.result_digest for item in prior) + assert observation.input_read_receipts == receipts + + +@pytest.mark.parametrize("failure", ("missing", "failed", "stale")) +def test_runner_stops_before_population_work(failure: str) -> None: + identity = _identity() + prior = list(_prior(identity)) + if failure == "missing": + prior.pop(2) + elif failure == "failed": + prior[2] = _result( + readiness.ReadinessStage.TRANSITIONS, + identity, + tuple(item.result_digest for item in prior[:2]), + result=readiness.StageResult.FAILED, + correction_lever="repair transitions", + ) + else: + stale_identity = replace(identity, semantic_vocabulary_digest=_digest("stale-semantics")) + prior[2] = _result( + readiness.ReadinessStage.TRANSITIONS, + stale_identity, + tuple(item.result_digest for item in prior[:2]), + ) + called = False + + def population_work() -> str: + nonlocal called + called = True + return "ran" + + with pytest.raises(readiness_contract.ReadinessBlocked) as caught: + development_runner.run_development_when_ready(tuple(prior), identity, population_work) + assert called is False + expected = { + "missing": ("missing transitions readiness evidence", "produce transitions readiness evidence"), + "failed": ("transitions readiness failed", "repair transitions"), + "stale": ("transitions readiness is stale", "rerun transitions with current analysis identity"), + } + reason, correction = expected[failure] + assert caught.value.failing_stage is readiness.ReadinessStage.TRANSITIONS + assert caught.value.reasons == (reason,) + assert caught.value.correction_lever == correction + + +def test_runner_executes_population_work_after_current_prerequisites() -> None: + identity = _identity() + assert development_runner.run_development_when_ready(_prior(identity), identity, lambda: "ran") == "ran" + + +def test_runner_refuses_wrong_evidence_use_before_population_work() -> None: + identity = _identity() + prior = list(_prior(identity)) + malformed = object.__new__(readiness.ReadinessResult) + for field, value in prior[2].__dict__.items(): + object.__setattr__(malformed, field, value) + object.__setattr__(malformed, "evidence_use", readiness.ReadinessEvidenceUse.FRESH_ADMISSION) + prior[2] = malformed + called = False + + def population_work() -> None: + nonlocal called + called = True + + with pytest.raises(readiness_contract.ReadinessBlocked, match="wrong evidence use"): + development_runner.run_development_when_ready(tuple(prior), identity, population_work) + assert called is False + + +def test_semantic_identity_change_prevents_later_admission() -> None: + identity = _identity() + prior = _prior(identity) + manifest, receipts = _instrument() + development = readiness_contract.build_development_observation( + arm_id="reference-arm", + family_id="reference-family", + ecosystem="python", + identity=identity, + population_digest=_digest("development-population"), + population_freeze_digest=_digest("development-freeze"), + population_use_receipt_digests=(_digest("development-use"),), + predecessor_digests=tuple(item.result_digest for item in prior), + fixture_results=_fixtures(readiness.ReadinessStage.DEVELOPMENT), + input_manifest=manifest, + input_read_receipts=receipts, + ).readiness_result + changed = replace(identity, semantic_vocabulary_digest=_digest("changed-semantics")) + with pytest.raises(readiness_contract.ReadinessBlocked, match="stale"): + readiness_contract.require_stage_authorized(readiness.ReadinessStage.ADMISSION, (*prior, development), changed) + + +def test_development_observation_refuses_zero_byte_input() -> None: + identity = _identity() + prior = _prior(identity) + manifest, receipts = _instrument() + zero = (replace(receipts[0], byte_count=0),) + with pytest.raises(coverage.InstrumentFailure, match="zero-byte"): + readiness_contract.build_development_observation( + arm_id="reference-arm", + family_id="reference-family", + ecosystem="python", + identity=identity, + population_digest=_digest("development-population"), + population_freeze_digest=_digest("development-freeze"), + population_use_receipt_digests=(_digest("development-use"),), + predecessor_digests=tuple(item.result_digest for item in prior), + fixture_results=_fixtures(readiness.ReadinessStage.DEVELOPMENT), + input_manifest=manifest, + input_read_receipts=zero, + )