diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index cbdbea8..3160d72 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -361,7 +361,7 @@ - Done when old writes remain byte-identical, the compatibility namespace is the only live v1 contract and write owner, and ownership/import tests refuse any duplicate implementation. - _Requirements: 14.3, 15.3_ - _Depends: 1.1_ -- [ ] 10.2 Convert the temporary public and module compatibility surfaces into explicit shims +- [x] 10.2 Convert the temporary public and module compatibility surfaces into explicit shims - Keep the old named top-level API and designated compatibility modules as explicit imports from the isolated namespace during migration. - Remove legacy definitions and re-exports from mixed v2 modules so each compatibility name has exactly one owner. - Keep the canonical v2 paths private until the final atomic cutover. diff --git a/benchmarks/valid/arm_regression.py b/benchmarks/valid/arm_regression.py index 0c33300..c601a00 100644 --- a/benchmarks/valid/arm_regression.py +++ b/benchmarks/valid/arm_regression.py @@ -26,8 +26,8 @@ from benchmarks.unseen.replay import canonical from benchmarks.valid.catalog import local_git from benchmarks.valid.development_runner import AnalyzerBundle, stage_funnel from benchmarks.valid.prover_arm import _write_json -from openultrasast.core.coverage import InstrumentFailure as CoreInstrumentFailure -from openultrasast.core.coverage import check_reads +from openultrasast.core.legacy_v1 import InstrumentFailure as CoreInstrumentFailure +from openultrasast.core.legacy_v1 import check_reads from openultrasast.hunt.producers import Context, _Absent, _safe_path from openultrasast.hunt.prover import prove_change diff --git a/src/openultrasast/core/__init__.py b/src/openultrasast/core/__init__.py index 7053174..35b5e35 100644 --- a/src/openultrasast/core/__init__.py +++ b/src/openultrasast/core/__init__.py @@ -4,31 +4,46 @@ Owns contracts only -- no analysis logic. Arms implement the trace contract; runners and consumers adopt coverage, budget, provenance, and admission contracts. """ -from openultrasast.core._legacy_factpack import LegacyFactPack as FactPack -from openultrasast.core.admission import ( +from openultrasast.core.legacy_v1 import ( REPORT_SCHEMA_KEYS, AdmissionManifest, AdmissionManifestError, - validate_admission_manifest, - validate_report_schema, -) -from openultrasast.core.budget import Budget, BudgetGateFailure, Complexity, check_budget, derive_budget -from openultrasast.core.consumer import PushRequest, PushVerdict, consume_verdict -from openultrasast.core.coverage import Coverage, InstrumentFailure, ResolutionOutcome, check_reads -from openultrasast.core.factpack import FactPackError, admit_fact_pack, verify_fact_pack -from openultrasast.core.probes import ( + 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, ) -from openultrasast.core.probes import ( - LegacyProbeResult as ProbeResult, -) -from openultrasast.core.provenance import ProvenanceMismatch, stamp, verify_stamp -from openultrasast.core.trace import Guard, Hop, TraceRecord, Witness -from openultrasast.core.verdict import BoundedSilence, Judgment, ProvenClaim, as_clean __all__ = [ "REPORT_SCHEMA_KEYS", diff --git a/src/openultrasast/core/budget.py b/src/openultrasast/core/budget.py index 3f80099..c319a32 100644 --- a/src/openultrasast/core/budget.py +++ b/src/openultrasast/core/budget.py @@ -7,7 +7,6 @@ import math import re from collections.abc import Sequence from dataclasses import dataclass -from typing import overload from openultrasast.core import canonical @@ -48,29 +47,6 @@ def _finite_number(value: object, field: str, *, positive: bool = False) -> floa return number -@dataclass(frozen=True) -class Complexity: - """Legacy measured change complexity.""" - - regions: int - calls: int - changed_bytes: int - - -@dataclass(frozen=True) -class Budget: - """Legacy derived per-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 ChangeComplexity: change_id: str @@ -302,73 +278,12 @@ def _compatible(value: ChangeComplexity, reference: ReferenceWorkload) -> bool: ) -@overload -def derive_budget(samples: Sequence[tuple[Complexity, float]], *, workload_class: str, percentile: float, hardware: str) -> Budget: ... - - -@overload def derive_budget( samples: Sequence[MeasuredSample], workload_class: str, percentile: float, reference_workload: ReferenceWorkload, reference_hardware: ReferenceHardware, -) -> BudgetDerivation: ... - - -def derive_budget( - samples: Sequence[tuple[Complexity, float]] | Sequence[MeasuredSample], - workload_class: str | None = None, - percentile: float | None = None, - *v2_context: object, - hardware: str | None = None, - reference_workload: ReferenceWorkload | None = None, - reference_hardware: ReferenceHardware | None = None, -) -> Budget | BudgetDerivation: - if v2_context: - if len(v2_context) != 2 or reference_workload is not None or reference_hardware is not None: - raise TypeError("v2 positional context must be reference workload and hardware") - positional_workload, positional_hardware = v2_context - if not isinstance(positional_workload, ReferenceWorkload) or not isinstance(positional_hardware, ReferenceHardware): - raise TypeError("v2 positional context must be reference workload and hardware") - reference_workload, reference_hardware = positional_workload, positional_hardware - if workload_class is None or percentile is None: - raise TypeError("workload_class and percentile are required") - if reference_workload is not None or reference_hardware is not None: - if ( - not isinstance(reference_workload, ReferenceWorkload) - or not isinstance(reference_hardware, ReferenceHardware) - or hardware is not None - ): - raise ValueError("v2 derivation requires reference workload and hardware only") - return _derive_v2(samples, workload_class, percentile, reference_workload, reference_hardware) - if hardware is None: - raise ValueError("legacy derivation requires hardware") - return _derive_legacy(samples, workload_class, percentile, hardware) - - -def _derive_legacy( - samples: Sequence[tuple[Complexity, float]] | Sequence[MeasuredSample], - 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 if isinstance(item, tuple)) - if len(seconds) != len(samples): - raise TypeError("legacy samples must be (Complexity, seconds) tuples") - return Budget(workload_class, _quantile(seconds, percentile), len(samples), percentile, hardware) - - -def _derive_v2( - samples: Sequence[tuple[Complexity, float]] | Sequence[MeasuredSample], - workload_class: str, - percentile: float, - reference_workload: ReferenceWorkload, - reference_hardware: ReferenceHardware, ) -> BudgetDerivation: _text(workload_class, "workload_class") value = _finite_number(percentile, "percentile", positive=True) @@ -415,63 +330,32 @@ def _validate_measurement(change: ChangeComplexity, resolver: canonical.ContentR raise ValueError("changed-byte measurement does not match complexity") -@overload -def check_budget(change_id: str, seconds: float, budget: Budget) -> None: ... - - -@overload def check_budget( change: ChangeComplexity, total_reply_ms: int, budget: BudgetDerivation, resolver: canonical.ContentResolver, - /, -) -> None: ... - - -def check_budget( - change_id: str | ChangeComplexity | None = None, - seconds: float | int | None = None, - budget: Budget | BudgetDerivation | None = None, - resolver: canonical.ContentResolver | None = None, - **v2_names: object, ) -> None: - if v2_names: - if set(v2_names) != {"change", "total_reply_ms"} or change_id is not None or seconds is not None: - raise TypeError("use either legacy or v2 budget parameter names") - candidate_change = v2_names["change"] - candidate_elapsed = v2_names["total_reply_ms"] - if not isinstance(candidate_change, ChangeComplexity) or not isinstance(candidate_elapsed, int): - raise TypeError("v2 budget parameter values have invalid types") - change_id, seconds = candidate_change, candidate_elapsed - if change_id is None or seconds is None or budget is None: - raise TypeError("change, elapsed time, and budget are required") - if isinstance(change_id, str) and isinstance(budget, Budget): - if seconds > budget.seconds: - raise BudgetGateFailure(f"budget:{change_id}:{seconds:g}s>{budget.seconds:g}s ({budget.describe()})") - return - if not isinstance(change_id, ChangeComplexity) or not isinstance(budget, BudgetDerivation) or resolver is None: + if not isinstance(change, ChangeComplexity) or not isinstance(budget, BudgetDerivation): raise TypeError("v2 budget check requires complexity, derivation, and resolver") - _validate_complexity(change_id) + _validate_complexity(change) _validate_derivation(budget) - _positive_int(seconds, "total_reply_ms") - if change_id.workload_class != budget.workload_class or not _compatible(change_id, budget.reference_workload): + _positive_int(total_reply_ms, "total_reply_ms") + if change.workload_class != budget.workload_class or not _compatible(change, budget.reference_workload): raise ValueError("change complexity is incompatible with budget derivation") - _validate_measurement(change_id, resolver) - if seconds > budget.budget_ms: + _validate_measurement(change, resolver) + if total_reply_ms > budget.budget_ms: raise BudgetGateFailure( - f"budget:{change_id.change_id}:{seconds}ms>{budget.budget_ms}ms " + f"budget:{change.change_id}:{total_reply_ms}ms>{budget.budget_ms}ms " f"class={budget.workload_class} derivation={budget.derivation_digest}" ) __all__ = [ - "Budget", "BudgetDerivation", "BudgetGateFailure", "CHANGED_BYTES_MEASUREMENT_SCHEMA", "ChangeComplexity", - "Complexity", "MeasuredSample", "ReferenceHardware", "ReferenceWorkload", diff --git a/src/openultrasast/core/coverage.py b/src/openultrasast/core/coverage.py index 469f3d0..2af7c1f 100644 --- a/src/openultrasast/core/coverage.py +++ b/src/openultrasast/core/coverage.py @@ -9,8 +9,8 @@ from __future__ import annotations import enum import re -from collections.abc import Mapping, Sequence -from dataclasses import dataclass, field +from collections.abc import Sequence +from dataclasses import dataclass from openultrasast.core.canonical import JsonValue, content_digest, require_unique_identities from openultrasast.core.semantics import RelationStatus @@ -20,45 +20,7 @@ _DIGEST = re.compile(r"[0-9a-f]{64}") class InstrumentFailure(ValueError): - """The run's input was not read; any result from it is invalid.""" - - -class ResolutionOutcome(enum.Enum): - """Exactly one bucket per call (Req 3.1).""" - - COMPLETE = "complete" - INCOMPLETE = "incomplete" - UNRESOLVED = "unresolved" - - -@dataclass(frozen=True) -class Coverage: - """Coverage of one analyzed change.""" - - 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: - """Refuse coverage where calls lack exactly one resolution bucket.""" - 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})" - ) - - -def check_reads(reads: Mapping[str, int]) -> None: - """Fail the run when any tracked input was read as zero bytes (Req 3.2).""" - 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)}") + """The v2 run did not read every tracked input.""" def _require_text(value: object, field_name: str) -> None: @@ -378,7 +340,6 @@ def diagnose_first_zero(funnel: CoverageFunnel, expected: Sequence[StageExpectat __all__ = [ - "Coverage", "CoverageFunnel", "CoverageStage", "InputManifest", @@ -386,10 +347,8 @@ __all__ = [ "InstrumentFailure", "ReasonCount", "RelationStatusCount", - "ResolutionOutcome", "StageDiagnosis", "StageExpectation", - "check_reads", "derive_input_manifest_digest", "diagnose_first_zero", "validate_instrument", diff --git a/src/openultrasast/core/factpack.py b/src/openultrasast/core/factpack.py index 6a7e606..8855e13 100644 --- a/src/openultrasast/core/factpack.py +++ b/src/openultrasast/core/factpack.py @@ -10,7 +10,6 @@ from types import MappingProxyType from openultrasast.core import _factpack_digest as digest_projection from openultrasast.core import canonical -from openultrasast.core._legacy_factpack import FactPackError, admit_fact_pack, admit_pack_file, load_pack_file, verify_fact_pack from openultrasast.core.custody import ( LineagePurpose, PopulationCustodyPolicy, @@ -497,7 +496,6 @@ def validate_fact_pack( __all__ = [ "FACT_PACK_SCHEMA", "FactPack", - "FactPackError", "FactPackValidationError", "IndependenceAttestation", "KnowledgeEvidence", @@ -507,13 +505,9 @@ __all__ = [ "SemanticFact", "StructuralRequirement", "StructuralRequirementKind", - "admit_fact_pack", - "admit_pack_file", "derive_fact_pack_content_digest", "derive_independence_attestation_digest", "derive_review_evidence_digest", "derive_review_independence_policy_digest", - "load_pack_file", "validate_fact_pack", - "verify_fact_pack", ] diff --git a/src/openultrasast/core/legacy_v1/__init__.py b/src/openultrasast/core/legacy_v1/__init__.py index 58ec318..dc58bc9 100644 --- a/src/openultrasast/core/legacy_v1/__init__.py +++ b/src/openultrasast/core/legacy_v1/__init__.py @@ -5,47 +5,88 @@ from openultrasast.core.legacy_v1.contracts import ( 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 index c390bd1..5b02cf2 100644 --- a/src/openultrasast/core/legacy_v1/contracts.py +++ b/src/openultrasast/core/legacy_v1/contracts.py @@ -3,7 +3,7 @@ from __future__ import annotations import enum -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -17,6 +17,32 @@ 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" @@ -121,6 +147,52 @@ class LegacyFactPack: 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", @@ -279,27 +351,89 @@ def admit_pack_file(path: str | Path, *, review_digest: str) -> LegacyFactPack: return pack -__all__ = [ - "REPORT_SCHEMA_KEYS", - "AdmissionManifest", - "AdmissionManifestError", - "BoundedSilence", - "FactPackError", - "Guard", - "Hop", - "Judgment", - "LegacyFactPack", - "ProvenClaim", - "PushRequest", - "PushVerdict", - "TraceRecord", - "Witness", - "admit_fact_pack", - "admit_pack_file", - "as_clean", - "consume_verdict", - "load_pack_file", - "validate_admission_manifest", - "validate_report_schema", - "verify_fact_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 index b836478..e8c5fae 100644 --- a/src/openultrasast/core/legacy_v1/writer.py +++ b/src/openultrasast/core/legacy_v1/writer.py @@ -31,6 +31,11 @@ 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): @@ -38,4 +43,4 @@ def read_json(path: Path) -> dict[str, Any]: return value -__all__ = ["fact_pack_digest", "read_json", "text_digest", "trace_bytes", "trace_record"] +__all__ = ["fact_pack_digest", "input_digest", "read_json", "text_digest", "trace_bytes", "trace_record"] diff --git a/src/openultrasast/core/probes.py b/src/openultrasast/core/probes.py index 8a170a5..24e2a4a 100644 --- a/src/openultrasast/core/probes.py +++ b/src/openultrasast/core/probes.py @@ -4,9 +4,7 @@ from __future__ import annotations import math import re -from collections.abc import Callable, Mapping from dataclasses import dataclass, replace -from typing import Any from openultrasast.core import canonical from openultrasast.core.evidence import EvidenceRecord @@ -21,23 +19,6 @@ class ProbeValidationError(ValueError): """A probe input or claimed result cannot be reproduced.""" -class ShortcutDetected(ValueError): - """A legacy component separates a fix from its reversed surface.""" - - -class FittedComponentBlocked(ValueError): - """A legacy fitted component lacks its required release evidence.""" - - -@dataclass(frozen=True) -class LegacyProbeResult: - """Frozen v1 shortcut result retained until the migration cutover.""" - - separated: bool - fix_score: float - reversed_score: float - - def _digest(value: object, field: str) -> None: if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: raise ProbeValidationError(f"{field} must be a full lowercase SHA-256 digest") @@ -379,39 +360,7 @@ def validate_probe_result( raise ProbeValidationError("claimed probe result does not match canonical evaluation") -def reversed_pair_probe( - score_fn: Callable[[Mapping[str, Any]], float], - fix_change: Mapping[str, Any], - reversed_change: Mapping[str, Any], -) -> LegacyProbeResult: - """Run the frozen v1 shape probe without granting it v2 authority.""" - 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: - """Preserve the v1 exception behavior during migration.""" - 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: - """Preserve the v1 fitted-component release guard during migration.""" - 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") - - __all__ = [ - "FittedComponentBlocked", - "LegacyProbeResult", "ProbeEvaluationInput", "ProbeObservation", "ProbeResult", @@ -420,12 +369,8 @@ __all__ = [ "RoleProjection", "SemanticEquivalenceReview", "SemanticProjection", - "ShortcutDetected", - "assert_no_shortcut", "derive_equivalence_subject_digest", "evaluate_probe", "project_semantics", - "require_admission_evidence", - "reversed_pair_probe", "validate_probe_result", ] diff --git a/src/openultrasast/core/provenance.py b/src/openultrasast/core/provenance.py index 4c5e297..01e5626 100644 --- a/src/openultrasast/core/provenance.py +++ b/src/openultrasast/core/provenance.py @@ -2,10 +2,7 @@ from __future__ import annotations -import hashlib -import json import re -from collections.abc import Mapping from dataclasses import dataclass, fields from enum import Enum @@ -50,52 +47,6 @@ PROVENANCE_POLICY_SCHEMA = "openultrasast.detection-core.provenance-policy.v2" _DIGEST = re.compile(r"[0-9a-f]{64}") -class ProvenanceMismatch(ValueError): - """A legacy recorded digest does not match its re-derived identity.""" - - -_LEGACY_DIGEST_FIELDS = ( - "analyzer_revision", - "fact_set_digest", - "acquisition_record_digest", - "configuration_digest", - "population_digest", - "input_digest", -) - - -def _input_digest(change_id: str, population_digest: str) -> str: - payload = json.dumps([population_digest, change_id], sort_keys=True, separators=(",", ":")) - return hashlib.sha256(payload.encode()).hexdigest()[:12] - - -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]: - """Preserve the v1 provenance stamp during migration.""" - 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": _input_digest(change_id, population_digest), - } - - -def verify_stamp(recorded: Mapping[str, object], rederived: Mapping[str, object]) -> None: - """Preserve v1 refusal when a legacy digest cannot be re-derived.""" - for field_name in _LEGACY_DIGEST_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}") - - class ProvenanceValidationError(ValueError): """A provenance identity cannot be reproduced from its named inputs.""" @@ -357,7 +308,6 @@ __all__ = [ "NamedInputs", "PROVENANCE_POLICY_SCHEMA", "ProvenanceInputRole", - "ProvenanceMismatch", "ProvenanceStamp", "ProvenanceTrustPolicy", "ProvenanceValidationError", @@ -365,8 +315,6 @@ __all__ = [ "derive_lineage_decision_digest", "derive_lineage_manifest_digest", "derive_trust_policy_digest", - "stamp", "validate_provenance", "validate_lineage", - "verify_stamp", ] diff --git a/src/openultrasast/hunt/prover.py b/src/openultrasast/hunt/prover.py index b80c438..d0dd98c 100644 --- a/src/openultrasast/hunt/prover.py +++ b/src/openultrasast/hunt/prover.py @@ -9,7 +9,7 @@ from dataclasses import asdict, dataclass, field, replace from pathlib import Path from typing import Any -from openultrasast.core.coverage import Coverage as CoreCoverage +from openultrasast.core.legacy_v1 import Coverage as CoreCoverage from .. import mapping, source from ..preprocess import FileTarget, count_loc, detect_language, detect_tags diff --git a/tests/core_migration_inventory.py b/tests/core_migration_inventory.py index 1937a4e..bfbe899 100644 --- a/tests/core_migration_inventory.py +++ b/tests/core_migration_inventory.py @@ -74,6 +74,85 @@ class LegacyMigrationInventory: entries: tuple[LegacyInventoryEntry, ...] +_APPROVED_MIXED_TRANSFERS: dict[ + str, + tuple[tuple[str, str, str, str, tuple[str, ...], str], str, tuple[str, ...]], +] = { + "9eb4cba855044fe025daa7a523458c4fba0b12758ca604c894849f824f55ccb4": ( + ( + "src/openultrasast/core/factpack.py", + "", + "reader", + "v1-contract", + ("import_from:openultrasast.core._legacy_factpack:FactPackError",), + "core", + ), + "openultrasast.core.legacy_v1.contracts", + ("FactPackError",), + ), + "1b90ea5474dff89121dc1317a86892bb0d68f2c64117923c5979cf4bb0cd8b76": ( + ( + "src/openultrasast/core/factpack.py", + "", + "reader", + "v1-contract", + ("import_from:openultrasast.core._legacy_factpack:admit_fact_pack",), + "core", + ), + "openultrasast.core.legacy_v1.contracts", + ("admit_fact_pack",), + ), + "005f1fee3a1a659a65a45d8134bc71f061e314bdc28ba95516e870997c2648cd": ( + ( + "src/openultrasast/core/factpack.py", + "", + "reader", + "v1-contract", + ("import_from:openultrasast.core._legacy_factpack:admit_pack_file",), + "core", + ), + "openultrasast.core.legacy_v1.contracts", + ("admit_pack_file",), + ), + "4c846ba1d97e5a5b4bbf8f1e99e10c40da10906ca93d68596cb987738438582a": ( + ( + "src/openultrasast/core/factpack.py", + "", + "reader", + "v1-contract", + ("import_from:openultrasast.core._legacy_factpack:load_pack_file",), + "core", + ), + "openultrasast.core.legacy_v1.contracts", + ("load_pack_file",), + ), + "23e6ccf25c185ca231d73ab02570b14a9cddae27f7506e95d517e241bb693ae4": ( + ( + "src/openultrasast/core/factpack.py", + "", + "reader", + "v1-contract", + ("import_from:openultrasast.core._legacy_factpack:verify_fact_pack",), + "core", + ), + "openultrasast.core.legacy_v1.contracts", + ("verify_fact_pack",), + ), + "c8aaf430009a592a8ed4361f38023d5faa12a6da6186fc0dcea9c7c522da85ad": ( + ( + "src/openultrasast/core/probes.py", + "reversed_pair_probe", + "writer", + "v1-probe", + ("definition:reversed_pair_probe", "name:LegacyProbeResult"), + "core", + ), + "openultrasast.core.legacy_v1.contracts", + ("reversed_pair_probe", "LegacyProbeResult"), + ), +} + + def _object(value: object, fields: frozenset[str], name: str) -> dict[str, object]: if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): raise LegacyInventoryError(f"{name} must be an object") @@ -311,6 +390,53 @@ def _explicit_legacy_shim(tree: ast.Module, symbol: str) -> bool: ) +def _stable_entry_fields(entry: LegacyInventoryEntry) -> tuple[str, str, str, str, tuple[str, ...], str]: + return ( + entry.path, + entry.symbol, + entry.dependency_kind.value, + entry.artifact_family, + entry.capability_anchors, + entry.owner_boundary, + ) + + +def _read_owner_module(path: Path, description: str) -> ast.Module: + try: + content = path.read_bytes() + except OSError as exc: + raise LegacyInventoryError(f"{description} is not readable") from exc + if not content: + raise LegacyInventoryError(f"{description} must contain nonzero bytes") + try: + return ast.parse(content, filename=path.as_posix()) + except (SyntaxError, UnicodeDecodeError) as exc: + raise LegacyInventoryError(f"{description} is not parseable") from exc + + +def _moved_mixed_v1_owner(entry: LegacyInventoryEntry, *, root: Path) -> bool: + transfer = _APPROVED_MIXED_TRANSFERS.get(entry.entry_id) + if transfer is None or entry.entry_id != entry_identity(entry): + return False + expected_fields, destination_module, destination_symbols = transfer + if _stable_entry_fields(entry) != expected_fields: + return False + facade_path = root / "src/openultrasast/core/legacy_v1/__init__.py" + facade = _read_owner_module(facade_path, "compatibility facade") + facade_imports = { + (node.module, alias.name, alias.asname or alias.name) + for node in facade.body + if isinstance(node, ast.ImportFrom) + for alias in node.names + } + if not all((destination_module, symbol, symbol) in facade_imports for symbol in destination_symbols): + return False + destination_path = root / "src" / Path(*destination_module.split(".")) + destination = _read_owner_module(destination_path.with_suffix(".py"), "compatibility destination") + owned = {node.name for node in destination.body if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef))} + return set(destination_symbols) <= owned + + def validate_inventory_sources(inventory: LegacyMigrationInventory, *, root: Path) -> None: """Prove every registered migration target is readable, nonzero, and present.""" for entry in inventory.entries: @@ -326,10 +452,11 @@ 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) - if entry.symbol not in _defined_symbols(tree) and not is_shim: + moved_owner = _moved_mixed_v1_owner(entry, root=root) + if entry.symbol not in _defined_symbols(tree) and not is_shim and not moved_owner: raise LegacyInventoryError(f"inventory target is absent: {entry.path}:{entry.symbol}") missing = _missing_anchors(entry, tree) - if missing and not is_shim: + if missing and not is_shim and not moved_owner: raise LegacyInventoryError(f"inventory capability is absent: {entry.path}:{missing}") diff --git a/tests/test_core_budget.py b/tests/test_core_budget.py index fa9f1a6..6b1610f 100644 --- a/tests/test_core_budget.py +++ b/tests/test_core_budget.py @@ -5,7 +5,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -from openultrasast.core.budget import ( # noqa: E402 +from openultrasast.core.legacy_v1 import ( # noqa: E402 Budget, BudgetGateFailure, Complexity, diff --git a/tests/test_core_budget_v2.py b/tests/test_core_budget_v2.py index cbb7f9e..0fbd652 100644 --- a/tests/test_core_budget_v2.py +++ b/tests/test_core_budget_v2.py @@ -12,6 +12,10 @@ from detection_core_fixtures import InMemoryContentResolver, stored_artifact from openultrasast.core import budget as budget_v2 from openultrasast.core import canonical, evidence, provenance, semantics +from openultrasast.core.legacy_v1 import Budget as LegacyBudget +from openultrasast.core.legacy_v1 import Complexity as LegacyComplexity +from openultrasast.core.legacy_v1 import check_budget as legacy_check_budget +from openultrasast.core.legacy_v1 import derive_budget as legacy_derive_budget def _digest(value: str) -> str: @@ -342,7 +346,7 @@ def test_scope_revalidates_post_construction_tampering() -> None: def test_legacy_budget_api_remains_available() -> None: - samples = ((budget_v2.Complexity(1, 1, 10), 1.0), (budget_v2.Complexity(2, 2, 20), 2.0)) - legacy = budget_v2.derive_budget(samples, workload_class="legacy", percentile=0.5, hardware="runner") - assert isinstance(legacy, budget_v2.Budget) - budget_v2.check_budget(change_id="change", seconds=1.0, budget=legacy) + samples = ((LegacyComplexity(1, 1, 10), 1.0), (LegacyComplexity(2, 2, 20), 2.0)) + legacy = legacy_derive_budget(samples, workload_class="legacy", percentile=0.5, hardware="runner") + assert isinstance(legacy, LegacyBudget) + legacy_check_budget(change_id="change", seconds=1.0, budget=legacy) diff --git a/tests/test_core_contract.py b/tests/test_core_contract.py index b2a4a1e..a7ccdc0 100644 --- a/tests/test_core_contract.py +++ b/tests/test_core_contract.py @@ -5,10 +5,22 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -from openultrasast.core.coverage import Coverage, InstrumentFailure, check_reads # noqa: E402 -from openultrasast.core.provenance import ProvenanceMismatch, stamp, verify_stamp # noqa: E402 -from openultrasast.core.trace import Guard, Hop, TraceRecord, Witness # noqa: E402 -from openultrasast.core.verdict import BoundedSilence, Judgment, ProvenClaim, as_clean # noqa: E402 +from openultrasast.core.legacy_v1 import ( # noqa: E402 + BoundedSilence, + Coverage, + Guard, + Hop, + InstrumentFailure, + Judgment, + ProvenanceMismatch, + ProvenClaim, + TraceRecord, + Witness, + as_clean, + check_reads, + stamp, + verify_stamp, +) def _trace() -> TraceRecord: diff --git a/tests/test_core_coverage_v2.py b/tests/test_core_coverage_v2.py index 2234e58..7777e8c 100644 --- a/tests/test_core_coverage_v2.py +++ b/tests/test_core_coverage_v2.py @@ -17,6 +17,7 @@ from openultrasast.core.coverage import ( diagnose_first_zero, validate_instrument, ) +from openultrasast.core.legacy_v1 import InstrumentFailure as LegacyInstrumentFailure from openultrasast.core.semantics import RelationStatus DIGEST_A = "a" * 64 @@ -74,6 +75,15 @@ def test_input_manifest_and_receipts_validate_exactly() -> None: validate_instrument(_manifest(), _receipts()) +def test_v2_instrument_failure_is_distinct_from_legacy_and_owns_input_refusal() -> None: + from openultrasast.core.coverage import InstrumentFailure + + assert InstrumentFailure is not LegacyInstrumentFailure + receipt = InputReadReceipt("src/app.py", "src/app.py", False, 12, DIGEST_A) + with pytest.raises(InstrumentFailure, match="not opened"): + validate_instrument(_manifest(), (receipt, _receipts()[1])) + + def test_zero_named_counts_are_allowed_for_measurements_but_not_unresolved_coverage() -> None: with pytest.raises(ValueError, match="unresolved reason count must be positive"): _funnel( @@ -84,6 +94,10 @@ def test_zero_named_counts_are_allowed_for_measurements_but_not_unresolved_cover ) +def test_legacy_instrument_failure_is_available_only_through_compatibility_boundary() -> None: + assert LegacyInstrumentFailure.__module__ == "openultrasast.core.legacy_v1.contracts" + + @pytest.mark.parametrize( ("opened", "byte_count", "digest", "match"), [ diff --git a/tests/test_core_factpack_consumer.py b/tests/test_core_factpack_consumer.py index 4f823d6..3556a55 100644 --- a/tests/test_core_factpack_consumer.py +++ b/tests/test_core_factpack_consumer.py @@ -6,7 +6,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from openultrasast.core.consumer import PushRequest, consume_verdict # noqa: E402 -from openultrasast.core.factpack import FactPackError, admit_fact_pack, verify_fact_pack # noqa: E402 +from openultrasast.core.legacy_v1 import FactPackError, admit_fact_pack, verify_fact_pack # noqa: E402 from openultrasast.core.trace import TraceRecord # noqa: E402 from openultrasast.core.verdict import BoundedSilence, Judgment, ProvenClaim # noqa: E402 diff --git a/tests/test_core_factpack_file.py b/tests/test_core_factpack_file.py index 6a0bbeb..c4ef7b9 100644 --- a/tests/test_core_factpack_file.py +++ b/tests/test_core_factpack_file.py @@ -9,7 +9,7 @@ import pytest sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -from openultrasast.core.factpack import FactPackError, admit_pack_file, verify_fact_pack # noqa: E402 +from openultrasast.core.legacy_v1 import FactPackError, admit_pack_file, verify_fact_pack # noqa: E402 from openultrasast.mapping.installed_fact_authority import ( # noqa: E402 InstalledAuthorityWireError, decode_fact_pack_bytes, diff --git a/tests/test_core_migration_namespace.py b/tests/test_core_migration_namespace.py index 22cbfb6..23f0180 100644 --- a/tests/test_core_migration_namespace.py +++ b/tests/test_core_migration_namespace.py @@ -14,6 +14,20 @@ PURE_COMPATIBILITY_MODULES = ( "verdict.py", "_legacy_factpack.py", ) +MIXED_LEGACY_NAMES = { + "budget": {"Budget", "Complexity"}, + "coverage": {"Coverage", "ResolutionOutcome", "check_reads"}, + "factpack": {"FactPackError", "admit_fact_pack", "admit_pack_file", "load_pack_file", "verify_fact_pack"}, + "probes": { + "FittedComponentBlocked", + "LegacyProbeResult", + "ShortcutDetected", + "assert_no_shortcut", + "require_admission_evidence", + "reversed_pair_probe", + }, + "provenance": {"ProvenanceMismatch", "stamp", "verify_stamp"}, +} def _trace() -> object: @@ -40,14 +54,24 @@ def test_legacy_contract_values_have_one_compatibility_owner() -> None: "AdmissionManifest", "AdmissionManifestError", "BoundedSilence", + "Budget", + "BudgetGateFailure", + "Complexity", + "Coverage", "FactPackError", + "FittedComponentBlocked", "Guard", "Hop", + "InstrumentFailure", "Judgment", "LegacyFactPack", + "LegacyProbeResult", "ProvenClaim", + "ProvenanceMismatch", "PushRequest", "PushVerdict", + "ResolutionOutcome", + "ShortcutDetected", "TraceRecord", "Witness", } @@ -61,6 +85,7 @@ def test_pure_compatibility_modules_define_no_contract_or_serializer() -> None: assert content, f"ownership instrument must read nonzero bytes from {name}" tree = ast.parse(content, filename=path.as_posix()) assert not [node.name for node in tree.body if isinstance(node, ast.ClassDef)] + assert not [node.name for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))] 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) @@ -69,6 +94,7 @@ def test_writer_is_the_only_compatibility_json_serializer() -> None: namespace = REPO / "src/openultrasast/core/legacy_v1" paths = sorted(namespace.glob("*.py")) assert paths and all(path.read_bytes() for path in paths) + assert {path.name for path in paths} == {"__init__.py", "contracts.py", "writer.py"} json_importers = [] for path in paths: tree = ast.parse(path.read_bytes(), filename=path.as_posix()) @@ -109,3 +135,44 @@ def test_historical_decoder_stays_independent_from_compatibility_namespace() -> tree = ast.parse(content, filename=path.as_posix()) imports = {node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom)} assert "openultrasast.core.legacy_v1" not in imports + + +def test_top_level_v1_api_imports_only_from_closed_namespace() -> None: + import openultrasast.core as public_core + from openultrasast.core import legacy_v1 + + path = REPO / "src/openultrasast/core/__init__.py" + content = path.read_bytes() + assert content + tree = ast.parse(content, filename=path.as_posix()) + legacy_imports = { + node.module + for node in tree.body + if isinstance(node, ast.ImportFrom) and any((alias.asname or alias.name) in public_core.__all__ for alias in node.names) + } + assert legacy_imports == {"openultrasast.core.legacy_v1"} + assert all(getattr(public_core, name) is getattr(legacy_v1, name) for name in public_core.__all__) + + +def test_mixed_v2_modules_expose_no_legacy_compatibility_names() -> None: + import importlib + + for module_name, legacy_names in MIXED_LEGACY_NAMES.items(): + module = importlib.import_module(f"openultrasast.core.{module_name}") + assert legacy_names.isdisjoint(vars(module)), f"{module_name} still exposes {legacy_names & vars(module).keys()}" + + +def test_closed_namespace_is_the_only_owner_of_moved_mixed_names() -> None: + from openultrasast.core import legacy_v1 + + expected = set().union(*MIXED_LEGACY_NAMES.values()) + assert all(hasattr(legacy_v1, name) for name in expected) + operation_names = { + "assert_no_shortcut", + "check_reads", + "require_admission_evidence", + "reversed_pair_probe", + "stamp", + "verify_stamp", + } + assert {getattr(legacy_v1, name).__module__ for name in operation_names} == {"openultrasast.core.legacy_v1.contracts"} diff --git a/tests/test_core_migration_provenance.py b/tests/test_core_migration_provenance.py index 501daed..cd63259 100644 --- a/tests/test_core_migration_provenance.py +++ b/tests/test_core_migration_provenance.py @@ -21,6 +21,15 @@ from core_migration_inventory import ( ) INVENTORY = REPO / "tests/fixtures/detection_core/v1/migration_inventory.json" +MIXED_V2_MODULES = frozenset( + { + "src/openultrasast/core/budget.py", + "src/openultrasast/core/coverage.py", + "src/openultrasast/core/factpack.py", + "src/openultrasast/core/probes.py", + "src/openultrasast/core/provenance.py", + } +) @pytest.mark.parametrize( @@ -51,22 +60,18 @@ def test_reader_audit_covers_direct_legacy_submodule_bindings( def test_reader_audit_covers_real_direct_legacy_submodule_reads() -> None: expected = { - ("tests/test_core_budget_v2.py", "module_attr:openultrasast.core.budget:Budget"), - ("tests/test_core_budget_v2.py", "module_attr:openultrasast.core.budget:Complexity"), - ("tests/test_core_probes.py", "module_attr:openultrasast.core.probes:FittedComponentBlocked"), - ("tests/test_core_probes.py", "module_attr:openultrasast.core.probes:LegacyProbeResult"), - ("tests/test_core_probes.py", "module_attr:openultrasast.core.probes:ShortcutDetected"), - ("tests/test_core_probes.py", "module_attr:openultrasast.core.probes:assert_no_shortcut"), - ("tests/test_core_probes.py", "module_attr:openultrasast.core.probes:require_admission_evidence"), - ("tests/test_core_probes.py", "module_attr:openultrasast.core.probes:reversed_pair_probe"), ("tests/test_core_contract.py", "module_attr:openultrasast.core.verdict:__all__"), } - paths = tuple(REPO / path for path in sorted({path for path, _ in expected})) + paths = tuple(REPO / path for path in ("tests/test_core_budget_v2.py", "tests/test_core_contract.py", "tests/test_core_probes.py")) report = audit_legacy_readers(paths) assert report.files_read == 3 and report.bytes_read > 0 assert expected <= set(report.capabilities) + assert not any( + capability.startswith(("module_attr:openultrasast.core.budget:", "module_attr:openultrasast.core.probes:")) + for _path, capability in report.capabilities + ) def test_frozen_inventory_resolves_nonzero_sources_and_targets() -> None: @@ -185,6 +190,9 @@ def test_independent_reader_audit_matches_split_inventory_capabilities() -> None for path, _anchor in open_registered - current: source = (REPO / path).read_bytes() assert source, f"migration shim instrument must read nonzero bytes from {path}" + if path in MIXED_V2_MODULES: + assert b"openultrasast.core.legacy_v1" not in source + continue assert b"openultrasast.core.legacy_v1" in source, f"open reader disappeared without an explicit compatibility shim: {path}" @@ -293,6 +301,55 @@ def test_inventory_refuses_unreadable_zero_and_unknown_targets(tmp_path: Path) - validate_inventory_sources(replace(frozen, entries=(unknown,)), root=tmp_path) +@pytest.mark.parametrize("mutation", ("symbol", "anchor")) +def test_mixed_owner_transfer_refuses_rekeyed_inventory_entries(mutation: str) -> None: + frozen = load_migration_inventory(INVENTORY) + original = next( + entry for entry in frozen.entries if entry.entry_id == "9eb4cba855044fe025daa7a523458c4fba0b12758ca604c894849f824f55ccb4" + ) + if mutation == "symbol": + changed = replace(original, symbol="InstrumentFailure", capability_anchors=("definition:InstrumentFailure",)) + else: + changed = replace( + original, + capability_anchors=("import_from:openultrasast.core._legacy_factpack:InstrumentFailure",), + ) + changed = replace(changed, entry_id=entry_identity(changed)) + + with pytest.raises(LegacyInventoryError, match="target|capability"): + validate_inventory_sources(replace(frozen, entries=(changed,)), root=REPO) + + +@pytest.mark.parametrize( + ("facade_source", "owner_source"), + ( + ( + "from openultrasast.core.legacy_v1.writer import FactPackError\n", + "class FactPackError(ValueError):\n pass\n", + ), + ( + "from openultrasast.core.legacy_v1.contracts import FactPackError\n", + "class AnotherContract:\n pass\n", + ), + ), +) +def test_mixed_owner_transfer_refuses_wrong_or_missing_destination(tmp_path: Path, facade_source: str, owner_source: str) -> None: + frozen = load_migration_inventory(INVENTORY) + original = next( + entry for entry in frozen.entries if entry.entry_id == "9eb4cba855044fe025daa7a523458c4fba0b12758ca604c894849f824f55ccb4" + ) + mixed = tmp_path / original.path + mixed.parent.mkdir(parents=True) + mixed.write_text("pass\n", encoding="utf-8") + namespace = tmp_path / "src/openultrasast/core/legacy_v1" + namespace.mkdir(parents=True) + (namespace / "__init__.py").write_text(facade_source, encoding="utf-8") + (namespace / "contracts.py").write_text(owner_source, encoding="utf-8") + + with pytest.raises(LegacyInventoryError, match="capability"): + validate_inventory_sources(replace(frozen, entries=(original,)), root=tmp_path) + + def test_inventory_refuses_changed_artifact_bytes(tmp_path: Path) -> None: frozen = load_migration_inventory(INVENTORY) artifact = frozen.historical_artifacts[0] diff --git a/tests/test_core_probes.py b/tests/test_core_probes.py index b6cc01a..705b00c 100644 --- a/tests/test_core_probes.py +++ b/tests/test_core_probes.py @@ -28,6 +28,14 @@ from openultrasast.core.family import ( derive_security_object_id, family_contract_digest, ) +from openultrasast.core.legacy_v1 import ( + FittedComponentBlocked, + LegacyProbeResult, + ShortcutDetected, + assert_no_shortcut, + require_admission_evidence, + reversed_pair_probe, +) @dataclass(frozen=True) @@ -393,15 +401,15 @@ def test_legacy_reversed_pair_probe_remains_compatible() -> None: def shape_score(change: Mapping[str, Any]) -> float: return float(change["lines"]) - result = probes.reversed_pair_probe(shape_score, forward, reversed_change) + result = reversed_pair_probe(shape_score, forward, reversed_change) assert result.separated is True - with pytest.raises(probes.ShortcutDetected): - probes.assert_no_shortcut(shape_score, forward, reversed_change) + with pytest.raises(ShortcutDetected): + assert_no_shortcut(shape_score, forward, reversed_change) - assert public_core.ProbeResult is probes.LegacyProbeResult + assert public_core.ProbeResult is LegacyProbeResult def test_legacy_admission_guard_remains_compatible() -> None: - probes.require_admission_evidence(holdout_passed=True, admission_decision="GO") - with pytest.raises(probes.FittedComponentBlocked): - probes.require_admission_evidence(holdout_passed=False, admission_decision="GO") + require_admission_evidence(holdout_passed=True, admission_decision="GO") + with pytest.raises(FittedComponentBlocked): + require_admission_evidence(holdout_passed=False, admission_decision="GO")