diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 2cf135c..0ffcb53 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -256,7 +256,7 @@ - Require every grouped threshold to have at least one jointly attainable count assignment. - Done when the historical impossible-denominator case is a configuration refusal rather than a measured no-go. - _Requirements: 12.3, 12.4_ -- [ ] 8.6 Validate the pre-label population-spend protocol inside core +- [x] 8.6 Validate the pre-label population-spend protocol inside core - Construct a label-release request bound to the exact validated authorization and current custody-log head. - Validate the returned receipt as the exact current-chain direct successor that changes the named population from available to spent. - Reject missing, stale, mismatched, non-successor, or availability-restoring receipts before scoring authorization. diff --git a/src/openultrasast/core/admission_spend.py b/src/openultrasast/core/admission_spend.py new file mode 100644 index 0000000..110e42f --- /dev/null +++ b/src/openultrasast/core/admission_spend.py @@ -0,0 +1,288 @@ +"""Pure spend-before-label authorization for admission scoring.""" + +from __future__ import annotations + +import re +from collections.abc import Sequence +from dataclasses import dataclass + +from openultrasast.core import canonical +from openultrasast.core._closed_graph import exact_dataclass_problem +from openultrasast.core.admission_context import AdmissionValidationContext, validate_admission_context +from openultrasast.core.admission_feasibility import evaluate_threshold_feasibility +from openultrasast.core.admission_review import AdmissionManifestV2 +from openultrasast.core.custody import ( + LineagePurpose, + PopulationCustodyLog, + PopulationCustodyPolicy, + PopulationCustodyPort, + PopulationStatus, + PopulationUseReceipt, + PopulationUseRequest, + append_population_use, + derive_population_use_request_digest, + empty_population_custody_log, + validate_population_custody_log, +) +from openultrasast.core.measurement import DenominatorPolicy +from openultrasast.core.readiness import ReadinessResult + +ADMISSION_AUTHORIZATION_SCHEMA = "openultrasast.detection-core.admission-authorization.v2" +ADMISSION_PREPARATION_SCHEMA = "openultrasast.detection-core.admission-preparation.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class AdmissionSpendError(ValueError): + """The pre-label population spend cannot authorize scoring.""" + + +@dataclass(frozen=True) +class AdmissionAuthorization: + manifest_digest: str + validation_context_digest: str + readiness_digests: tuple[str, ...] + custody_log_digest: str + feasibility_digest: str + authorization_digest: str + + +@dataclass(frozen=True) +class AdmissionPreparation: + authorization: AdmissionAuthorization + label_release_request: PopulationUseRequest + preparation_digest: str + + +@dataclass(frozen=True) +class _HistoricalCustody: + current: PopulationCustodyPort + population_digest: str + historical_log: PopulationCustodyLog + + def current_log(self, population_digest: str) -> PopulationCustodyLog: + if population_digest == self.population_digest: + return self.historical_log + return self.current.current_log(population_digest) + + +def _require_digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise AdmissionSpendError(f"{field} must be a full lowercase SHA-256 digest") + + +def _authorization_value(authorization: AdmissionAuthorization) -> canonical.JsonValue: + return { + "schema_version": ADMISSION_AUTHORIZATION_SCHEMA, + "manifest_digest": authorization.manifest_digest, + "validation_context_digest": authorization.validation_context_digest, + "readiness_digests": authorization.readiness_digests, + "custody_log_digest": authorization.custody_log_digest, + "feasibility_digest": authorization.feasibility_digest, + } + + +def derive_admission_authorization_digest(authorization: AdmissionAuthorization) -> str: + return canonical.content_digest(_authorization_value(authorization)) + + +def _preparation_value(preparation: AdmissionPreparation) -> canonical.JsonValue: + return { + "schema_version": ADMISSION_PREPARATION_SCHEMA, + "authorization": _authorization_value(preparation.authorization), + "authorization_digest": preparation.authorization.authorization_digest, + "label_release_request_digest": derive_population_use_request_digest(preparation.label_release_request), + } + + +def derive_admission_preparation_digest(preparation: AdmissionPreparation) -> str: + return canonical.content_digest(_preparation_value(preparation)) + + +def _validate_authorization(authorization: AdmissionAuthorization) -> None: + problem = exact_dataclass_problem(authorization, AdmissionAuthorization, "admission authorization") + if problem is not None: + raise AdmissionSpendError(f"closed admission authorization violation: {problem}") + for field in ( + "manifest_digest", + "validation_context_digest", + "custody_log_digest", + "feasibility_digest", + "authorization_digest", + ): + _require_digest(getattr(authorization, field), field) + if type(authorization.readiness_digests) is not tuple: + raise AdmissionSpendError("readiness digests must be an exact tuple") + for digest in authorization.readiness_digests: + _require_digest(digest, "readiness digest") + if authorization.readiness_digests != tuple(sorted(set(authorization.readiness_digests))): + raise AdmissionSpendError("readiness digests must be unique and canonical") + if authorization.authorization_digest != derive_admission_authorization_digest(authorization): + raise AdmissionSpendError("admission authorization digest cannot be re-derived") + + +def _validate_preparation(preparation: AdmissionPreparation) -> None: + problem = exact_dataclass_problem(preparation, AdmissionPreparation, "admission preparation") + if problem is not None: + raise AdmissionSpendError(f"closed admission preparation violation: {problem}") + _validate_authorization(preparation.authorization) + request = preparation.label_release_request + if type(request) is not PopulationUseRequest: + raise AdmissionSpendError("label-release request must be exact") + authorization = preparation.authorization + expected = ( + LineagePurpose.ADMISSION_LABEL_OPEN, + True, + authorization.authorization_digest, + authorization.custody_log_digest, + ) + actual = (request.purpose, request.labels_accessed, request.influenced_artifact_digest, request.expected_log_digest) + if actual != expected: + raise AdmissionSpendError("label-release request does not match exact authorization") + _require_digest(preparation.preparation_digest, "preparation_digest") + if preparation.preparation_digest != derive_admission_preparation_digest(preparation): + raise AdmissionSpendError("admission preparation digest cannot be re-derived") + + +def _make_authorization( + manifest: AdmissionManifestV2, + context: AdmissionValidationContext, + custody_log_digest: str, + feasibility_digest: str, +) -> AdmissionAuthorization: + provisional = AdmissionAuthorization( + manifest.manifest_digest, + context.context_digest, + manifest.readiness_digests, + custody_log_digest, + feasibility_digest, + "0" * 64, + ) + return AdmissionAuthorization( + provisional.manifest_digest, + provisional.validation_context_digest, + provisional.readiness_digests, + provisional.custody_log_digest, + provisional.feasibility_digest, + derive_admission_authorization_digest(provisional), + ) + + +def _make_preparation(manifest: AdmissionManifestV2, authorization: AdmissionAuthorization) -> AdmissionPreparation: + request = PopulationUseRequest( + manifest.identity.population_digest, + manifest.identity.arm_id, + LineagePurpose.ADMISSION_LABEL_OPEN, + True, + authorization.authorization_digest, + authorization.custody_log_digest, + ) + provisional = AdmissionPreparation(authorization, request, "0" * 64) + return AdmissionPreparation(authorization, request, derive_admission_preparation_digest(provisional)) + + +def authorize_admission( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + readiness: Sequence[ReadinessResult], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> AdmissionPreparation: + """Revalidate label-blind inputs and return a pure head-bound release request.""" + try: + validate_admission_context(manifest, denominator, readiness, context, custody, resolver) + feasibility = evaluate_threshold_feasibility(manifest, denominator, context.admission_review_policy, resolver) + current = custody.current_log(manifest.identity.population_digest) + validate_population_custody_log(current, context.evidence.custody_policy, current=custody) + if current.log_digest != manifest.custody_log_digest or current.status is not PopulationStatus.AVAILABLE: + raise AdmissionSpendError("authorization custody head is stale or unavailable") + authorization = _make_authorization(manifest, context, current.log_digest, feasibility.feasibility_digest) + preparation = _make_preparation(manifest, authorization) + _validate_preparation(preparation) + return preparation + except AdmissionSpendError: + raise + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise AdmissionSpendError(f"admission authorization refused: {exc}") from exc + + +def _historical_log( + current: PopulationCustodyLog, + expected_digest: str, + policy: PopulationCustodyPolicy, +) -> PopulationCustodyLog: + historical = empty_population_custody_log(current.population_digest, current.custodian_identity) + if historical.log_digest == expected_digest: + return historical + for receipt in current.receipts: + historical = append_population_use(historical, receipt, policy) + if historical.log_digest == expected_digest: + return historical + raise AdmissionSpendError("authorization custody head is not in the current complete chain") + + +def _canonical_preparation( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + readiness: Sequence[ReadinessResult], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> tuple[AdmissionPreparation, PopulationCustodyLog]: + population = manifest.identity.population_digest + current = custody.current_log(population) + policy = context.evidence.custody_policy + validate_population_custody_log(current, policy, current=custody) + historical = _historical_log(current, manifest.custody_log_digest, policy) + view = _HistoricalCustody(custody, population, historical) + expected = authorize_admission(manifest, denominator, readiness, context, view, resolver) + return expected, current + + +def validate_pre_label_spend( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + readiness: Sequence[ReadinessResult], + context: AdmissionValidationContext, + preparation: AdmissionPreparation, + receipt: PopulationUseReceipt | None, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> AdmissionAuthorization: + """Authorize scoring only for the exact current AVAILABLE-to-SPENT successor.""" + try: + _validate_preparation(preparation) + expected, current = _canonical_preparation(manifest, denominator, readiness, context, custody, resolver) + if preparation != expected: + raise AdmissionSpendError("supplied value is not the canonical preparation") + if receipt is None: + raise AdmissionSpendError("label-release spend receipt is missing") + if type(receipt) is not PopulationUseReceipt: + raise AdmissionSpendError("label-release spend receipt must be exact") + if receipt.prior_status is not PopulationStatus.AVAILABLE or receipt.status is not PopulationStatus.SPENT: + raise AdmissionSpendError("label-release receipt must change available to spent") + if receipt.request != preparation.label_release_request: + raise AdmissionSpendError("label-release receipt does not match preparation") + is_head = bool(current.receipts) and current.receipts[-1] == receipt + if not is_head or current.receipt_count != receipt.sequence or current.status is not PopulationStatus.SPENT: + raise AdmissionSpendError("label-release receipt is not the current-chain direct successor") + if custody.current_log(current.population_digest).log_digest != current.log_digest: + raise AdmissionSpendError("label-release receipt is no longer the current-chain direct successor") + return preparation.authorization + except AdmissionSpendError: + raise + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise AdmissionSpendError(f"label-release spend refused: {exc}") from exc + + +__all__ = [ + "ADMISSION_AUTHORIZATION_SCHEMA", + "ADMISSION_PREPARATION_SCHEMA", + "AdmissionAuthorization", + "AdmissionPreparation", + "AdmissionSpendError", + "authorize_admission", + "derive_admission_authorization_digest", + "derive_admission_preparation_digest", + "validate_pre_label_spend", +] diff --git a/tests/test_core_admission_spend.py b/tests/test_core_admission_spend.py new file mode 100644 index 0000000..b395468 --- /dev/null +++ b/tests/test_core_admission_spend.py @@ -0,0 +1,305 @@ +"""Pre-label population spend protocol (Requirements 12.2 and 13.4).""" + +from __future__ import annotations + +from dataclasses import replace + +import pytest +from test_core_admission_context import AdmissionCase, _case, _resign_manifest + +from openultrasast.core import canonical, custody, validation +from openultrasast.core.admission_context import make_admission_validation_context +from openultrasast.core.admission_review import AdmissionThreshold, derive_threshold_set_digest +from openultrasast.core.admission_spend import AdmissionAuthorization, AdmissionPreparation +from openultrasast.core.lineage import LineageManifest, LineageUse, derive_lineage_manifest_digest +from openultrasast.core.measurement import Comparator, MetricId, Statistic + + +def _feasible_case() -> AdmissionCase: + case = _case() + threshold = AdmissionThreshold(MetricId.VULNERABLE_RECALL, Statistic.COUNT, Comparator.AT_LEAST, 1.0) + threshold_digest = derive_threshold_set_digest((threshold,)) + old_digest = case.manifest.threshold_lineage_manifest_digest + old_lineage = next(item for item in case.context.evidence.lineages if item.manifest_digest == old_digest) + population = old_lineage.uses[0].source_population_digest + current = case.port.current_log(population) + request = custody.PopulationUseRequest( + population, + case.manifest.identity.arm_id, + custody.LineagePurpose.THRESHOLD, + False, + threshold_digest, + current.log_digest, + ) + receipt = case.port.append(request, current.custodian_identity) + use = LineageUse(custody.LineagePurpose.THRESHOLD, population, False, threshold_digest, receipt.receipt_digest) + lineage = LineageManifest( + case.manifest.identity.arm_id, + threshold_digest, + (use,), + derive_lineage_manifest_digest(case.manifest.identity.arm_id, threshold_digest, (use,)), + ) + lineages = tuple(lineage if item.manifest_digest == old_digest else item for item in case.context.evidence.lineages) + evidence = validation.make_evidence_validation_context( + family_catalog=case.context.evidence.family_catalog, + fact_packs=case.context.evidence.fact_packs, + model_programs=case.context.evidence.model_programs, + lineages=lineages, + fact_review_policy=case.context.evidence.fact_review_policy, + provenance_trust_policy=case.context.evidence.provenance_trust_policy, + custody_policy=case.context.evidence.custody_policy, + ) + context = make_admission_validation_context( + evidence=evidence, + admission_review_policy=case.context.admission_review_policy, + proof_trust_policy=case.context.proof_trust_policy, + ) + identity = replace(case.manifest.identity, threshold_set_digest=threshold_digest) + manifest = _resign_manifest( + replace( + case.manifest, + identity=identity, + thresholds=(threshold,), + threshold_set_digest=threshold_digest, + threshold_lineage_manifest_digest=lineage.manifest_digest, + validation_context_digest=context.context_digest, + ), + case.manifest.review.conflict_disclosure, + ) + return replace(case, manifest=manifest, context=context) + + +def _prepare(case: AdmissionCase) -> AdmissionPreparation: + from openultrasast.core.admission_spend import authorize_admission + + return authorize_admission( + case.manifest, + case.denominator, + case.readiness, + case.context, + case.port, + case.resolver, + ) + + +def _spend(case: AdmissionCase, preparation: AdmissionPreparation) -> custody.PopulationUseReceipt: + current = case.port.current_log(preparation.label_release_request.population_digest) + receipt = custody.record_population_use( + preparation.label_release_request, + current, + current.custodian_identity, + case.context.evidence.custody_policy, + ) + case.port.logs[current.population_digest] = custody.append_population_use( + current, + receipt, + case.context.evidence.custody_policy, + ) + return receipt + + +def _validate_spend( + case: AdmissionCase, + preparation: AdmissionPreparation, + receipt: custody.PopulationUseReceipt | None, +) -> AdmissionAuthorization: + from openultrasast.core.admission_spend import validate_pre_label_spend + + return validate_pre_label_spend( + case.manifest, + case.denominator, + case.readiness, + case.context, + preparation, + receipt, + case.port, + case.resolver, + ) + + +def test_preparation_binds_exact_authorization_and_current_available_head_without_append() -> None: + case = _feasible_case() + population = case.manifest.identity.population_digest + before = case.port.current_log(population) + + preparation = _prepare(case) + + request = preparation.label_release_request + assert case.port.current_log(population) == before + assert preparation.authorization.manifest_digest == case.manifest.manifest_digest + assert preparation.authorization.validation_context_digest == case.context.context_digest + assert preparation.authorization.readiness_digests == case.manifest.readiness_digests + assert preparation.authorization.custody_log_digest == before.log_digest + assert request == custody.PopulationUseRequest( + population, + case.manifest.identity.arm_id, + custody.LineagePurpose.ADMISSION_LABEL_OPEN, + True, + preparation.authorization.authorization_digest, + before.log_digest, + ) + + +def test_exact_current_chain_available_to_spent_successor_authorizes_scoring() -> None: + case = _feasible_case() + preparation = _prepare(case) + receipt = _spend(case, preparation) + spent_log = case.port.current_log(case.manifest.identity.population_digest) + + authorization = _validate_spend(case, preparation, receipt) + + assert authorization == preparation.authorization + assert case.port.current_log(case.manifest.identity.population_digest) == spent_log + assert receipt.prior_status is custody.PopulationStatus.AVAILABLE + assert receipt.status is custody.PopulationStatus.SPENT + + +def test_substituted_authorization_or_release_request_is_refused() -> None: + from openultrasast.core.admission_spend import AdmissionSpendError + + case = _feasible_case() + preparation = _prepare(case) + receipt = _spend(case, preparation) + changed_authorization = replace( + preparation.authorization, + manifest_digest=canonical.content_digest("substituted-manifest"), + ) + changed_request = replace( + preparation.label_release_request, + influenced_artifact_digest=canonical.content_digest("substituted-authorization"), + ) + + with pytest.raises(AdmissionSpendError, match="authorization digest"): + _validate_spend(case, replace(preparation, authorization=changed_authorization), receipt) + with pytest.raises(AdmissionSpendError, match="exact authorization"): + _validate_spend(case, replace(preparation, label_release_request=changed_request), receipt) + + +@pytest.mark.parametrize("field", ("arm_id", "population_digest")) +def test_self_rehashed_request_cannot_replace_canonical_preparation(field: str) -> None: + from openultrasast.core.admission_spend import ( + AdmissionSpendError, + derive_admission_authorization_digest, + derive_admission_preparation_digest, + ) + + case = _feasible_case() + preparation = _prepare(case) + if field == "arm_id": + request = replace(preparation.label_release_request, arm_id="substituted-arm") + changed = replace(preparation, label_release_request=request, preparation_digest="0" * 64) + else: + population = case.context.evidence.lineages[0].uses[0].source_population_digest + wrong_log = case.port.current_log(population) + authorization = replace( + preparation.authorization, + custody_log_digest=wrong_log.log_digest, + authorization_digest="0" * 64, + ) + authorization = replace( + authorization, + authorization_digest=derive_admission_authorization_digest(authorization), + ) + request = replace( + preparation.label_release_request, + population_digest=population, + influenced_artifact_digest=authorization.authorization_digest, + expected_log_digest=wrong_log.log_digest, + ) + changed = replace( + preparation, + authorization=authorization, + label_release_request=request, + preparation_digest="0" * 64, + ) + changed = replace(changed, preparation_digest=derive_admission_preparation_digest(changed)) + receipt = _spend(case, changed) + spent_log = case.port.current_log(changed.label_release_request.population_digest) + + with pytest.raises(AdmissionSpendError, match="canonical preparation"): + _validate_spend(case, changed, receipt) + assert case.port.current_log(changed.label_release_request.population_digest) == spent_log + + +def test_missing_receipt_is_refused_while_population_remains_available() -> None: + from openultrasast.core.admission_spend import AdmissionSpendError + + case = _feasible_case() + preparation = _prepare(case) + + with pytest.raises(AdmissionSpendError, match="missing"): + _validate_spend(case, preparation, None) + + assert case.port.current_log(case.manifest.identity.population_digest).status is custody.PopulationStatus.AVAILABLE + + +def test_stale_or_mismatched_receipt_never_authorizes_scoring() -> None: + from openultrasast.core.admission_spend import AdmissionSpendError + + case = _feasible_case() + first = _prepare(case) + other_request = replace( + first.label_release_request, + influenced_artifact_digest=canonical.content_digest("different-authorization"), + ) + prior = case.port.current_log(case.manifest.identity.population_digest) + other_receipt = custody.record_population_use( + other_request, + prior, + prior.custodian_identity, + case.context.evidence.custody_policy, + ) + case.port.logs[prior.population_digest] = custody.append_population_use( + prior, + other_receipt, + case.context.evidence.custody_policy, + ) + + with pytest.raises(AdmissionSpendError, match="match preparation"): + _validate_spend(case, first, other_receipt) + + +def test_non_successor_and_availability_restoring_receipts_are_refused() -> None: + from openultrasast.core.admission_spend import AdmissionSpendError + + case = _feasible_case() + preparation = _prepare(case) + receipt = _spend(case, preparation) + non_successor = replace(receipt, sequence=receipt.sequence + 1) + restoring = replace(receipt, status=custody.PopulationStatus.AVAILABLE) + + with pytest.raises(AdmissionSpendError, match="current-chain direct successor"): + _validate_spend(case, preparation, non_successor) + with pytest.raises(AdmissionSpendError, match="available to spent"): + _validate_spend(case, preparation, restoring) + + +def test_preparation_is_stale_after_any_competing_current_head_change() -> None: + from openultrasast.core.admission_spend import AdmissionSpendError + + case = _feasible_case() + preparation = _prepare(case) + winner = _spend(case, preparation) + current = case.port.current_log(case.manifest.identity.population_digest) + later_request = custody.PopulationUseRequest( + current.population_digest, + "other-arm", + custody.LineagePurpose.PROMOTION, + False, + canonical.content_digest("later-artifact"), + current.log_digest, + ) + later = custody.record_population_use( + later_request, + current, + current.custodian_identity, + case.context.evidence.custody_policy, + ) + case.port.logs[current.population_digest] = custody.append_population_use( + current, + later, + case.context.evidence.custody_policy, + ) + + with pytest.raises(AdmissionSpendError, match="current-chain direct successor"): + _validate_spend(case, preparation, winner)