From 586543d7991a0980d13575dc8338eeccd9518aeb Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Wed, 4 Nov 2026 09:54:47 +0100 Subject: [PATCH] feat(detection-core): complete single-writer v2 cutover --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/__init__.py | 88 +--- .../core/_admission_case_shape.py | 2 +- src/openultrasast/core/_legacy_factpack.py | 19 - src/openultrasast/core/_proof_apply.py | 2 +- src/openultrasast/core/admission.py | 172 ++++++- src/openultrasast/core/admission_case.py | 2 +- src/openultrasast/core/admission_metrics.py | 2 +- src/openultrasast/core/consumer.py | 341 +++++++++++++- src/openultrasast/core/consumer_v2.py | 341 -------------- src/openultrasast/core/legacy_v1/__init__.py | 92 ---- src/openultrasast/core/legacy_v1/contracts.py | 439 ------------------ src/openultrasast/core/legacy_v1/writer.py | 46 -- src/openultrasast/core/proof.py | 2 +- src/openultrasast/core/trace.py | 5 - src/openultrasast/core/verdict.py | 272 ++++++++++- src/openultrasast/core/verdict_v2.py | 271 ----------- tests/core_migration_audit.py | 20 +- tests/core_migration_boundary.py | 24 +- tests/core_migration_inventory.py | 30 +- tests/core_migration_scope.py | 43 +- tests/test_core_admission_metrics.py | 2 +- tests/test_core_consumer_v2.py | 4 +- tests/test_core_cutover.py | 165 +++++++ tests/test_core_migration_boundary.py | 21 +- tests/test_core_migration_namespace.py | 112 ++--- tests/test_core_proof.py | 2 +- tests/test_core_verdict.py | 2 +- 28 files changed, 1106 insertions(+), 1417 deletions(-) delete mode 100644 src/openultrasast/core/_legacy_factpack.py delete mode 100644 src/openultrasast/core/consumer_v2.py delete mode 100644 src/openultrasast/core/legacy_v1/__init__.py delete mode 100644 src/openultrasast/core/legacy_v1/contracts.py delete mode 100644 src/openultrasast/core/legacy_v1/writer.py delete mode 100644 src/openultrasast/core/trace.py delete mode 100644 src/openultrasast/core/verdict_v2.py create mode 100644 tests/test_core_cutover.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index d78ef29..271d7ee 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -398,7 +398,7 @@ - _Requirements: 14.3, 15.3_ - _Boundary: Migration Tests and Operational Tooling_ - _Depends: 10.4, 10.5_ -- [ ] 10.7 Delete v1 write capability and publish the single v2 surface atomically +- [x] 10.7 Delete v1 write capability and publish the single v2 surface atomically - Require every inventory entry to be terminal and the migration-time production-consumer boundary to be zero before deletion begins. - Delete the compatibility namespace, v1 writer facade, and temporary shims while publishing the canonical v2 modules and top-level exports in the same integration change. - Prove old top-level and submodule imports and writes fail and that no v1 serializer or converter remains. diff --git a/src/openultrasast/core/__init__.py b/src/openultrasast/core/__init__.py index 35b5e35..b0227fe 100644 --- a/src/openultrasast/core/__init__.py +++ b/src/openultrasast/core/__init__.py @@ -1,87 +1,5 @@ -"""Detection core: the shared security-trace contract package. +"""Family-neutral detection-core v2 public surface.""" -Owns contracts only -- no analysis logic. Arms implement the trace contract; -runners and consumers adopt coverage, budget, provenance, and admission contracts. -""" +from openultrasast.core import admission, consumer, verdict -from openultrasast.core.legacy_v1 import ( - REPORT_SCHEMA_KEYS, - AdmissionManifest, - AdmissionManifestError, - BoundedSilence, - Budget, - BudgetGateFailure, - Complexity, - Coverage, - FactPack, - FactPackError, - FittedComponentBlocked, - Guard, - Hop, - InstrumentFailure, - Judgment, - ProbeResult, - ProvenanceMismatch, - ProvenClaim, - PushRequest, - PushVerdict, - ResolutionOutcome, - ShortcutDetected, - TraceRecord, - Witness, - admit_fact_pack, - as_clean, - assert_no_shortcut, - check_budget, - check_reads, - consume_verdict, - derive_budget, - require_admission_evidence, - reversed_pair_probe, - stamp, - validate_admission_manifest, - validate_report_schema, - verify_fact_pack, - verify_stamp, -) - -__all__ = [ - "REPORT_SCHEMA_KEYS", - "AdmissionManifest", - "AdmissionManifestError", - "BoundedSilence", - "Budget", - "BudgetGateFailure", - "Complexity", - "Coverage", - "FactPack", - "FactPackError", - "FittedComponentBlocked", - "Guard", - "Hop", - "InstrumentFailure", - "Judgment", - "ProbeResult", - "ProvenClaim", - "ProvenanceMismatch", - "PushRequest", - "PushVerdict", - "ResolutionOutcome", - "ShortcutDetected", - "TraceRecord", - "Witness", - "admit_fact_pack", - "as_clean", - "assert_no_shortcut", - "check_budget", - "check_reads", - "consume_verdict", - "derive_budget", - "require_admission_evidence", - "reversed_pair_probe", - "stamp", - "validate_admission_manifest", - "validate_report_schema", - "verify_fact_pack", - "verify_stamp", -] +__all__ = ["admission", "consumer", "verdict"] diff --git a/src/openultrasast/core/_admission_case_shape.py b/src/openultrasast/core/_admission_case_shape.py index d9bcf87..e3f2f88 100644 --- a/src/openultrasast/core/_admission_case_shape.py +++ b/src/openultrasast/core/_admission_case_shape.py @@ -59,7 +59,7 @@ from openultrasast.core.semantics import ( SourceLocation, ) from openultrasast.core.validation import EvidenceValidationInput -from openultrasast.core.verdict_v2 import BoundedSatisfaction, BoundedSilence, DefectClaim +from openultrasast.core.verdict import BoundedSatisfaction, BoundedSilence, DefectClaim _DATACLASSES: tuple[type[object], ...] = ( ScopeKey, diff --git a/src/openultrasast/core/_legacy_factpack.py b/src/openultrasast/core/_legacy_factpack.py deleted file mode 100644 index 6179801..0000000 --- a/src/openultrasast/core/_legacy_factpack.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Temporary explicit import shim for the pre-v2 fact-pack contract.""" - -from openultrasast.core.legacy_v1.contracts import ( - FactPackError, - LegacyFactPack, - admit_fact_pack, - admit_pack_file, - load_pack_file, - verify_fact_pack, -) - -__all__ = [ - "FactPackError", - "LegacyFactPack", - "admit_fact_pack", - "admit_pack_file", - "load_pack_file", - "verify_fact_pack", -] diff --git a/src/openultrasast/core/_proof_apply.py b/src/openultrasast/core/_proof_apply.py index da20095..9d23844 100644 --- a/src/openultrasast/core/_proof_apply.py +++ b/src/openultrasast/core/_proof_apply.py @@ -16,7 +16,7 @@ from openultrasast.core.proof import ( ProofVerificationReceipt, derive_producer_identity_digest, ) -from openultrasast.core.verdict_v2 import DefectClaim +from openultrasast.core.verdict import DefectClaim def _validate_claim_binding(claim: DefectClaim, attachment: ProofAttachment) -> None: diff --git a/src/openultrasast/core/admission.py b/src/openultrasast/core/admission.py index 09cacb8..3ff26b4 100644 --- a/src/openultrasast/core/admission.py +++ b/src/openultrasast/core/admission.py @@ -1,17 +1,165 @@ -"""Temporary explicit import shim for the pre-v2 admission contract.""" +"""Canonical v2 admission facade.""" -from openultrasast.core.legacy_v1.contracts import ( - REPORT_SCHEMA_KEYS, - AdmissionManifest, - AdmissionManifestError, - validate_admission_manifest, - validate_report_schema, +from openultrasast.core.admission_case import ( + ADMISSION_CASE_RELEASE_PAYLOAD_SCHEMA, + ADMISSION_CASE_RELEASE_SCHEMA, + RELEASED_CASE_BINDING_SCHEMA, + AdmissionCaseRelease, + AdmissionCaseReleaseError, + AdmissionCaseResult, + AdmissionCaseRole, + ReleasedCaseBinding, + create_admission_case_result, + derive_admission_case_release_digest, + derive_admission_case_release_payload_digest, + derive_admission_case_result_digest, + derive_case_coverage_digest, + derive_released_case_binding_digest, + validate_admission_case_release, + validate_admission_case_shape, +) +from openultrasast.core.admission_context import ( + ADMISSION_VALIDATION_CONTEXT_SCHEMA, + AdmissionContextError, + AdmissionValidationContext, + derive_admission_validation_context_digest, + derive_analysis_identity, + derive_fact_schema_digest, + derive_outcome_mapping_digest, + derive_semantic_vocabulary_digest, + make_admission_validation_context, + validate_admission_context, +) +from openultrasast.core.admission_feasibility import ( + AdmissionFeasibilityError, + FeasibleCountAssignment, + ThresholdFeasibility, + evaluate_threshold_feasibility, +) +from openultrasast.core.admission_fitted import ( + REPOSITORY_POPULATION_MANIFEST_SCHEMA, + FittedComponentAuthorizationError, + FittedComponentEvidence, + authorize_fitted_component, +) +from openultrasast.core.admission_metrics import ( + AdmissionMetricEvidence, + AdmissionMetricsError, + derive_admission_metrics, + derive_scoped_pair_key, +) +from openultrasast.core.admission_report import ( + ADMISSION_REPLAY_SCHEMA, + ADMISSION_REPORT_SCHEMA, + AdmissionDecision, + AdmissionReplay, + AdmissionReportError, + AdmissionReportV2, + CaseBudgetEvaluation, + GateKind, + GateResult, + ThresholdEvaluation, + admission_replay_value, + admission_report_value, + score_admission, + validate_admission_replay, +) +from openultrasast.core.admission_review import ( + AdmissionManifestV2, + AdmissionReviewError, + AdmissionThreshold, + PopulationCensus, + ScopeCensus, + derive_census_digest, + derive_census_payload_digest, + derive_manifest_digest, + derive_manifest_payload_digest, + derive_threshold_set_digest, + validate_reviewed_admission_payloads, +) +from openultrasast.core.admission_spend import ( + ADMISSION_AUTHORIZATION_SCHEMA, + ADMISSION_PREPARATION_SCHEMA, + AdmissionAuthorization, + AdmissionPreparation, + AdmissionSpendError, + authorize_admission, + derive_admission_authorization_digest, + derive_admission_preparation_digest, + validate_pre_label_spend, ) __all__ = [ - "REPORT_SCHEMA_KEYS", - "AdmissionManifest", - "AdmissionManifestError", - "validate_admission_manifest", - "validate_report_schema", + "ADMISSION_AUTHORIZATION_SCHEMA", + "ADMISSION_CASE_RELEASE_PAYLOAD_SCHEMA", + "ADMISSION_CASE_RELEASE_SCHEMA", + "ADMISSION_PREPARATION_SCHEMA", + "ADMISSION_REPLAY_SCHEMA", + "ADMISSION_REPORT_SCHEMA", + "ADMISSION_VALIDATION_CONTEXT_SCHEMA", + "RELEASED_CASE_BINDING_SCHEMA", + "REPOSITORY_POPULATION_MANIFEST_SCHEMA", + "AdmissionAuthorization", + "AdmissionCaseRelease", + "AdmissionCaseReleaseError", + "AdmissionCaseResult", + "AdmissionCaseRole", + "AdmissionContextError", + "AdmissionDecision", + "AdmissionFeasibilityError", + "AdmissionManifestV2", + "AdmissionMetricEvidence", + "AdmissionMetricsError", + "AdmissionPreparation", + "AdmissionReplay", + "AdmissionReportError", + "AdmissionReportV2", + "AdmissionReviewError", + "AdmissionSpendError", + "AdmissionThreshold", + "AdmissionValidationContext", + "CaseBudgetEvaluation", + "FeasibleCountAssignment", + "FittedComponentAuthorizationError", + "FittedComponentEvidence", + "GateKind", + "GateResult", + "PopulationCensus", + "ReleasedCaseBinding", + "ScopeCensus", + "ThresholdEvaluation", + "ThresholdFeasibility", + "admission_replay_value", + "admission_report_value", + "authorize_admission", + "authorize_fitted_component", + "create_admission_case_result", + "derive_admission_authorization_digest", + "derive_admission_case_release_digest", + "derive_admission_case_release_payload_digest", + "derive_admission_case_result_digest", + "derive_admission_preparation_digest", + "derive_admission_metrics", + "derive_admission_validation_context_digest", + "derive_analysis_identity", + "derive_case_coverage_digest", + "derive_census_digest", + "derive_census_payload_digest", + "derive_fact_schema_digest", + "derive_manifest_digest", + "derive_manifest_payload_digest", + "derive_outcome_mapping_digest", + "derive_released_case_binding_digest", + "derive_scoped_pair_key", + "derive_semantic_vocabulary_digest", + "derive_threshold_set_digest", + "evaluate_threshold_feasibility", + "make_admission_validation_context", + "score_admission", + "validate_admission_case_release", + "validate_admission_case_shape", + "validate_admission_context", + "validate_admission_replay", + "validate_pre_label_spend", + "validate_reviewed_admission_payloads", ] diff --git a/src/openultrasast/core/admission_case.py b/src/openultrasast/core/admission_case.py index 82ffc9d..e298742 100644 --- a/src/openultrasast/core/admission_case.py +++ b/src/openultrasast/core/admission_case.py @@ -19,7 +19,7 @@ from openultrasast.core.measurement import DenominatorPolicy, ScopeKey from openultrasast.core.proof import ProofAttachment from openultrasast.core.readiness import ReadinessResult from openultrasast.core.validation import EvidenceValidationInput -from openultrasast.core.verdict_v2 import BoundedSatisfaction, BoundedSilence, DefectClaim +from openultrasast.core.verdict import BoundedSatisfaction, BoundedSilence, DefectClaim ADMISSION_CASE_RELEASE_SCHEMA = "openultrasast.detection-core.admission-case-release.v2" ADMISSION_CASE_RELEASE_PAYLOAD_SCHEMA = "openultrasast.detection-core.admission-case-release-payload.v2" diff --git a/src/openultrasast/core/admission_metrics.py b/src/openultrasast/core/admission_metrics.py index a643916..5c4dd82 100644 --- a/src/openultrasast/core/admission_metrics.py +++ b/src/openultrasast/core/admission_metrics.py @@ -36,7 +36,7 @@ from openultrasast.core.model import ModelStatus from openultrasast.core.proof import ProofAttachment, attach_differential_proof from openultrasast.core.readiness import ReadinessResult from openultrasast.core.validation import validate_evidence_input -from openultrasast.core.verdict_v2 import DefectClaim, derive_verdict_outputs +from openultrasast.core.verdict import DefectClaim, derive_verdict_outputs class AdmissionMetricsError(ValueError): diff --git a/src/openultrasast/core/consumer.py b/src/openultrasast/core/consumer.py index cc7996c..a9849e6 100644 --- a/src/openultrasast/core/consumer.py +++ b/src/openultrasast/core/consumer.py @@ -1,5 +1,340 @@ -"""Temporary explicit import shim for the pre-v2 consumer contract.""" +"""Closed v2 push-consumer edge for validated bounded verdicts.""" -from openultrasast.core.legacy_v1.contracts import PushRequest, PushVerdict, consume_verdict +from __future__ import annotations -__all__ = ["PushRequest", "PushVerdict", "consume_verdict"] +import time +from collections.abc import Sequence +from dataclasses import dataclass, replace + +from openultrasast.core import canonical, verdict +from openultrasast.core.budget import BudgetDerivation, BudgetGateFailure, ChangeComplexity, check_budget +from openultrasast.core.evidence import AnalyzedScope, EvidenceRecord, SourceRegion, derive_packet_digest, validate_scope +from openultrasast.core.family import FamilyContract, family_contract_digest +from openultrasast.core.family_catalog import resolve_contract +from openultrasast.core.proof import ProofAttachment, ProofTrustPolicy, attach_differential_proof +from openultrasast.core.provenance import PopulationCustodyPort +from openultrasast.core.semantics import SourceLocation +from openultrasast.core.validation import EvidenceValidationContext, EvidenceValidationInput, validate_evidence_input +from openultrasast.core.verdict import ( + BoundedSatisfaction, + DefectClaim, + VerdictOutputs, + derive_verdict_outputs, +) + +PUSH_REQUEST_SCHEMA = "openultrasast.detection-core.push-request.v2" +CONSUMER_COVERAGE_SCHEMA = "openultrasast.detection-core.consumer-coverage.v2" +_monotonic_ns = time.monotonic_ns + + +class ConsumerContractError(ValueError): + """A raw push request cannot produce a bounded v2 verdict.""" + + +@dataclass(frozen=True) +class PushRequest: + request_id: str + scope: AnalyzedScope + workload_class: str + complexity: ChangeComplexity + budget: BudgetDerivation + + +@dataclass(frozen=True) +class PushVerdict: + request_digest: str + defects: tuple[DefectClaim, ...] + satisfactions: tuple[BoundedSatisfaction, ...] + silence: tuple[verdict.BoundedSilence, ...] + input_record_digests: tuple[str, ...] + complexity: ChangeComplexity + total_reply_ms: int + budget_derivation_digest: str + + +def _content_ref_value(value: canonical.ContentRef) -> canonical.JsonValue: + return { + "artifact_id": value.artifact_id, + "content_digest": value.content_digest, + "media_type": value.media_type, + "byte_length": value.byte_length, + "retention_policy_digest": value.retention_policy_digest, + "storage_receipt_digest": value.storage_receipt_digest, + } + + +def _scope_value(value: AnalyzedScope) -> canonical.JsonValue: + return { + "base_revision": value.base_revision, + "head_revision": value.head_revision, + "change_digest": value.change_digest, + "regions": tuple({"path": region.path, "start_line": region.start_line, "end_line": region.end_line} for region in value.regions), + } + + +def _request_value(request: PushRequest) -> canonical.JsonValue: + complexity = request.complexity + return { + "schema_version": PUSH_REQUEST_SCHEMA, + "scope": _scope_value(request.scope), + "workload_class": request.workload_class, + "complexity": { + "change_id": complexity.change_id, + "workload_class": complexity.workload_class, + "changed_regions": complexity.changed_regions, + "transition_count": complexity.transition_count, + "changed_bytes": complexity.changed_bytes, + "measurement": _content_ref_value(complexity.measurement), + }, + "budget_derivation_digest": request.budget.derivation_digest, + } + + +def derive_push_request_id(request: PushRequest) -> str: + if not isinstance(request, PushRequest): + raise ConsumerContractError("request must be a PushRequest") + return canonical.content_digest(_request_value(request)) + + +def make_push_request( + *, + scope: AnalyzedScope, + workload_class: str, + complexity: ChangeComplexity, + budget: BudgetDerivation, +) -> PushRequest: + provisional = PushRequest("0" * 64, scope, workload_class, complexity, budget) + return PushRequest(derive_push_request_id(provisional), scope, workload_class, complexity, budget) + + +def _validate_request(request: PushRequest) -> None: + if not isinstance(request, PushRequest): + raise ConsumerContractError("request must be a PushRequest") + try: + request.scope.__post_init__() + request.complexity.__post_init__() + request.budget.__post_init__() + except (AttributeError, TypeError, ValueError) as exc: + raise ConsumerContractError(f"push request is invalid: {exc}") from exc + if request.request_id != derive_push_request_id(request): + raise ConsumerContractError("push request identity cannot be re-derived") + if request.workload_class != request.complexity.workload_class or request.workload_class != request.budget.workload_class: + raise ConsumerContractError("push workload class does not match complexity and budget") + if request.complexity.change_id != request.scope.change_digest: + raise ConsumerContractError("complexity change identity does not match request scope") + if request.complexity.changed_regions != len(request.scope.regions): + raise ConsumerContractError("complexity changed-region count does not match request scope") + + +def _validated_records( + values: Sequence[EvidenceValidationInput], + context: EvidenceValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> tuple[EvidenceRecord, ...]: + selected = tuple(values) + if not selected: + raise ConsumerContractError("push inputs must be nonempty") + if not all(isinstance(value, EvidenceValidationInput) for value in selected): + raise ConsumerContractError("every push input must be an EvidenceValidationInput") + input_ids = tuple(value.input_digest for value in selected) + if len(set(input_ids)) != len(input_ids): + raise ConsumerContractError("duplicate validation input identity") + try: + return tuple(validate_evidence_input(value, context, custody, resolver) for value in selected) + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise ConsumerContractError(f"evidence validation failed: {exc}") from exc + + +def _contract(context: EvidenceValidationContext, record: EvidenceRecord) -> FamilyContract: + packet = record.packet + try: + return resolve_contract(context.family_catalog, packet.family_id, packet.family_contract_digest, packet.ecosystem) + except (LookupError, TypeError, ValueError) as exc: + raise ConsumerContractError(f"record family cannot be resolved: {exc}") from exc + + +def _coverage_digest(request: PushRequest, contract: FamilyContract, records: tuple[EvidenceRecord, ...]) -> str: + return canonical.content_digest( + { + "schema_version": CONSUMER_COVERAGE_SCHEMA, + "request_id": request.request_id, + "family_contract_digest": family_contract_digest(contract), + "record_ids": tuple(sorted(record.record_id for record in records)), + } + ) + + +def _derive_outputs( + request: PushRequest, + records: tuple[EvidenceRecord, ...], + context: EvidenceValidationContext, +) -> VerdictOutputs: + grouped: dict[str, tuple[FamilyContract, list[EvidenceRecord]]] = {} + for record in records: + try: + validate_scope(record.packet, request.scope) + except (TypeError, ValueError) as exc: + raise ConsumerContractError(f"record scope is outside the push request: {exc}") from exc + contract = _contract(context, record) + digest = record.packet.family_contract_digest + if digest not in grouped: + grouped[digest] = (contract, []) + grouped[digest][1].append(record) + defects: list[DefectClaim] = [] + satisfactions: list[BoundedSatisfaction] = [] + silence: list[verdict.BoundedSilence] = [] + try: + for digest in sorted(grouped): + contract, members = grouped[digest] + selected = tuple(members) + outputs = derive_verdict_outputs(selected, contract, _coverage_digest(request, contract, selected)) + defects.extend(outputs.defects) + satisfactions.extend(outputs.satisfactions) + silence.extend(outputs.silence) + except (TypeError, ValueError) as exc: + raise ConsumerContractError(f"bounded verdict derivation failed: {exc}") from exc + return VerdictOutputs(tuple(defects), tuple(satisfactions), tuple(silence)) + + +def _region_within(allowed: AnalyzedScope, candidate: SourceRegion) -> bool: + return any( + candidate.path == region.path and candidate.start_line >= region.start_line and candidate.end_line <= region.end_line + for region in allowed.regions + ) + + +def _proof_target_index(claims: tuple[DefectClaim, ...]) -> dict[tuple[str, str, SourceLocation], int]: + keys: dict[tuple[str, str, SourceLocation], int] = {} + for index, claim in enumerate(claims): + key = (claim.record.packet.family_id, derive_packet_digest(claim.record.packet), claim.location) + if key in keys: + raise ConsumerContractError("proof target is ambiguous across defect claims") + keys[key] = index + return keys + + +def _apply_attachment( + request: PushRequest, + result: list[DefectClaim], + keys: dict[tuple[str, str, SourceLocation], int], + consumed: set[int], + attachment: ProofAttachment, + trust: ProofTrustPolicy, + resolver: canonical.ContentResolver, +) -> None: + if not isinstance(attachment, ProofAttachment): + raise ConsumerContractError("every proof attachment must use the v2 proof contract") + key = (attachment.family_id, attachment.packet_digest, attachment.location) + if key not in keys: + raise ConsumerContractError("proof attachment does not match one exact defect claim") + index = keys[key] + if index in consumed: + raise ConsumerContractError("multiple proof attachments target one defect claim") + if not _region_within(request.scope, result[index].change_anchor): + raise ConsumerContractError("proof claim anchor is outside the push request scope") + result[index] = attach_differential_proof(result[index], attachment, trust, resolver) + consumed.add(index) + + +def _apply_proofs( + request: PushRequest, + claims: tuple[DefectClaim, ...], + attachments: Sequence[ProofAttachment], + trust: ProofTrustPolicy | None, + resolver: canonical.ContentResolver, +) -> tuple[DefectClaim, ...]: + selected = tuple(attachments) + if not selected: + return claims + if not isinstance(trust, ProofTrustPolicy): + raise ConsumerContractError("nonempty proof attachments require a proof trust policy") + keys = _proof_target_index(claims) + result = list(claims) + consumed: set[int] = set() + try: + for attachment in selected: + _apply_attachment(request, result, keys, consumed, attachment, trust, resolver) + except ConsumerContractError: + raise + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise ConsumerContractError(f"proof validation failed: {exc}") from exc + return tuple(result) + + +def _total_reply_ms(elapsed_before_entry_ms: int, started_ns: int) -> int: + if isinstance(elapsed_before_entry_ms, bool) or not isinstance(elapsed_before_entry_ms, int) or elapsed_before_entry_ms < 0: + raise ConsumerContractError("elapsed-before-entry measurement must be a non-negative integer") + finished_ns = _monotonic_ns() + ticks = (started_ns, finished_ns) + if any(isinstance(tick, bool) or not isinstance(tick, int) or tick < 0 for tick in ticks): + raise ConsumerContractError("monotonic core timing ticks must be non-negative integers") + if finished_ns < started_ns: + raise ConsumerContractError("monotonic core timing measurement moved backwards") + core_elapsed_ms = (finished_ns - started_ns + 999_999) // 1_000_000 + total_reply_ms = elapsed_before_entry_ms + core_elapsed_ms + if total_reply_ms <= 0: + raise ConsumerContractError("combined elapsed measurement must be a strictly positive integer") + return total_reply_ms + + +def _validate_budget_artifacts(request: PushRequest, resolver: canonical.ContentResolver) -> None: + try: + check_budget(request.complexity, 1, request.budget, resolver) + except (BudgetGateFailure, AttributeError, LookupError, TypeError, ValueError) as exc: + raise ConsumerContractError(f"measured budget validation failed: {exc}") from exc + + +def _enforce_validated_deadline(request: PushRequest, total_reply_ms: int) -> None: + if total_reply_ms > request.budget.budget_ms: + raise ConsumerContractError( + f"measured budget gate failed: budget:{request.complexity.change_id}:" + f"{total_reply_ms}ms>{request.budget.budget_ms}ms " + f"class={request.budget.workload_class} derivation={request.budget.derivation_digest}" + ) + + +def consume_records( + request: PushRequest, + inputs: Sequence[EvidenceValidationInput], + validation_context: EvidenceValidationContext, + custody: PopulationCustodyPort, + elapsed_before_entry_ms: int, + resolver: canonical.ContentResolver, + attachments: Sequence[ProofAttachment] = (), + proof_trust: ProofTrustPolicy | None = None, +) -> PushVerdict: + started_ns = _monotonic_ns() + _validate_request(request) + records = _validated_records(inputs, validation_context, custody, resolver) + transition_count = sum(len(record.packet.transitions) for record in records) + if request.complexity.transition_count != transition_count: + raise ConsumerContractError("complexity transition count does not match validated records") + outputs = _derive_outputs(request, records, validation_context) + defects = _apply_proofs(request, outputs.defects, attachments, proof_trust, resolver) + if not defects and not outputs.satisfactions and not outputs.silence: + raise ConsumerContractError("push verdict cannot be an unbounded clean result") + _validate_budget_artifacts(request, resolver) + provisional = PushVerdict( + request.request_id, + defects, + outputs.satisfactions, + outputs.silence, + tuple(sorted(record.record_id for record in records)), + request.complexity, + 0, + request.budget.derivation_digest, + ) + total_reply_ms = _total_reply_ms(elapsed_before_entry_ms, started_ns) + _enforce_validated_deadline(request, total_reply_ms) + return replace(provisional, total_reply_ms=total_reply_ms) + + +__all__ = [ + "ConsumerContractError", + "PushRequest", + "PushVerdict", + "consume_records", + "derive_push_request_id", + "make_push_request", +] diff --git a/src/openultrasast/core/consumer_v2.py b/src/openultrasast/core/consumer_v2.py deleted file mode 100644 index ae7be08..0000000 --- a/src/openultrasast/core/consumer_v2.py +++ /dev/null @@ -1,341 +0,0 @@ -"""Closed v2 push-consumer edge for validated bounded verdicts.""" - -from __future__ import annotations - -import time -from collections.abc import Sequence -from dataclasses import dataclass, replace - -from openultrasast.core import canonical -from openultrasast.core.budget import BudgetDerivation, BudgetGateFailure, ChangeComplexity, check_budget -from openultrasast.core.evidence import AnalyzedScope, EvidenceRecord, SourceRegion, derive_packet_digest, validate_scope -from openultrasast.core.family import FamilyContract, family_contract_digest -from openultrasast.core.family_catalog import resolve_contract -from openultrasast.core.proof import ProofAttachment, ProofTrustPolicy, attach_differential_proof -from openultrasast.core.provenance import PopulationCustodyPort -from openultrasast.core.semantics import SourceLocation -from openultrasast.core.validation import EvidenceValidationContext, EvidenceValidationInput, validate_evidence_input -from openultrasast.core.verdict_v2 import ( - BoundedSatisfaction, - BoundedSilence, - DefectClaim, - VerdictOutputs, - derive_verdict_outputs, -) - -PUSH_REQUEST_SCHEMA = "openultrasast.detection-core.push-request.v2" -CONSUMER_COVERAGE_SCHEMA = "openultrasast.detection-core.consumer-coverage.v2" -_monotonic_ns = time.monotonic_ns - - -class ConsumerContractError(ValueError): - """A raw push request cannot produce a bounded v2 verdict.""" - - -@dataclass(frozen=True) -class PushRequest: - request_id: str - scope: AnalyzedScope - workload_class: str - complexity: ChangeComplexity - budget: BudgetDerivation - - -@dataclass(frozen=True) -class PushVerdict: - request_digest: str - defects: tuple[DefectClaim, ...] - satisfactions: tuple[BoundedSatisfaction, ...] - silence: tuple[BoundedSilence, ...] - input_record_digests: tuple[str, ...] - complexity: ChangeComplexity - total_reply_ms: int - budget_derivation_digest: str - - -def _content_ref_value(value: canonical.ContentRef) -> canonical.JsonValue: - return { - "artifact_id": value.artifact_id, - "content_digest": value.content_digest, - "media_type": value.media_type, - "byte_length": value.byte_length, - "retention_policy_digest": value.retention_policy_digest, - "storage_receipt_digest": value.storage_receipt_digest, - } - - -def _scope_value(value: AnalyzedScope) -> canonical.JsonValue: - return { - "base_revision": value.base_revision, - "head_revision": value.head_revision, - "change_digest": value.change_digest, - "regions": tuple({"path": region.path, "start_line": region.start_line, "end_line": region.end_line} for region in value.regions), - } - - -def _request_value(request: PushRequest) -> canonical.JsonValue: - complexity = request.complexity - return { - "schema_version": PUSH_REQUEST_SCHEMA, - "scope": _scope_value(request.scope), - "workload_class": request.workload_class, - "complexity": { - "change_id": complexity.change_id, - "workload_class": complexity.workload_class, - "changed_regions": complexity.changed_regions, - "transition_count": complexity.transition_count, - "changed_bytes": complexity.changed_bytes, - "measurement": _content_ref_value(complexity.measurement), - }, - "budget_derivation_digest": request.budget.derivation_digest, - } - - -def derive_push_request_id(request: PushRequest) -> str: - if not isinstance(request, PushRequest): - raise ConsumerContractError("request must be a PushRequest") - return canonical.content_digest(_request_value(request)) - - -def make_push_request( - *, - scope: AnalyzedScope, - workload_class: str, - complexity: ChangeComplexity, - budget: BudgetDerivation, -) -> PushRequest: - provisional = PushRequest("0" * 64, scope, workload_class, complexity, budget) - return PushRequest(derive_push_request_id(provisional), scope, workload_class, complexity, budget) - - -def _validate_request(request: PushRequest) -> None: - if not isinstance(request, PushRequest): - raise ConsumerContractError("request must be a PushRequest") - try: - request.scope.__post_init__() - request.complexity.__post_init__() - request.budget.__post_init__() - except (AttributeError, TypeError, ValueError) as exc: - raise ConsumerContractError(f"push request is invalid: {exc}") from exc - if request.request_id != derive_push_request_id(request): - raise ConsumerContractError("push request identity cannot be re-derived") - if request.workload_class != request.complexity.workload_class or request.workload_class != request.budget.workload_class: - raise ConsumerContractError("push workload class does not match complexity and budget") - if request.complexity.change_id != request.scope.change_digest: - raise ConsumerContractError("complexity change identity does not match request scope") - if request.complexity.changed_regions != len(request.scope.regions): - raise ConsumerContractError("complexity changed-region count does not match request scope") - - -def _validated_records( - values: Sequence[EvidenceValidationInput], - context: EvidenceValidationContext, - custody: PopulationCustodyPort, - resolver: canonical.ContentResolver, -) -> tuple[EvidenceRecord, ...]: - selected = tuple(values) - if not selected: - raise ConsumerContractError("push inputs must be nonempty") - if not all(isinstance(value, EvidenceValidationInput) for value in selected): - raise ConsumerContractError("every push input must be an EvidenceValidationInput") - input_ids = tuple(value.input_digest for value in selected) - if len(set(input_ids)) != len(input_ids): - raise ConsumerContractError("duplicate validation input identity") - try: - return tuple(validate_evidence_input(value, context, custody, resolver) for value in selected) - except (AttributeError, LookupError, TypeError, ValueError) as exc: - raise ConsumerContractError(f"evidence validation failed: {exc}") from exc - - -def _contract(context: EvidenceValidationContext, record: EvidenceRecord) -> FamilyContract: - packet = record.packet - try: - return resolve_contract(context.family_catalog, packet.family_id, packet.family_contract_digest, packet.ecosystem) - except (LookupError, TypeError, ValueError) as exc: - raise ConsumerContractError(f"record family cannot be resolved: {exc}") from exc - - -def _coverage_digest(request: PushRequest, contract: FamilyContract, records: tuple[EvidenceRecord, ...]) -> str: - return canonical.content_digest( - { - "schema_version": CONSUMER_COVERAGE_SCHEMA, - "request_id": request.request_id, - "family_contract_digest": family_contract_digest(contract), - "record_ids": tuple(sorted(record.record_id for record in records)), - } - ) - - -def _derive_outputs( - request: PushRequest, - records: tuple[EvidenceRecord, ...], - context: EvidenceValidationContext, -) -> VerdictOutputs: - grouped: dict[str, tuple[FamilyContract, list[EvidenceRecord]]] = {} - for record in records: - try: - validate_scope(record.packet, request.scope) - except (TypeError, ValueError) as exc: - raise ConsumerContractError(f"record scope is outside the push request: {exc}") from exc - contract = _contract(context, record) - digest = record.packet.family_contract_digest - if digest not in grouped: - grouped[digest] = (contract, []) - grouped[digest][1].append(record) - defects: list[DefectClaim] = [] - satisfactions: list[BoundedSatisfaction] = [] - silence: list[BoundedSilence] = [] - try: - for digest in sorted(grouped): - contract, members = grouped[digest] - selected = tuple(members) - outputs = derive_verdict_outputs(selected, contract, _coverage_digest(request, contract, selected)) - defects.extend(outputs.defects) - satisfactions.extend(outputs.satisfactions) - silence.extend(outputs.silence) - except (TypeError, ValueError) as exc: - raise ConsumerContractError(f"bounded verdict derivation failed: {exc}") from exc - return VerdictOutputs(tuple(defects), tuple(satisfactions), tuple(silence)) - - -def _region_within(allowed: AnalyzedScope, candidate: SourceRegion) -> bool: - return any( - candidate.path == region.path and candidate.start_line >= region.start_line and candidate.end_line <= region.end_line - for region in allowed.regions - ) - - -def _proof_target_index(claims: tuple[DefectClaim, ...]) -> dict[tuple[str, str, SourceLocation], int]: - keys: dict[tuple[str, str, SourceLocation], int] = {} - for index, claim in enumerate(claims): - key = (claim.record.packet.family_id, derive_packet_digest(claim.record.packet), claim.location) - if key in keys: - raise ConsumerContractError("proof target is ambiguous across defect claims") - keys[key] = index - return keys - - -def _apply_attachment( - request: PushRequest, - result: list[DefectClaim], - keys: dict[tuple[str, str, SourceLocation], int], - consumed: set[int], - attachment: ProofAttachment, - trust: ProofTrustPolicy, - resolver: canonical.ContentResolver, -) -> None: - if not isinstance(attachment, ProofAttachment): - raise ConsumerContractError("every proof attachment must use the v2 proof contract") - key = (attachment.family_id, attachment.packet_digest, attachment.location) - if key not in keys: - raise ConsumerContractError("proof attachment does not match one exact defect claim") - index = keys[key] - if index in consumed: - raise ConsumerContractError("multiple proof attachments target one defect claim") - if not _region_within(request.scope, result[index].change_anchor): - raise ConsumerContractError("proof claim anchor is outside the push request scope") - result[index] = attach_differential_proof(result[index], attachment, trust, resolver) - consumed.add(index) - - -def _apply_proofs( - request: PushRequest, - claims: tuple[DefectClaim, ...], - attachments: Sequence[ProofAttachment], - trust: ProofTrustPolicy | None, - resolver: canonical.ContentResolver, -) -> tuple[DefectClaim, ...]: - selected = tuple(attachments) - if not selected: - return claims - if not isinstance(trust, ProofTrustPolicy): - raise ConsumerContractError("nonempty proof attachments require a proof trust policy") - keys = _proof_target_index(claims) - result = list(claims) - consumed: set[int] = set() - try: - for attachment in selected: - _apply_attachment(request, result, keys, consumed, attachment, trust, resolver) - except ConsumerContractError: - raise - except (AttributeError, LookupError, TypeError, ValueError) as exc: - raise ConsumerContractError(f"proof validation failed: {exc}") from exc - return tuple(result) - - -def _total_reply_ms(elapsed_before_entry_ms: int, started_ns: int) -> int: - if isinstance(elapsed_before_entry_ms, bool) or not isinstance(elapsed_before_entry_ms, int) or elapsed_before_entry_ms < 0: - raise ConsumerContractError("elapsed-before-entry measurement must be a non-negative integer") - finished_ns = _monotonic_ns() - ticks = (started_ns, finished_ns) - if any(isinstance(tick, bool) or not isinstance(tick, int) or tick < 0 for tick in ticks): - raise ConsumerContractError("monotonic core timing ticks must be non-negative integers") - if finished_ns < started_ns: - raise ConsumerContractError("monotonic core timing measurement moved backwards") - core_elapsed_ms = (finished_ns - started_ns + 999_999) // 1_000_000 - total_reply_ms = elapsed_before_entry_ms + core_elapsed_ms - if total_reply_ms <= 0: - raise ConsumerContractError("combined elapsed measurement must be a strictly positive integer") - return total_reply_ms - - -def _validate_budget_artifacts(request: PushRequest, resolver: canonical.ContentResolver) -> None: - try: - check_budget(request.complexity, 1, request.budget, resolver) - except (BudgetGateFailure, AttributeError, LookupError, TypeError, ValueError) as exc: - raise ConsumerContractError(f"measured budget validation failed: {exc}") from exc - - -def _enforce_validated_deadline(request: PushRequest, total_reply_ms: int) -> None: - if total_reply_ms > request.budget.budget_ms: - raise ConsumerContractError( - f"measured budget gate failed: budget:{request.complexity.change_id}:" - f"{total_reply_ms}ms>{request.budget.budget_ms}ms " - f"class={request.budget.workload_class} derivation={request.budget.derivation_digest}" - ) - - -def consume_records( - request: PushRequest, - inputs: Sequence[EvidenceValidationInput], - validation_context: EvidenceValidationContext, - custody: PopulationCustodyPort, - elapsed_before_entry_ms: int, - resolver: canonical.ContentResolver, - attachments: Sequence[ProofAttachment] = (), - proof_trust: ProofTrustPolicy | None = None, -) -> PushVerdict: - started_ns = _monotonic_ns() - _validate_request(request) - records = _validated_records(inputs, validation_context, custody, resolver) - transition_count = sum(len(record.packet.transitions) for record in records) - if request.complexity.transition_count != transition_count: - raise ConsumerContractError("complexity transition count does not match validated records") - outputs = _derive_outputs(request, records, validation_context) - defects = _apply_proofs(request, outputs.defects, attachments, proof_trust, resolver) - if not defects and not outputs.satisfactions and not outputs.silence: - raise ConsumerContractError("push verdict cannot be an unbounded clean result") - _validate_budget_artifacts(request, resolver) - provisional = PushVerdict( - request.request_id, - defects, - outputs.satisfactions, - outputs.silence, - tuple(sorted(record.record_id for record in records)), - request.complexity, - 0, - request.budget.derivation_digest, - ) - total_reply_ms = _total_reply_ms(elapsed_before_entry_ms, started_ns) - _enforce_validated_deadline(request, total_reply_ms) - return replace(provisional, total_reply_ms=total_reply_ms) - - -__all__ = [ - "ConsumerContractError", - "PushRequest", - "PushVerdict", - "consume_records", - "derive_push_request_id", - "make_push_request", -] diff --git a/src/openultrasast/core/legacy_v1/__init__.py b/src/openultrasast/core/legacy_v1/__init__.py deleted file mode 100644 index dc58bc9..0000000 --- a/src/openultrasast/core/legacy_v1/__init__.py +++ /dev/null @@ -1,92 +0,0 @@ -"""Explicit temporary exports for the closed detection-core v1 namespace.""" - -from openultrasast.core.legacy_v1.contracts import ( - REPORT_SCHEMA_KEYS, - AdmissionManifest, - AdmissionManifestError, - BoundedSilence, - Budget, - BudgetGateFailure, - Complexity, - Coverage, - FactPackError, - FittedComponentBlocked, - Guard, - Hop, - InstrumentFailure, - Judgment, - LegacyFactPack, - LegacyProbeResult, - ProvenanceMismatch, - ProvenClaim, - PushRequest, - PushVerdict, - ResolutionOutcome, - ShortcutDetected, - TraceRecord, - Witness, - admit_fact_pack, - admit_pack_file, - as_clean, - assert_no_shortcut, - check_budget, - check_reads, - consume_verdict, - derive_budget, - load_pack_file, - require_admission_evidence, - reversed_pair_probe, - stamp, - validate_admission_manifest, - validate_report_schema, - verify_fact_pack, - verify_stamp, -) - -FactPack = LegacyFactPack -ProbeResult = LegacyProbeResult - -__all__ = [ - "REPORT_SCHEMA_KEYS", - "AdmissionManifest", - "AdmissionManifestError", - "BoundedSilence", - "Budget", - "BudgetGateFailure", - "Complexity", - "Coverage", - "FactPack", - "FactPackError", - "FittedComponentBlocked", - "Guard", - "Hop", - "Judgment", - "LegacyFactPack", - "LegacyProbeResult", - "InstrumentFailure", - "ProvenClaim", - "PushRequest", - "PushVerdict", - "ProvenanceMismatch", - "ProbeResult", - "ResolutionOutcome", - "ShortcutDetected", - "TraceRecord", - "Witness", - "admit_fact_pack", - "admit_pack_file", - "as_clean", - "assert_no_shortcut", - "check_budget", - "check_reads", - "consume_verdict", - "derive_budget", - "load_pack_file", - "require_admission_evidence", - "reversed_pair_probe", - "stamp", - "validate_admission_manifest", - "validate_report_schema", - "verify_fact_pack", - "verify_stamp", -] diff --git a/src/openultrasast/core/legacy_v1/contracts.py b/src/openultrasast/core/legacy_v1/contracts.py deleted file mode 100644 index 5b02cf2..0000000 --- a/src/openultrasast/core/legacy_v1/contracts.py +++ /dev/null @@ -1,439 +0,0 @@ -"""Closed temporary owner of detection-core v1 contracts and behavior.""" - -from __future__ import annotations - -import enum -from collections.abc import Callable, Mapping, Sequence -from dataclasses import dataclass, field -from pathlib import Path -from typing import Any - - -class AdmissionManifestError(ValueError): - """A legacy admission manifest was refused.""" - - -class FactPackError(ValueError): - """A legacy fact pack is unreviewed or inconsistent.""" - - -class InstrumentFailure(ValueError): - """A legacy run did not read every tracked input.""" - - -class BudgetGateFailure(ValueError): - """A legacy change exceeded its measured budget.""" - - -class ProvenanceMismatch(ValueError): - """A legacy recorded digest differs from its re-derived identity.""" - - -class ShortcutDetected(ValueError): - """A legacy component separates reversed surface shapes.""" - - -class FittedComponentBlocked(ValueError): - """A legacy fitted component lacks release evidence.""" - - -class ResolutionOutcome(enum.Enum): - COMPLETE = "complete" - INCOMPLETE = "incomplete" - UNRESOLVED = "unresolved" - - -class Judgment(enum.Enum): - UNGUARDED = "unguarded" - DISCHARGED = "discharged" - INDETERMINATE = "indeterminate" - - -@dataclass(frozen=True) -class ProvenClaim: - trace_id: str - witness: str - path: str - line: int - family: str - - -@dataclass(frozen=True) -class BoundedSilence: - path: str - reason: str - coverage: dict[str, Any] - - -@dataclass(frozen=True) -class Hop: - function: str - call_line: int | None - complete: bool - - -@dataclass(frozen=True) -class Witness: - kind: str - provenance: str - relation: str - - -@dataclass(frozen=True) -class Guard: - kind: str - effect: str - - -@dataclass(frozen=True) -class TraceRecord: - entry: str - hops: tuple[Hop, ...] - operation_kind: str - resource: str - authority: str - witnesses: tuple[Witness, ...] = () - guards: tuple[Guard, ...] = () - completeness_reasons: tuple[str, ...] = () - judgment: Judgment = Judgment.INDETERMINATE - digests: dict[str, str] = field(default_factory=dict) - - def record(self) -> dict[str, Any]: - from openultrasast.core.legacy_v1 import writer - - return writer.trace_record(self) - - def canonical(self) -> bytes: - from openultrasast.core.legacy_v1 import writer - - return writer.trace_bytes(self) - - -@dataclass(frozen=True) -class PushRequest: - base: str - head: str - budget_class: str - - -@dataclass(frozen=True) -class PushVerdict: - claims: tuple[ProvenClaim, ...] - silence: tuple[BoundedSilence, ...] - digests: dict[str, str] - - -@dataclass(frozen=True) -class AdmissionManifest: - 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, ...] = () - - -@dataclass(frozen=True) -class LegacyFactPack: - family: str - facts: tuple[dict[str, Any], ...] - review_digest: str - fact_set_digest: str - - -@dataclass(frozen=True) -class Coverage: - entry_points: int - frameworks_recognised: dict[str, int] = field(default_factory=dict) - calls: int = 0 - complete: int = 0 - incomplete: int = 0 - unresolved: int = 0 - unchecked_reasons: dict[str, int] = field(default_factory=dict) - - def validate(self) -> None: - buckets = (self.complete, self.incomplete, self.unresolved) - if min(buckets) < 0: - raise ValueError(f"negative resolution bucket in {buckets}") - if sum(buckets) != self.calls: - raise ValueError( - f"resolution buckets ({self.complete}+{self.incomplete}+{self.unresolved}) must equal the call count ({self.calls})" - ) - - -@dataclass(frozen=True) -class Complexity: - regions: int - calls: int - changed_bytes: int - - -@dataclass(frozen=True) -class Budget: - workload_class: str - seconds: float - n: int - percentile: float - hardware: str - - def describe(self) -> str: - return f"class={self.workload_class} n={self.n} p{int(self.percentile * 100)} hardware={self.hardware} budget={self.seconds:g}s" - - -@dataclass(frozen=True) -class LegacyProbeResult: - separated: bool - fix_score: float - reversed_score: float - - -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", - } -) -_REQUIRED_MANIFEST_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" -_FORBIDDEN_LABEL_MARKERS = ("vulnerability", "fix-location", "ordinary-change") - - -def as_clean(judgment: Judgment) -> Judgment: - if judgment is Judgment.INDETERMINATE: - raise ValueError("indeterminate never converts to a clean judgment") - return judgment - - -def consume_verdict( - request: PushRequest, - record: TraceRecord, - claims: Sequence[ProvenClaim], - silence: Sequence[BoundedSilence], -) -> PushVerdict: - del request - for claim in claims: - if not claim.witness or not claim.path or claim.line <= 0: - raise ValueError(f"proven claim lacks witness or exact location: {claim}") - for entry in silence: - if not entry.reason: - raise ValueError(f"bounded silence lacks a coverage reason: {entry}") - return PushVerdict(tuple(claims), tuple(silence), dict(record.digests)) - - -def validate_admission_manifest( - payload: dict[str, object], - *, - expected_acquisition_digest: str | None = None, -) -> AdmissionManifest: - if not all(field in payload for field in _REQUIRED_MANIFEST_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)): - raise AdmissionManifestError("label-exposure: manifest carries admission labels") - if 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 _manifest(payload, raw_labels, raw_families) - - -def _manifest(payload: dict[str, object], labels: Sequence[object], families: Sequence[object]) -> AdmissionManifest: - 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 labels), - families=tuple(str(family) for family in families), - ) - - -def validate_report_schema(report: dict[str, Any]) -> None: - missing = sorted(REPORT_SCHEMA_KEYS - set(report)) - if missing: - raise AdmissionManifestError(f"report schema: missing keys {missing}") - for rate_key in ( - "pair_correctness", - "fix_recall", - "verified_location_recall", - "flagged_ordinary_per_100", - "precision", - "indeterminate_rate", - ): - if not {"n", "count", "rate", "wilson95"} <= set(report[rate_key]): - raise AdmissionManifestError(f"report schema: {rate_key} lacks n/count/rate/wilson95") - - -def admit_fact_pack(family: str, facts: Sequence[Mapping[str, Any]], *, review_digest: str) -> LegacyFactPack: - from openultrasast.core.legacy_v1 import writer - - if not review_digest: - raise FactPackError("unreviewed fact pack refused: no review digest") - normalized = tuple(dict(fact) for fact in facts) - return LegacyFactPack(family, normalized, review_digest, writer.fact_pack_digest(family, normalized)) - - -def verify_fact_pack(pack: LegacyFactPack, facts: Sequence[Mapping[str, Any]]) -> None: - from openultrasast.core.legacy_v1 import writer - - normalized = tuple(dict(fact) for fact in facts) - if writer.fact_pack_digest(pack.family, normalized) != pack.fact_set_digest: - raise FactPackError(f"fact content does not match the stamped digest of pack {pack.family!r}") - - -def load_pack_file(path: str | Path) -> LegacyFactPack: - from openultrasast.core.legacy_v1 import writer - - payload = writer.read_json(Path(path)) - review = payload.get("review", {}) - if not isinstance(review, dict): - raise FactPackError(f"pack {path} lacks a verifiable review attestation") - note = review.get("note", "") - if not isinstance(note, str) or not note or review.get("digest") != writer.text_digest(note): - raise FactPackError(f"pack {path} lacks a verifiable review attestation") - facts = payload.get("facts") - if not isinstance(facts, list) or not all(isinstance(fact, dict) for fact in facts): - raise FactPackError(f"pack {path}: facts must be a list of objects") - return admit_fact_pack(str(payload.get("family", "")), facts, review_digest=str(review["digest"])) - - -def admit_pack_file(path: str | Path, *, review_digest: str) -> LegacyFactPack: - pack = load_pack_file(path) - if review_digest != pack.review_digest: - raise FactPackError(f"pack {path}: supplied review digest does not match the embedded attestation") - return pack - - -_STAMP_FIELDS = ( - "analyzer_revision", - "fact_set_digest", - "acquisition_record_digest", - "configuration_digest", - "population_digest", - "input_digest", -) - - -def check_reads(reads: Mapping[str, int]) -> None: - zero = sorted(path for path, size in reads.items() if size <= 0) - if zero: - raise InstrumentFailure(f"zero-byte read of tracked input: {', '.join(zero)}") - - -def _quantile(values: list[float], percentile: float) -> float: - position = (len(values) - 1) * percentile - lower = int(position) - upper = min(lower + 1, len(values) - 1) - weight = position - lower - return values[lower] * (1 - weight) + values[upper] * weight - - -def derive_budget(samples: Sequence[tuple[Complexity, float]], *, workload_class: str, percentile: float, hardware: str) -> Budget: - if not samples: - raise ValueError("budget derivation needs at least one measured sample") - if not 0.0 < percentile < 1.0: - raise ValueError(f"percentile {percentile} outside (0, 1)") - seconds = sorted(float(item[1]) for item in samples) - return Budget(workload_class, _quantile(seconds, percentile), len(samples), percentile, hardware) - - -def check_budget(change_id: str, seconds: float, budget: Budget) -> None: - if seconds > budget.seconds: - raise BudgetGateFailure(f"budget:{change_id}:{seconds:g}s>{budget.seconds:g}s ({budget.describe()})") - - -def stamp( - *, - analyzer_revision: str, - fact_set_digest: str, - acquisition_record_digest: str, - configuration_digest: str, - population_digest: str, - change_id: str, -) -> dict[str, str]: - from openultrasast.core.legacy_v1 import writer - - return { - "analyzer_revision": analyzer_revision, - "fact_set_digest": fact_set_digest, - "acquisition_record_digest": acquisition_record_digest, - "configuration_digest": configuration_digest, - "population_digest": population_digest, - "input_digest": writer.input_digest(change_id, population_digest), - } - - -def verify_stamp(recorded: Mapping[str, object], rederived: Mapping[str, object]) -> None: - for field_name in _STAMP_FIELDS: - if recorded.get(field_name) != rederived.get(field_name): - raise ProvenanceMismatch(f"{field_name}: recorded {recorded.get(field_name)!r} != rederived {rederived.get(field_name)!r}") - - -def reversed_pair_probe( - score_fn: Callable[[Mapping[str, Any]], float], fix_change: Mapping[str, Any], reversed_change: Mapping[str, Any] -) -> LegacyProbeResult: - fix_score = float(score_fn(fix_change)) - reversed_score = float(score_fn(reversed_change)) - return LegacyProbeResult(fix_score != reversed_score, fix_score, reversed_score) - - -def assert_no_shortcut( - score_fn: Callable[[Mapping[str, Any]], float], fix_change: Mapping[str, Any], reversed_change: Mapping[str, Any] -) -> None: - result = reversed_pair_probe(score_fn, fix_change, reversed_change) - if result.separated: - raise ShortcutDetected(f"shape shortcut: fix={result.fix_score} vs reversed={result.reversed_score}") - - -def require_admission_evidence(*, holdout_passed: bool, admission_decision: str) -> None: - if not holdout_passed: - raise FittedComponentBlocked("fitted component lacks per-repository holdout evidence") - if admission_decision != "GO": - raise FittedComponentBlocked(f"fitted component admission decision is {admission_decision!r}, not GO") diff --git a/src/openultrasast/core/legacy_v1/writer.py b/src/openultrasast/core/legacy_v1/writer.py deleted file mode 100644 index e8c5fae..0000000 --- a/src/openultrasast/core/legacy_v1/writer.py +++ /dev/null @@ -1,46 +0,0 @@ -"""The sole temporary detection-core v1 JSON serialization capability.""" - -from __future__ import annotations - -import hashlib -import json -from collections.abc import Mapping, Sequence -from dataclasses import asdict -from pathlib import Path -from typing import Any, cast - -from openultrasast.core.legacy_v1 import contracts - - -def trace_record(value: contracts.TraceRecord) -> dict[str, Any]: - payload: dict[str, Any] = asdict(value) - payload["judgment"] = value.judgment.value - return cast(dict[str, Any], json.loads(json.dumps(payload, sort_keys=True))) - - -def trace_bytes(value: contracts.TraceRecord) -> bytes: - return json.dumps(trace_record(value), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() - - -def fact_pack_digest(family: str, facts: Sequence[Mapping[str, Any]]) -> str: - encoded = json.dumps([family, tuple(dict(fact) for fact in facts)], sort_keys=True, separators=(",", ":")).encode() - return hashlib.sha256(encoded).hexdigest()[:12] - - -def text_digest(value: str) -> str: - return hashlib.sha256(value.encode()).hexdigest() - - -def input_digest(change_id: str, population_digest: str) -> str: - payload = json.dumps([population_digest, change_id], sort_keys=True, separators=(",", ":")).encode() - return hashlib.sha256(payload).hexdigest()[:12] - - -def read_json(path: Path) -> dict[str, Any]: - value = json.loads(path.read_text(encoding="utf-8")) - if not isinstance(value, dict): - raise contracts.FactPackError(f"pack {path}: root must be an object") - return value - - -__all__ = ["fact_pack_digest", "input_digest", "read_json", "text_digest", "trace_bytes", "trace_record"] diff --git a/src/openultrasast/core/proof.py b/src/openultrasast/core/proof.py index f996c86..ef7582a 100644 --- a/src/openultrasast/core/proof.py +++ b/src/openultrasast/core/proof.py @@ -8,7 +8,7 @@ from enum import Enum from openultrasast.core import canonical from openultrasast.core.semantics import SourceLocation -from openultrasast.core.verdict_v2 import DefectClaim +from openultrasast.core.verdict import DefectClaim DIFFERENTIAL_PROOF_SCHEMA = "openultrasast.detection-core.differential-proof.v2" _DIGEST = re.compile(r"[0-9a-f]{64}") diff --git a/src/openultrasast/core/trace.py b/src/openultrasast/core/trace.py deleted file mode 100644 index 07ed8db..0000000 --- a/src/openultrasast/core/trace.py +++ /dev/null @@ -1,5 +0,0 @@ -"""Temporary explicit import shim for the pre-v2 trace contract.""" - -from openultrasast.core.legacy_v1.contracts import Guard, Hop, TraceRecord, Witness - -__all__ = ["Guard", "Hop", "TraceRecord", "Witness"] diff --git a/src/openultrasast/core/verdict.py b/src/openultrasast/core/verdict.py index 30f3e27..1db97fb 100644 --- a/src/openultrasast/core/verdict.py +++ b/src/openultrasast/core/verdict.py @@ -1,5 +1,271 @@ -"""Temporary explicit import shim for the pre-v2 verdict contract.""" +"""Bounded v2 verdict outputs derived from complete evidence records.""" -from openultrasast.core.legacy_v1.contracts import BoundedSilence, Judgment, ProvenClaim, as_clean +from __future__ import annotations -__all__ = ["BoundedSilence", "Judgment", "ProvenClaim", "as_clean"] +import re +from collections.abc import Sequence +from dataclasses import dataclass + +from openultrasast.core.evidence import ( + AnalyzedScope, + EvidenceBasis, + EvidenceRecord, + EvidenceValidationError, + NeutralOutcome, + ScopeValidationError, + SourceRegion, + derive_record_id, + validate_scope, +) +from openultrasast.core.family import FamilyContract, FamilyContractError, ResidualQuestion, validate_contract, validate_record +from openultrasast.core.semantics import Observation, RelationStatus, SemanticRelation, SourceLocation + +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class VerdictContractError(ValueError): + """Validated evidence cannot be represented as a bounded verdict.""" + + +@dataclass(frozen=True) +class DefectClaim: + record: EvidenceRecord + basis: EvidenceBasis + location: SourceLocation + change_anchor: SourceRegion + + def __post_init__(self) -> None: + if not isinstance(self.record, EvidenceRecord) or self.record.decision.outcome is not NeutralOutcome.VIOLATION: + raise VerdictContractError("defect claim requires a violation evidence record") + try: + self.record.__post_init__() + validate_scope(self.record.packet, self.record.packet.scope) + except (TypeError, ValueError) as exc: + raise VerdictContractError(f"defect claim evidence record is invalid: {exc}") from exc + if not isinstance(self.basis, EvidenceBasis) or self.basis is not self.record.decision.basis: + raise VerdictContractError("defect claim basis must equal its evidence decision basis") + if not isinstance(self.location, SourceLocation) or not isinstance(self.change_anchor, SourceRegion): + raise VerdictContractError("defect claim requires an exact location and change anchor") + members: tuple[SemanticRelation | Observation, ...] = ( + *self.record.packet.transitions, + *self.record.packet.observations, + ) + if self.record.packet.completeness_reasons or any(item.status is not RelationStatus.COMPLETE for item in members): + raise VerdictContractError("defect claim requires complete evidence") + if self.location not in tuple(item.location for item in members): + raise VerdictContractError("defect location must name an exact evidence location") + if self.change_anchor not in self.record.decision.bounded_scope.regions: + raise VerdictContractError("defect change anchor must belong to the bounded decision scope") + if not _region_contains(self.change_anchor, self.location): + raise VerdictContractError("defect location is outside its change anchor") + + +@dataclass(frozen=True) +class BoundedSatisfaction: + question_ids: tuple[str, ...] + records: tuple[EvidenceRecord, ...] + scope: AnalyzedScope + + def __post_init__(self) -> None: + _text_tuple(self.question_ids, "satisfaction question") + if ( + not isinstance(self.records, tuple) + or not self.records + or not all(isinstance(record, EvidenceRecord) for record in self.records) + ): + raise VerdictContractError("bounded satisfaction requires evidence records") + if not isinstance(self.scope, AnalyzedScope): + raise VerdictContractError("bounded satisfaction requires an AnalyzedScope") + if any(record.decision.outcome is not NeutralOutcome.SATISFIED for record in self.records): + raise VerdictContractError("bounded satisfaction accepts only satisfied evidence records") + if any(record.decision.bounded_scope != self.scope for record in self.records): + raise VerdictContractError("satisfaction scope must equal every supporting record scope") + + +@dataclass(frozen=True) +class BoundedSilence: + question_ids: tuple[str, ...] + reasons: tuple[str, ...] + evidence_record_digests: tuple[str, ...] + coverage_digest: str + + def __post_init__(self) -> None: + _text_tuple(self.question_ids, "silence question") + _text_tuple(self.reasons, "silence reason") + _digest_tuple(self.evidence_record_digests, "silence evidence record") + _digest(self.coverage_digest, "coverage") + + +@dataclass(frozen=True) +class VerdictOutputs: + defects: tuple[DefectClaim, ...] + satisfactions: tuple[BoundedSatisfaction, ...] + silence: tuple[BoundedSilence, ...] + + def __post_init__(self) -> None: + values = ( + (self.defects, DefectClaim, "defects"), + (self.satisfactions, BoundedSatisfaction, "satisfactions"), + (self.silence, BoundedSilence, "silence"), + ) + for items, item_type, field in values: + if not isinstance(items, tuple) or not all(isinstance(item, item_type) for item in items): + raise VerdictContractError(f"{field} must be a tuple of {item_type.__name__}") + + +def _digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise VerdictContractError(f"{field} must be a full lowercase SHA-256 digest") + + +def _text_tuple(value: object, field: str) -> None: + if not isinstance(value, tuple) or not value or not all(isinstance(item, str) and item for item in value): + raise VerdictContractError(f"{field} identities must be a nonempty tuple") + if len(set(value)) != len(value): + raise VerdictContractError(f"duplicate {field} identity") + + +def _digest_tuple(value: object, field: str) -> None: + _text_tuple(value, field) + assert isinstance(value, tuple) + for item in value: + _digest(item, field) + + +def _region_contains(region: SourceRegion, location: SourceLocation) -> bool: + return location.path == region.path and region.start_line <= location.line <= region.end_line + + +def _scope_contains(scope: AnalyzedScope, location: SourceLocation) -> bool: + return any(_region_contains(region, location) for region in scope.regions) + + +def _members(record: EvidenceRecord, role: str) -> tuple[SemanticRelation | Observation, ...]: + transitions = tuple(item for item in record.packet.transitions if item.role == role) + observations = tuple(item for item in record.packet.observations if item.role == role) + return (*transitions, *observations) + + +def _question_state(record: EvidenceRecord, question: ResidualQuestion) -> tuple[bool, tuple[str, ...]]: + reasons: set[str] = set() + if record.decision.family_code not in question.allowed_decisions: + reasons.add(f"decision_not_declared_for_question:{record.decision.family_code}") + for role in question.required_roles: + members = _members(record, role) + if not members: + reasons.add(f"missing_required_role:{role}") + elif not any( + item.status is RelationStatus.COMPLETE and _scope_contains(record.decision.bounded_scope, item.location) for item in members + ): + relation_reasons = {reason for item in members if isinstance(item, SemanticRelation) for reason in item.reasons} + complete_outside = any(item.status is RelationStatus.COMPLETE for item in members) + fallback = f"evidence_outside_bounded_scope:{role}" if complete_outside else f"incomplete_required_role:{role}" + reasons.update(relation_reasons or record.packet.completeness_reasons or (fallback,)) + if record.decision.outcome is NeutralOutcome.INDETERMINATE: + reasons.update(record.decision.reasons) + complete = not reasons and record.decision.outcome is not NeutralOutcome.INDETERMINATE + return complete, tuple(sorted(reasons)) + + +def _location(record: EvidenceRecord, question: ResidualQuestion) -> tuple[SourceLocation, SourceRegion]: + candidates = tuple( + item.location for role in question.required_roles for item in _members(record, role) if item.status is RelationStatus.COMPLETE + ) + ordered = sorted(candidates, key=lambda item: (item.path, item.line, -1 if item.column is None else item.column, item.symbol or "")) + for location in ordered: + anchors = tuple(region for region in record.decision.bounded_scope.regions if _region_contains(region, location)) + if anchors: + anchor = min(anchors, key=lambda item: (item.end_line - item.start_line, item.path, item.start_line)) + return location, anchor + raise VerdictContractError("complete defect evidence has no exact location inside its bounded change scope") + + +def _validate_inputs(records: Sequence[EvidenceRecord], contract: FamilyContract, coverage_digest: str) -> tuple[EvidenceRecord, ...]: + _digest(coverage_digest, "coverage") + try: + validate_contract(contract) + except (TypeError, ValueError) as exc: + raise VerdictContractError(f"family contract is invalid: {exc}") from exc + values = tuple(records) + if not values or not all(isinstance(record, EvidenceRecord) for record in values): + raise VerdictContractError("records must be a nonempty sequence of EvidenceRecord values") + identities: set[str] = set() + try: + for record in values: + record.__post_init__() + if record.record_id != derive_record_id(record.packet, record.decision): + raise EvidenceValidationError("record identity cannot be re-derived") + if record.record_id in identities: + raise EvidenceValidationError("duplicate evidence record identity") + identities.add(record.record_id) + validate_scope(record.packet, record.packet.scope) + validate_record(contract, record) + except (TypeError, ValueError, EvidenceValidationError, ScopeValidationError, FamilyContractError) as exc: + raise VerdictContractError(f"evidence record validation failed: {exc}") from exc + return values + + +def _silence( + question: ResidualQuestion, + reasons: set[str], + record_ids: set[str], + coverage_digest: str, +) -> BoundedSilence | None: + if not reasons: + return None + return BoundedSilence( + (question.question_id,), + tuple(sorted(reasons)), + tuple(sorted(record_ids)), + coverage_digest, + ) + + +def derive_verdict_outputs( + records: Sequence[EvidenceRecord], + contract: FamilyContract, + coverage_digest: str, +) -> VerdictOutputs: + """Derive bounded outputs without converting incomplete evidence to safety.""" + validated = _validate_inputs(records, contract, coverage_digest) + defects: dict[str, DefectClaim] = {} + satisfactions: dict[str, set[str]] = {} + silence: list[BoundedSilence] = [] + for question in contract.residual_questions: + reasons: set[str] = set() + undecided: set[str] = set() + for record in validated: + complete, record_reasons = _question_state(record, question) + if not complete: + reasons.update(record_reasons) + undecided.add(record.record_id) + continue + if record.decision.outcome is NeutralOutcome.VIOLATION: + if record.record_id not in defects: + location, anchor = _location(record, question) + defects[record.record_id] = DefectClaim(record, record.decision.basis, location, anchor) + elif record.decision.outcome is NeutralOutcome.SATISFIED: + satisfactions.setdefault(record.record_id, set()).add(question.question_id) + bounded = _silence(question, reasons, undecided, coverage_digest) + if bounded is not None: + silence.append(bounded) + records_by_id = {record.record_id: record for record in validated} + satisfied = tuple( + BoundedSatisfaction(tuple(sorted(question_ids)), (records_by_id[record_id],), records_by_id[record_id].decision.bounded_scope) + for record_id, question_ids in sorted(satisfactions.items()) + ) + return VerdictOutputs( + tuple(defects[identity] for identity in sorted(defects)), + satisfied, + tuple(sorted(silence, key=lambda item: item.question_ids)), + ) + + +__all__ = [ + "BoundedSatisfaction", + "BoundedSilence", + "DefectClaim", + "VerdictContractError", + "VerdictOutputs", + "derive_verdict_outputs", +] diff --git a/src/openultrasast/core/verdict_v2.py b/src/openultrasast/core/verdict_v2.py deleted file mode 100644 index 1db97fb..0000000 --- a/src/openultrasast/core/verdict_v2.py +++ /dev/null @@ -1,271 +0,0 @@ -"""Bounded v2 verdict outputs derived from complete evidence records.""" - -from __future__ import annotations - -import re -from collections.abc import Sequence -from dataclasses import dataclass - -from openultrasast.core.evidence import ( - AnalyzedScope, - EvidenceBasis, - EvidenceRecord, - EvidenceValidationError, - NeutralOutcome, - ScopeValidationError, - SourceRegion, - derive_record_id, - validate_scope, -) -from openultrasast.core.family import FamilyContract, FamilyContractError, ResidualQuestion, validate_contract, validate_record -from openultrasast.core.semantics import Observation, RelationStatus, SemanticRelation, SourceLocation - -_DIGEST = re.compile(r"[0-9a-f]{64}") - - -class VerdictContractError(ValueError): - """Validated evidence cannot be represented as a bounded verdict.""" - - -@dataclass(frozen=True) -class DefectClaim: - record: EvidenceRecord - basis: EvidenceBasis - location: SourceLocation - change_anchor: SourceRegion - - def __post_init__(self) -> None: - if not isinstance(self.record, EvidenceRecord) or self.record.decision.outcome is not NeutralOutcome.VIOLATION: - raise VerdictContractError("defect claim requires a violation evidence record") - try: - self.record.__post_init__() - validate_scope(self.record.packet, self.record.packet.scope) - except (TypeError, ValueError) as exc: - raise VerdictContractError(f"defect claim evidence record is invalid: {exc}") from exc - if not isinstance(self.basis, EvidenceBasis) or self.basis is not self.record.decision.basis: - raise VerdictContractError("defect claim basis must equal its evidence decision basis") - if not isinstance(self.location, SourceLocation) or not isinstance(self.change_anchor, SourceRegion): - raise VerdictContractError("defect claim requires an exact location and change anchor") - members: tuple[SemanticRelation | Observation, ...] = ( - *self.record.packet.transitions, - *self.record.packet.observations, - ) - if self.record.packet.completeness_reasons or any(item.status is not RelationStatus.COMPLETE for item in members): - raise VerdictContractError("defect claim requires complete evidence") - if self.location not in tuple(item.location for item in members): - raise VerdictContractError("defect location must name an exact evidence location") - if self.change_anchor not in self.record.decision.bounded_scope.regions: - raise VerdictContractError("defect change anchor must belong to the bounded decision scope") - if not _region_contains(self.change_anchor, self.location): - raise VerdictContractError("defect location is outside its change anchor") - - -@dataclass(frozen=True) -class BoundedSatisfaction: - question_ids: tuple[str, ...] - records: tuple[EvidenceRecord, ...] - scope: AnalyzedScope - - def __post_init__(self) -> None: - _text_tuple(self.question_ids, "satisfaction question") - if ( - not isinstance(self.records, tuple) - or not self.records - or not all(isinstance(record, EvidenceRecord) for record in self.records) - ): - raise VerdictContractError("bounded satisfaction requires evidence records") - if not isinstance(self.scope, AnalyzedScope): - raise VerdictContractError("bounded satisfaction requires an AnalyzedScope") - if any(record.decision.outcome is not NeutralOutcome.SATISFIED for record in self.records): - raise VerdictContractError("bounded satisfaction accepts only satisfied evidence records") - if any(record.decision.bounded_scope != self.scope for record in self.records): - raise VerdictContractError("satisfaction scope must equal every supporting record scope") - - -@dataclass(frozen=True) -class BoundedSilence: - question_ids: tuple[str, ...] - reasons: tuple[str, ...] - evidence_record_digests: tuple[str, ...] - coverage_digest: str - - def __post_init__(self) -> None: - _text_tuple(self.question_ids, "silence question") - _text_tuple(self.reasons, "silence reason") - _digest_tuple(self.evidence_record_digests, "silence evidence record") - _digest(self.coverage_digest, "coverage") - - -@dataclass(frozen=True) -class VerdictOutputs: - defects: tuple[DefectClaim, ...] - satisfactions: tuple[BoundedSatisfaction, ...] - silence: tuple[BoundedSilence, ...] - - def __post_init__(self) -> None: - values = ( - (self.defects, DefectClaim, "defects"), - (self.satisfactions, BoundedSatisfaction, "satisfactions"), - (self.silence, BoundedSilence, "silence"), - ) - for items, item_type, field in values: - if not isinstance(items, tuple) or not all(isinstance(item, item_type) for item in items): - raise VerdictContractError(f"{field} must be a tuple of {item_type.__name__}") - - -def _digest(value: object, field: str) -> None: - if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: - raise VerdictContractError(f"{field} must be a full lowercase SHA-256 digest") - - -def _text_tuple(value: object, field: str) -> None: - if not isinstance(value, tuple) or not value or not all(isinstance(item, str) and item for item in value): - raise VerdictContractError(f"{field} identities must be a nonempty tuple") - if len(set(value)) != len(value): - raise VerdictContractError(f"duplicate {field} identity") - - -def _digest_tuple(value: object, field: str) -> None: - _text_tuple(value, field) - assert isinstance(value, tuple) - for item in value: - _digest(item, field) - - -def _region_contains(region: SourceRegion, location: SourceLocation) -> bool: - return location.path == region.path and region.start_line <= location.line <= region.end_line - - -def _scope_contains(scope: AnalyzedScope, location: SourceLocation) -> bool: - return any(_region_contains(region, location) for region in scope.regions) - - -def _members(record: EvidenceRecord, role: str) -> tuple[SemanticRelation | Observation, ...]: - transitions = tuple(item for item in record.packet.transitions if item.role == role) - observations = tuple(item for item in record.packet.observations if item.role == role) - return (*transitions, *observations) - - -def _question_state(record: EvidenceRecord, question: ResidualQuestion) -> tuple[bool, tuple[str, ...]]: - reasons: set[str] = set() - if record.decision.family_code not in question.allowed_decisions: - reasons.add(f"decision_not_declared_for_question:{record.decision.family_code}") - for role in question.required_roles: - members = _members(record, role) - if not members: - reasons.add(f"missing_required_role:{role}") - elif not any( - item.status is RelationStatus.COMPLETE and _scope_contains(record.decision.bounded_scope, item.location) for item in members - ): - relation_reasons = {reason for item in members if isinstance(item, SemanticRelation) for reason in item.reasons} - complete_outside = any(item.status is RelationStatus.COMPLETE for item in members) - fallback = f"evidence_outside_bounded_scope:{role}" if complete_outside else f"incomplete_required_role:{role}" - reasons.update(relation_reasons or record.packet.completeness_reasons or (fallback,)) - if record.decision.outcome is NeutralOutcome.INDETERMINATE: - reasons.update(record.decision.reasons) - complete = not reasons and record.decision.outcome is not NeutralOutcome.INDETERMINATE - return complete, tuple(sorted(reasons)) - - -def _location(record: EvidenceRecord, question: ResidualQuestion) -> tuple[SourceLocation, SourceRegion]: - candidates = tuple( - item.location for role in question.required_roles for item in _members(record, role) if item.status is RelationStatus.COMPLETE - ) - ordered = sorted(candidates, key=lambda item: (item.path, item.line, -1 if item.column is None else item.column, item.symbol or "")) - for location in ordered: - anchors = tuple(region for region in record.decision.bounded_scope.regions if _region_contains(region, location)) - if anchors: - anchor = min(anchors, key=lambda item: (item.end_line - item.start_line, item.path, item.start_line)) - return location, anchor - raise VerdictContractError("complete defect evidence has no exact location inside its bounded change scope") - - -def _validate_inputs(records: Sequence[EvidenceRecord], contract: FamilyContract, coverage_digest: str) -> tuple[EvidenceRecord, ...]: - _digest(coverage_digest, "coverage") - try: - validate_contract(contract) - except (TypeError, ValueError) as exc: - raise VerdictContractError(f"family contract is invalid: {exc}") from exc - values = tuple(records) - if not values or not all(isinstance(record, EvidenceRecord) for record in values): - raise VerdictContractError("records must be a nonempty sequence of EvidenceRecord values") - identities: set[str] = set() - try: - for record in values: - record.__post_init__() - if record.record_id != derive_record_id(record.packet, record.decision): - raise EvidenceValidationError("record identity cannot be re-derived") - if record.record_id in identities: - raise EvidenceValidationError("duplicate evidence record identity") - identities.add(record.record_id) - validate_scope(record.packet, record.packet.scope) - validate_record(contract, record) - except (TypeError, ValueError, EvidenceValidationError, ScopeValidationError, FamilyContractError) as exc: - raise VerdictContractError(f"evidence record validation failed: {exc}") from exc - return values - - -def _silence( - question: ResidualQuestion, - reasons: set[str], - record_ids: set[str], - coverage_digest: str, -) -> BoundedSilence | None: - if not reasons: - return None - return BoundedSilence( - (question.question_id,), - tuple(sorted(reasons)), - tuple(sorted(record_ids)), - coverage_digest, - ) - - -def derive_verdict_outputs( - records: Sequence[EvidenceRecord], - contract: FamilyContract, - coverage_digest: str, -) -> VerdictOutputs: - """Derive bounded outputs without converting incomplete evidence to safety.""" - validated = _validate_inputs(records, contract, coverage_digest) - defects: dict[str, DefectClaim] = {} - satisfactions: dict[str, set[str]] = {} - silence: list[BoundedSilence] = [] - for question in contract.residual_questions: - reasons: set[str] = set() - undecided: set[str] = set() - for record in validated: - complete, record_reasons = _question_state(record, question) - if not complete: - reasons.update(record_reasons) - undecided.add(record.record_id) - continue - if record.decision.outcome is NeutralOutcome.VIOLATION: - if record.record_id not in defects: - location, anchor = _location(record, question) - defects[record.record_id] = DefectClaim(record, record.decision.basis, location, anchor) - elif record.decision.outcome is NeutralOutcome.SATISFIED: - satisfactions.setdefault(record.record_id, set()).add(question.question_id) - bounded = _silence(question, reasons, undecided, coverage_digest) - if bounded is not None: - silence.append(bounded) - records_by_id = {record.record_id: record for record in validated} - satisfied = tuple( - BoundedSatisfaction(tuple(sorted(question_ids)), (records_by_id[record_id],), records_by_id[record_id].decision.bounded_scope) - for record_id, question_ids in sorted(satisfactions.items()) - ) - return VerdictOutputs( - tuple(defects[identity] for identity in sorted(defects)), - satisfied, - tuple(sorted(silence, key=lambda item: item.question_ids)), - ) - - -__all__ = [ - "BoundedSatisfaction", - "BoundedSilence", - "DefectClaim", - "VerdictContractError", - "VerdictOutputs", - "derive_verdict_outputs", -] diff --git a/tests/core_migration_audit.py b/tests/core_migration_audit.py index 8457f3f..079d1f2 100644 --- a/tests/core_migration_audit.py +++ b/tests/core_migration_audit.py @@ -7,6 +7,8 @@ import subprocess from dataclasses import dataclass from pathlib import Path +from core_migration_scope import CUTOVER_DELETIONS + REPO = Path(__file__).resolve().parents[1] _EXCLUDED_PREFIXES = ("benchmarks/fixtures/", "benchmarks/pairs/", "tests/fixtures/") _LEGACY_IMPORTS: dict[str, frozenset[str] | None] = { @@ -47,7 +49,7 @@ _LEGACY_IMPORTS: dict[str, frozenset[str] | None] = { } ), "openultrasast.core.budget": frozenset({"Budget", "Complexity"}), - "openultrasast.core.consumer": None, + "openultrasast.core.consumer": frozenset({"consume_verdict"}), "openultrasast.core.coverage": frozenset({"Coverage", "InstrumentFailure", "ResolutionOutcome", "check_reads"}), "openultrasast.core.factpack": frozenset({"FactPackError", "admit_fact_pack", "admit_pack_file", "load_pack_file", "verify_fact_pack"}), "openultrasast.core.probes": frozenset( @@ -62,7 +64,7 @@ _LEGACY_IMPORTS: dict[str, frozenset[str] | None] = { ), "openultrasast.core.provenance": frozenset({"ProvenanceMismatch", "stamp", "verify_stamp"}), "openultrasast.core.trace": None, - "openultrasast.core.verdict": None, + "openultrasast.core.verdict": frozenset({"Judgment", "ProvenClaim", "as_clean"}), } _TOP_LEVEL_COMPATIBILITY = _LEGACY_IMPORTS["openultrasast.core"] or frozenset() @@ -80,10 +82,22 @@ class ReaderAuditReport: def tracked_python() -> tuple[Path, ...]: completed = subprocess.run(["git", "ls-files", "*.py"], cwd=REPO, check=True, capture_output=True, text=True) + deleted = set( + subprocess.run( + ["git", "ls-files", "--deleted", "--", "*.py"], + cwd=REPO, + check=True, + capture_output=True, + text=True, + ).stdout.splitlines() + ) + unexpected = deleted - CUTOVER_DELETIONS + if unexpected: + raise LegacyReaderAuditError(f"unexpected tracked Python deletion: {min(unexpected)}") return tuple( REPO / relative for relative in completed.stdout.splitlines() - if relative.endswith(".py") and not relative.startswith(_EXCLUDED_PREFIXES) + if relative.endswith(".py") and relative not in deleted and not relative.startswith(_EXCLUDED_PREFIXES) ) diff --git a/tests/core_migration_boundary.py b/tests/core_migration_boundary.py index 6f38c10..2e555b9 100644 --- a/tests/core_migration_boundary.py +++ b/tests/core_migration_boundary.py @@ -18,6 +18,7 @@ from core_migration_scope import ( COMPATIBILITY_PATHS, COMPATIBILITY_SHIMS, FROZEN_TOP_LEVEL_V1, + REPLACED_SHIM_MEMBERS, REPO, MigrationScopeError, authoritative_cutover_scope, @@ -29,9 +30,7 @@ from core_migration_scope import ( relative_path, scope_identity, ) -from core_migration_scope import ( - static_top_level_v1 as _static_top_level_v1, -) +from core_migration_scope import static_top_level_exports as _static_top_level_exports class LegacyBoundaryError(ValueError): @@ -71,9 +70,9 @@ class _Bindings: import_functions: frozenset[str] -def static_top_level_v1(root: Path = REPO) -> tuple[str, ...]: +def static_top_level_exports(root: Path = REPO) -> tuple[str, ...]: try: - return _static_top_level_v1(root) + return _static_top_level_exports(root) except MigrationScopeError as exc: raise LegacyBoundaryError(str(exc)) from exc @@ -215,6 +214,19 @@ def _classify_import_from( legacy.add(_site(relative, node, parents, "compatibility_shim_import", shim)) if module in COMPATIBILITY_SHIMS: legacy.add(_site(relative, node, parents, "compatibility_shim_import", module)) + _record_replaced_shim_members(relative, node, parents, module, names, legacy) + + +def _record_replaced_shim_members( + relative: str, + node: ast.ImportFrom, + parents: dict[ast.AST, ast.AST], + module: str, + names: set[str], + legacy: set[str], +) -> None: + for name in sorted(names.intersection(REPLACED_SHIM_MEMBERS.get(module, frozenset()))): + legacy.add(_site(relative, node, parents, "retired_shim_member_import", f"{module}.{name}")) def _reference_sites( @@ -449,5 +461,5 @@ __all__ = [ "boundary_python", "inspect_legacy_boundary", "require_cutover_ready", - "static_top_level_v1", + "static_top_level_exports", ] diff --git a/tests/core_migration_inventory.py b/tests/core_migration_inventory.py index e45170e..7d4ca23 100644 --- a/tests/core_migration_inventory.py +++ b/tests/core_migration_inventory.py @@ -14,6 +14,7 @@ from pathlib import Path from benchmark_migration_evidence import BENCHMARK_TERMINAL_EVIDENCE, BenchmarkTerminalEvidence from consumer_migration_evidence import CONSUMER_TERMINAL_EVIDENCE, REVIEWED_SUCCESSOR_MODULES, ConsumerTerminalEvidence +from core_migration_scope import CUTOVER_DELETIONS from openultrasast.core.legacy import LegacyArtifact, decode_legacy_artifact @@ -830,16 +831,43 @@ def _approved_terminal_source_closure(entry: LegacyInventoryEntry, *, root: Path return True if _benchmark_terminal_source_closure(entry, root=root, tree=tree): return True + transfer = _APPROVED_MIXED_TRANSFERS.get(entry.entry_id) + if ( + transfer is not None + and entry.disposition is not MigrationDisposition.OPEN + and entry.entry_id == entry_identity(entry) + and _stable_entry_fields(entry) == transfer[0] + and _cutover_complete(root) + and not any(_exact_anchor_present(anchor, tree) for anchor in entry.capability_anchors) + ): + return True approved = _APPROVED_TERMINAL_SOURCE_CLOSURES.get(entry.entry_id) is entry.disposition return approved and entry.entry_id == entry_identity(entry) +def _cutover_complete(root: Path) -> bool: + if any((root / relative).exists() for relative in CUTOVER_DELETIONS): + return False + public = tuple(root / f"src/openultrasast/core/{name}.py" for name in ("__init__", "admission", "consumer", "verdict")) + return all(path.is_file() and path.stat().st_size > 0 for path in public) + + +def _approved_cutover_deletion(entry: LegacyInventoryEntry, *, root: Path) -> bool: + if entry.disposition is MigrationDisposition.OPEN or entry.path not in CUTOVER_DELETIONS or entry.entry_id != entry_identity(entry): + return False + return _cutover_complete(root) + + def validate_inventory_sources(inventory: LegacyMigrationInventory, *, root: Path) -> None: """Prove sources are readable and every open migration target remains present.""" for entry in inventory.entries: path = root / entry.path try: content = path.read_bytes() + except FileNotFoundError as exc: + if _approved_cutover_deletion(entry, root=root): + continue + raise LegacyInventoryError(f"inventory source is not readable: {entry.path}") from exc except OSError as exc: raise LegacyInventoryError(f"inventory source is not readable: {entry.path}") from exc if not content: @@ -849,8 +877,8 @@ def validate_inventory_sources(inventory: LegacyMigrationInventory, *, root: Pat except (SyntaxError, UnicodeDecodeError) as exc: raise LegacyInventoryError(f"inventory source is not parseable: {entry.path}") from exc is_shim = _explicit_legacy_shim(tree, entry.symbol) - moved_owner = _moved_mixed_v1_owner(entry, root=root) terminal_closure = _approved_terminal_source_closure(entry, root=root, tree=tree) + moved_owner = False if terminal_closure else _moved_mixed_v1_owner(entry, root=root) if entry.symbol not in _defined_symbols(tree) and not is_shim and not moved_owner and not terminal_closure: raise LegacyInventoryError(f"inventory target is absent: {entry.path}:{entry.symbol}") missing = _missing_anchors(entry, tree) diff --git a/tests/core_migration_scope.py b/tests/core_migration_scope.py index 2955df8..f714fb2 100644 --- a/tests/core_migration_scope.py +++ b/tests/core_migration_scope.py @@ -15,10 +15,14 @@ COMPATIBILITY_PATHS = frozenset( src/openultrasast/core/admission.py src/openultrasast/core/consumer.py src/openultrasast/core/trace.py src/openultrasast/core/verdict.py""".split() # noqa: SIM905 ) -COMPATIBILITY_SHIMS = frozenset( - """openultrasast.core._legacy_factpack openultrasast.core.admission - openultrasast.core.consumer openultrasast.core.trace openultrasast.core.verdict""".split() # noqa: SIM905 -) +COMPATIBILITY_SHIMS = frozenset(["openultrasast.core._legacy_factpack", "openultrasast.core.trace"]) +REPLACED_SHIM_MEMBERS = { + "openultrasast.core.admission": frozenset( + {"REPORT_SCHEMA_KEYS", "AdmissionManifest", "AdmissionManifestError", "validate_admission_manifest", "validate_report_schema"} + ), + "openultrasast.core.consumer": frozenset({"consume_verdict"}), + "openultrasast.core.verdict": frozenset({"Judgment", "ProvenClaim", "as_clean"}), +} FROZEN_TOP_LEVEL_V1 = tuple( """REPORT_SCHEMA_KEYS AdmissionManifest AdmissionManifestError BoundedSilence Budget BudgetGateFailure Complexity Coverage FactPack FactPackError FittedComponentBlocked Guard Hop InstrumentFailure Judgment ProbeResult @@ -27,6 +31,17 @@ FROZEN_TOP_LEVEL_V1 = tuple( require_admission_evidence reversed_pair_probe stamp validate_admission_manifest validate_report_schema verify_fact_pack verify_stamp""".split() # noqa: SIM905 ) +CUTOVER_DELETIONS = frozenset( + { + "src/openultrasast/core/_legacy_factpack.py", + "src/openultrasast/core/consumer_v2.py", + "src/openultrasast/core/legacy_v1/__init__.py", + "src/openultrasast/core/legacy_v1/contracts.py", + "src/openultrasast/core/legacy_v1/writer.py", + "src/openultrasast/core/trace.py", + "src/openultrasast/core/verdict_v2.py", + } +) class MigrationScopeError(ValueError): @@ -47,6 +62,19 @@ def boundary_python(root: Path = REPO) -> tuple[Path, ...]: commands = (("git", "ls-files", "*.py"), ("git", "ls-files", "--others", "--exclude-standard", "--", "*.py")) outputs = (subprocess.run(command, cwd=root, check=True, capture_output=True, text=True).stdout for command in commands) relatives = {value for output in outputs for value in output.splitlines() if _owned_python(value)} + deleted = set( + subprocess.run( + ("git", "ls-files", "--deleted", "--", "*.py"), + cwd=root, + check=True, + capture_output=True, + text=True, + ).stdout.splitlines() + ) + unexpected = deleted - CUTOVER_DELETIONS + if unexpected: + raise MigrationScopeError(f"unexpected tracked Python deletion: {min(unexpected)}") + relatives.difference_update(deleted) return tuple(root / relative for relative in sorted(relatives)) @@ -149,8 +177,8 @@ def relative_path(path: Path, root: Path) -> str: return path.as_posix() -def static_top_level_v1(root: Path = REPO) -> tuple[str, ...]: - """Read the migration shim's literal ``__all__`` without importing it.""" +def static_top_level_exports(root: Path = REPO) -> tuple[str, ...]: + """Read the public package's literal ``__all__`` without importing it.""" path = root / "src/openultrasast/core/__init__.py" try: content = path.read_bytes() @@ -185,6 +213,7 @@ __all__ = [ "FROZEN_TOP_LEVEL_V1", "COMPATIBILITY_PATHS", "COMPATIBILITY_SHIMS", + "CUTOVER_DELETIONS", "MigrationScopeError", "REPO", "authoritative_cutover_scope", @@ -197,5 +226,5 @@ __all__ = [ "relative_path", "scope_identity_error", "scope_identity", - "static_top_level_v1", + "static_top_level_exports", ] diff --git a/tests/test_core_admission_metrics.py b/tests/test_core_admission_metrics.py index 35c121b..79990db 100644 --- a/tests/test_core_admission_metrics.py +++ b/tests/test_core_admission_metrics.py @@ -50,7 +50,7 @@ from openultrasast.core.family_catalog import FamilyCatalog, FamilyCatalogEntry, from openultrasast.core.measurement import DenominatorPolicy, IntervalMethod, MetricId, ScopeKey, standard_metric_rules from openultrasast.core.readiness import ReadinessResult from openultrasast.core.semantics import RelationStatus -from openultrasast.core.verdict_v2 import derive_verdict_outputs +from openultrasast.core.verdict import derive_verdict_outputs class _InputManifestSubclass(InputManifest): diff --git a/tests/test_core_consumer_v2.py b/tests/test_core_consumer_v2.py index 8ac4f0d..f398b23 100644 --- a/tests/test_core_consumer_v2.py +++ b/tests/test_core_consumer_v2.py @@ -19,10 +19,10 @@ from test_core_validation_v2 import ( _with_fact_set, ) -import openultrasast.core.consumer_v2 as consumer_v2 +import openultrasast.core.consumer as consumer_v2 from openultrasast.core import budget, canonical, evidence, validation from openultrasast.core._proof_attestation import proof_attestation_bytes -from openultrasast.core.consumer_v2 import ConsumerContractError, PushRequest, consume_records, make_push_request +from openultrasast.core.consumer import ConsumerContractError, PushRequest, consume_records, make_push_request from openultrasast.core.family import FamilyContract from openultrasast.core.proof import ( DIFFERENTIAL_PROOF_SCHEMA, diff --git a/tests/test_core_cutover.py b/tests/test_core_cutover.py new file mode 100644 index 0000000..cb40aa9 --- /dev/null +++ b/tests/test_core_cutover.py @@ -0,0 +1,165 @@ +"""Final deletion-backed gate for the detection-core v1 migration.""" + +from __future__ import annotations + +import ast +import hashlib +import json +import subprocess +import sys +from pathlib import Path + +from core_migration_boundary import BoundaryPhase, boundary_python, inspect_legacy_boundary, require_cutover_ready +from core_migration_inventory import load_migration_inventory, validate_historical_artifacts + +REPO = Path(__file__).resolve().parents[1] +CORE = REPO / "src/openultrasast/core" +INVENTORY = REPO / "tests/fixtures/detection_core/v1/migration_inventory.json" +REMOVED_PATHS = ( + CORE / "legacy_v1", + CORE / "trace.py", + CORE / "_legacy_factpack.py", + CORE / "consumer_v2.py", + CORE / "verdict_v2.py", +) +CANONICAL_V2_MODULES = ("admission", "consumer", "verdict") +OLD_TOP_LEVEL = ( + "AdmissionManifest", + "Budget", + "Coverage", + "FactPack", + "PushRequest", + "PushVerdict", + "TraceRecord", +) + + +def _module_exports(path: Path) -> tuple[str, ...]: + content = path.read_bytes() + assert content, f"public-surface instrument must read nonzero bytes from {path}" + tree = ast.parse(content, filename=path.relative_to(REPO).as_posix()) + declarations = [ + node + for node in tree.body + if isinstance(node, ast.Assign) and any(isinstance(target, ast.Name) and target.id == "__all__" for target in node.targets) + ] + assert len(declarations) == 1 + value = declarations[0].value + assert isinstance(value, (ast.List, ast.Tuple)) + assert all(isinstance(item, ast.Constant) and isinstance(item.value, str) for item in value.elts) + return tuple(item.value for item in value.elts) + + +def _fresh_import_probe(*module_names: str) -> dict[str, object]: + program = """ +import importlib +import json +import sys + +core = importlib.import_module("openultrasast" + ".core") +result = { + "all": tuple(core.__all__), + "old_top_level": tuple(name for name in sys.argv[1].split(",") if hasattr(core, name)), + "modules": {}, +} +for name in sys.argv[2:]: + try: + module = importlib.import_module(name) + except ModuleNotFoundError: + result["modules"][name] = None + else: + result["modules"][name] = tuple(module.__all__) +print(json.dumps(result, sort_keys=True)) +""" + completed = subprocess.run( + [sys.executable, "-c", program, ",".join(OLD_TOP_LEVEL), *module_names], + cwd=REPO, + check=True, + capture_output=True, + text=True, + env={"PYTHONPATH": str(REPO / "src")}, + ) + assert completed.stdout, "fresh import probe must produce nonzero output" + value = json.loads(completed.stdout) + assert isinstance(value, dict) + return value + + +def test_cutover_deletes_every_v1_write_and_temporary_shim_path() -> None: + for path in REMOVED_PATHS: + assert not path.exists(), f"cutover residue remains: {path.relative_to(REPO)}" + for name in CANONICAL_V2_MODULES: + path = CORE / f"{name}.py" + assert path.is_file() + assert path.stat().st_size > 0 + + +def test_fresh_interpreter_rejects_old_imports_and_loads_canonical_v2() -> None: + old_modules = ( + "openultrasast.core.legacy_v1", + "openultrasast.core.trace", + "openultrasast.core._legacy_factpack", + "openultrasast.core." + "consumer_v2", + "openultrasast.core." + "verdict_v2", + ) + canonical = tuple(f"openultrasast.core.{name}" for name in CANONICAL_V2_MODULES) + result = _fresh_import_probe(*old_modules, *canonical) + + assert result["old_top_level"] == [] + assert set(CANONICAL_V2_MODULES) <= set(result["all"]) + modules = result["modules"] + assert isinstance(modules, dict) + assert all(modules[name] is None for name in old_modules) + assert "AdmissionManifestV2" in modules[canonical[0]] + assert "consume_records" in modules[canonical[1]] + assert "derive_verdict_outputs" in modules[canonical[2]] + + +def test_public_v2_surface_has_no_frozen_v1_names() -> None: + exports = _module_exports(CORE / "__init__.py") + + assert set(CANONICAL_V2_MODULES) <= set(exports) + assert set(OLD_TOP_LEVEL).isdisjoint(exports) + + +def test_repository_post_deletion_boundary_is_nonzero_and_exact_zero() -> None: + paths = boundary_python() + inventory = load_migration_inventory(INVENTORY) + report = inspect_legacy_boundary( + paths, + frozen_inventory=inventory, + candidate_inventory=inventory, + expected_scope=paths, + root=REPO, + phase=BoundaryPhase.POST_DELETION, + ) + + assert report.files_read == len(paths) > 0 + assert report.bytes_read > 0 + assert report.open_inventory_entries == () + assert report.legacy_import_sites == () + assert report.legacy_writer_sites == () + assert report.consumer_legacy_import_sites == () + assert report.consumer_legacy_writer_sites == () + assert report.compatibility_implementation_sites == () + assert report.forbidden_dynamic_sites == () + require_cutover_ready( + report, + frozen_inventory=inventory, + candidate_inventory=inventory, + root=REPO, + ) + + +def test_historical_decoder_remains_exact_read_only_and_outside_public_v2() -> None: + inventory_bytes = INVENTORY.read_bytes() + assert inventory_bytes, "historical gate must read nonzero inventory bytes" + inventory = load_migration_inventory(INVENTORY) + artifacts = validate_historical_artifacts(inventory, root=REPO) + + assert len(artifacts) == 2 + assert all(artifact.content for artifact in artifacts) + assert all(hashlib.sha256(artifact.content).hexdigest() == artifact.content_digest for artifact in artifacts) + legacy_exports = _module_exports(CORE / "legacy.py") + assert {"write", "serialize", "to_v2", "compare_identity"}.isdisjoint(legacy_exports) + assert "legacy" not in _module_exports(CORE / "__init__.py") diff --git a/tests/test_core_migration_boundary.py b/tests/test_core_migration_boundary.py index 7eae800..485c44a 100644 --- a/tests/test_core_migration_boundary.py +++ b/tests/test_core_migration_boundary.py @@ -14,7 +14,7 @@ from core_migration_boundary import ( boundary_python, inspect_legacy_boundary, require_cutover_ready, - static_top_level_v1, + static_top_level_exports, ) from core_migration_inventory import ( LegacyInventoryError, @@ -214,7 +214,9 @@ def test_cutover_refuses_empty_scope_and_migration_compatibility(tmp_path: Path) def test_complete_frozen_top_level_surface_is_reported_and_star_is_forbidden(tmp_path: Path) -> None: - assert static_top_level_v1(REPO) == FROZEN_TOP_LEVEL_V1 + exports = static_top_level_exports(REPO) + assert exports == ("admission", "consumer", "verdict") + assert set(exports).isdisjoint(FROZEN_TOP_LEVEL_V1) assert len(FROZEN_TOP_LEVEL_V1) == 38 imports = ", ".join(FROZEN_TOP_LEVEL_V1) attributes = "\n".join(f"value_{index} = core_api.{name}" for index, name in enumerate(FROZEN_TOP_LEVEL_V1)) @@ -283,16 +285,25 @@ def test_repository_scope_proves_nonzero_inputs_before_reporting_sites() -> None paths = boundary_python() frozen = load_migration_inventory(INVENTORY) - report = inspect_legacy_boundary(paths, frozen_inventory=frozen, root=REPO) + report = inspect_legacy_boundary( + paths, + frozen_inventory=frozen, + candidate_inventory=frozen, + expected_scope=paths, + root=REPO, + phase=BoundaryPhase.POST_DELETION, + ) assert report.files_read == len(paths) > 0 assert report.bytes_read > 0 - assert report.legacy_import_sites + assert report.legacy_import_sites == () + assert report.legacy_writer_sites == () assert report.open_inventory_entries == () assert report.consumer_legacy_import_sites == () assert report.consumer_legacy_writer_sites == () assert report.forbidden_dynamic_sites == () - assert report.compatibility_implementation_sites + assert report.compatibility_implementation_sites == () + _require_cutover(report, frozen=frozen, candidate=frozen, root=REPO) def test_cutover_rejects_positive_but_incomplete_repository_scope() -> None: diff --git a/tests/test_core_migration_namespace.py b/tests/test_core_migration_namespace.py index fc584b1..f6bb49c 100644 --- a/tests/test_core_migration_namespace.py +++ b/tests/test_core_migration_namespace.py @@ -1,4 +1,4 @@ -"""Structural ownership tests for the temporary v1 compatibility implementation.""" +"""Structural ownership tests for the final v2-only core namespace.""" from __future__ import annotations @@ -7,7 +7,7 @@ from pathlib import Path REPO = Path(__file__).resolve().parents[1] CORE = REPO / "src/openultrasast/core" -PURE_COMPATIBILITY_MODULES = ("admission.py", "consumer.py", "trace.py", "verdict.py", "_legacy_factpack.py") +REMOVED_COMPATIBILITY = ("trace.py", "_legacy_factpack.py", "consumer_v2.py", "verdict_v2.py") MIXED_LEGACY_NAMES = { "budget": {"Budget", "Complexity"}, "coverage": {"Coverage", "ResolutionOutcome", "check_reads"}, @@ -38,76 +38,58 @@ def _imported(tree: ast.Module) -> set[str]: return {alias.asname or alias.name for node in tree.body if isinstance(node, (ast.Import, ast.ImportFrom)) for alias in node.names} -def test_legacy_contract_values_have_one_compatibility_owner() -> None: - contracts = _tree(CORE / "legacy_v1/contracts.py") - expected = { - "AdmissionManifest", - "AdmissionManifestError", - "BoundedSilence", - "Budget", - "BudgetGateFailure", - "Complexity", - "Coverage", - "FactPackError", - "FittedComponentBlocked", - "Guard", - "Hop", - "InstrumentFailure", - "Judgment", - "LegacyFactPack", - "LegacyProbeResult", - "ProvenClaim", - "ProvenanceMismatch", - "PushRequest", - "PushVerdict", - "ResolutionOutcome", - "ShortcutDetected", - "TraceRecord", - "Witness", - } - assert expected <= _defined(contracts) - - -def test_pure_compatibility_modules_define_no_contract_or_serializer() -> None: - for name in PURE_COMPATIBILITY_MODULES: - tree = _tree(CORE / name) - assert not _defined(tree) - assert not any(isinstance(node, (ast.Import, ast.ImportFrom)) and "json" in ast.unparse(node) for node in tree.body) - assert any(isinstance(node, ast.ImportFrom) and node.module == "openultrasast.core.legacy_v1.contracts" for node in tree.body) - - -def test_writer_is_the_only_compatibility_json_serializer() -> None: - namespace = CORE / "legacy_v1" - paths = sorted(namespace.glob("*.py")) - assert {path.name for path in paths} == {"__init__.py", "contracts.py", "writer.py"} - json_importers = [ - path.name - for path in paths - if any(isinstance(node, (ast.Import, ast.ImportFrom)) and "json" in ast.unparse(node) for node in _tree(path).body) +def _literal_all(tree: ast.Module) -> tuple[str, ...]: + values = [ + node.value + for node in tree.body + if isinstance(node, ast.Assign) and any(isinstance(target, ast.Name) and target.id == "__all__" for target in node.targets) ] - assert json_importers == ["writer.py"] + assert len(values) == 1 and isinstance(values[0], (ast.List, ast.Tuple)) + assert all(isinstance(item, ast.Constant) and isinstance(item.value, str) for item in values[0].elts) + return tuple(item.value for item in values[0].elts) -def test_namespace_and_v2_dependency_directions_are_closed() -> None: - for path in (CORE / "legacy_v1").glob("*.py"): - imports = {node.module for node in ast.walk(_tree(path)) if isinstance(node, ast.ImportFrom) and node.module} - assert all(module.startswith("openultrasast.core.legacy_v1") for module in imports if module.startswith("openultrasast")) - compatibility = {*PURE_COMPATIBILITY_MODULES, "legacy.py", "__init__.py"} - for path in CORE.glob("*.py"): - if path.name not in compatibility: - imports = {node.module for node in ast.walk(_tree(path)) if isinstance(node, ast.ImportFrom) and node.module} - assert not any(module.startswith("openultrasast.core.legacy_v1") for module in imports) +def test_compatibility_namespace_writer_and_temporary_shims_are_absent() -> None: + assert not (CORE / "legacy_v1").exists() + for name in REMOVED_COMPATIBILITY: + assert not (CORE / name).exists() -def test_historical_decoder_stays_independent_from_compatibility_namespace() -> None: - imports = {node.module for node in ast.walk(_tree(CORE / "legacy.py")) if isinstance(node, ast.ImportFrom)} +def test_canonical_v2_replacements_own_the_public_paths() -> None: + consumer = _tree(CORE / "consumer.py") + verdict = _tree(CORE / "verdict.py") + admission = _tree(CORE / "admission.py") + + assert {"PushRequest", "PushVerdict", "consume_records"} <= _defined(consumer) + assert {"BoundedSilence", "DefectClaim", "derive_verdict_outputs"} <= _defined(verdict) + assert {"AdmissionManifestV2", "authorize_admission", "score_admission"} <= set(_literal_all(admission)) + + +def test_core_modules_have_no_legacy_namespace_dependency() -> None: + paths = tuple(sorted(CORE.glob("*.py"))) + assert paths and all(path.read_bytes() for path in paths) + for path in paths: + tree = _tree(path) + imports = {node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) and node.module} + imported_modules = {alias.name for node in ast.walk(tree) if isinstance(node, ast.Import) for alias in node.names} + assert not any(module.startswith("openultrasast.core.legacy_v1") for module in imports | imported_modules) + + +def test_historical_decoder_stays_independent_and_read_only() -> None: + tree = _tree(CORE / "legacy.py") + imports = {node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom)} + exports = set(_literal_all(tree)) + assert "openultrasast.core.legacy_v1" not in imports + assert {"write", "serialize", "to_v2", "compare_identity"}.isdisjoint(exports) -def test_top_level_v1_api_is_only_a_closed_namespace_shim() -> None: +def test_top_level_api_publishes_only_canonical_v2_modules() -> None: tree = _tree(CORE / "__init__.py") modules = {node.module for node in tree.body if isinstance(node, ast.ImportFrom)} - assert modules == {"openultrasast.core.legacy_v1"} + + assert modules == {"openultrasast.core"} + assert _literal_all(tree) == ("admission", "consumer", "verdict") assert not _defined(tree) @@ -115,9 +97,3 @@ def test_mixed_v2_modules_expose_no_legacy_compatibility_names() -> None: for module_name, legacy_names in MIXED_LEGACY_NAMES.items(): tree = _tree(CORE / f"{module_name}.py") assert legacy_names.isdisjoint(_defined(tree) | _imported(tree)) - - -def test_closed_namespace_is_the_only_owner_of_moved_mixed_names() -> None: - contracts = _tree(CORE / "legacy_v1/contracts.py") - expected = set().union(*MIXED_LEGACY_NAMES.values()) - assert expected <= _defined(contracts) diff --git a/tests/test_core_proof.py b/tests/test_core_proof.py index bfa7539..8cbfa68 100644 --- a/tests/test_core_proof.py +++ b/tests/test_core_proof.py @@ -46,7 +46,7 @@ from openultrasast.core.semantics import ( make_observation, make_relation, ) -from openultrasast.core.verdict_v2 import DefectClaim +from openultrasast.core.verdict import DefectClaim def _digest(value: str) -> str: diff --git a/tests/test_core_verdict.py b/tests/test_core_verdict.py index a5da69e..7558e1b 100644 --- a/tests/test_core_verdict.py +++ b/tests/test_core_verdict.py @@ -42,7 +42,7 @@ from openultrasast.core.semantics import ( SourceLocation, make_relation, ) -from openultrasast.core.verdict_v2 import DefectClaim, VerdictContractError, derive_verdict_outputs +from openultrasast.core.verdict import DefectClaim, VerdictContractError, derive_verdict_outputs def _digest(value: str) -> str: