feat(detection-core): isolate legacy compatibility shims

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",

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",

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",

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",
 ]

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",
 ]

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")

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"]

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",
 ]

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",
 ]

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