feat(detection-core): 8.1 prover native conformance to core trace/coverage contracts

src/openultrasast/hunt/core_adapter.py

@@ -1,97 +0,0 @@
-"""Conformance adapters from the prover's shapes onto the detection-core contracts.
-
-Task 3.1/3.2 of detection-core: the access-control arm's trace and coverage satisfy
-the core schema field-for-field, without changing existing analysis behavior.
-"""
-
-from __future__ import annotations
-
-from collections.abc import Mapping
-from typing import Any
-
-from openultrasast.core.coverage import Coverage as CoreCoverage
-from openultrasast.core.trace import Guard as CoreGuard
-from openultrasast.core.trace import Hop as CoreHop
-from openultrasast.core.trace import TraceRecord
-from openultrasast.core.trace import Witness as CoreWitness
-from openultrasast.core.verdict import Judgment
-from openultrasast.semantic.obligations.abstract_domain import PairJudgment
-from openultrasast.semantic.obligations.trace import Trace
-
-_JUDGMENT_MAP = {
-    PairJudgment.UNGUARDED: Judgment.UNGUARDED,
-    PairJudgment.DISCHARGED: Judgment.DISCHARGED,
-    PairJudgment.INDETERMINATE: Judgment.INDETERMINATE,
-}
-
-_RESOLUTION_KEYS = {
-    "complete_calls": "complete",
-    "incomplete_calls": "incomplete",
-    "unresolved_calls": "unresolved",
-}
-
-
-def trace_to_core(trace: Trace, *, entry: str, digests: Mapping[str, str], complete: bool = False) -> TraceRecord:
-    """Map one prover trace onto the canonical core trace record.
-
-    ``complete`` defaults to False: the prover Trace carries no per-hop
-    completeness, so the adapter never asserts resolved call targets it did not
-    see (Req 2.2 -- incomplete evidence never becomes clean).
-    """
-    state = trace.abstract_state
-    if state is None:
-        judgment = Judgment.INDETERMINATE
-        reasons: tuple[str, ...] = ("abstract_state_unset",)
-        witnesses: tuple[CoreWitness, ...] = ()
-        resource = "unknown"
-        authority = "unknown"
-    else:
-        judgment = _JUDGMENT_MAP[state.path_judgment.judgment]
-        reasons = tuple(state.path_judgment.completeness.reasons) or (
-            ("incomplete",) if not state.path_judgment.completeness.is_complete else ()
-        )
-        witnesses = tuple(
-            CoreWitness(kind="witness", provenance=evidence, relation=match.relation.value)
-            for match in state.path_judgment.matches
-            for evidence in match.evidence
-        )
-        resource = ",".join(state.resource.values) or "unknown"
-        authority = ",".join(state.authority.values) or "unknown"
-    hops = tuple(CoreHop(function=hop.function, call_line=hop.call_line, complete=complete) for hop in trace.hops)
-    hop_guards = tuple(guard for hop in trace.hops for guard in hop.guards_before_call)
-    guards = tuple(CoreGuard(kind=guard.kind, effect=guard.effect) for guard in (*hop_guards, *trace.guards_after))
-    return TraceRecord(
-        entry=entry,
-        hops=hops,
-        operation_kind=trace.operation_kind,
-        resource=resource,
-        authority=authority,
-        witnesses=witnesses,
-        guards=guards,
-        completeness_reasons=reasons,
-        judgment=judgment,
-        digests=dict(digests),
-    )
-
-
-def coverage_to_core(coverage: Any) -> CoreCoverage:
-    """Map the prover's coverage counters onto the core coverage schema."""
-    resolution = dict(getattr(coverage, "resolution_counts", {}) or {})
-    unknown = sorted(key for key in resolution if key not in _RESOLUTION_KEYS)
-    if unknown:
-        raise ValueError(f"unknown resolution bucket keys, refusing to under-count calls: {unknown}")
-    buckets = {name: 0 for name in ("complete", "incomplete", "unresolved")}
-    for key, value in resolution.items():
-        buckets[_RESOLUTION_KEYS[key]] = buckets[_RESOLUTION_KEYS[key]] + int(value)
-    return CoreCoverage(
-        entry_points=int(coverage.entry_points),
-        frameworks_recognised=dict(coverage.frameworks_recognised),
-        calls=sum(buckets.values()),
-        complete=buckets["complete"],
-        incomplete=buckets["incomplete"],
-        unresolved=buckets["unresolved"],
-        unchecked_reasons=dict(coverage.unchecked_by_reason),
-    )
-
-
-__all__ = ["coverage_to_core", "trace_to_core"]

src/openultrasast/hunt/prover.py

@@ -9,6 +9,8 @@ from dataclasses import asdict, dataclass, field, replace
 from pathlib import Path
 from typing import Any
 
+from openultrasast.core.coverage import Coverage as CoreCoverage
+
 from .. import mapping, source
 from ..preprocess import FileTarget, count_loc, detect_language, detect_tags
 from ..semantic.cfg import ControlContext, FunctionCfg
@@ -39,6 +41,12 @@ from ..semantic.obligations.trace import AbstractTraceState, build_trace
 from ..semantic.obligations.witness import match_witness
 from .producers import Context, _safe_path
 
+_CORE_RESOLUTION_KEYS = {
+    "complete_calls": "complete",
+    "incomplete_calls": "incomplete",
+    "unresolved_calls": "unresolved",
+}
+
 
 @dataclass(frozen=True)
 class Coverage:
@@ -58,6 +66,27 @@ class Coverage:
     # resolution has no bare-name fallback; every call lands in exactly one bucket.
     resolution_counts: dict[str, int] = field(default_factory=dict)
 
+    def to_core(self) -> CoreCoverage:
+        """Native conformance to the detection-core coverage contract (detection-core task 8.1).
+
+        Unknown resolution bucket keys fail loud rather than under-counting calls.
+        """
+        unknown = sorted(key for key in self.resolution_counts if key not in _CORE_RESOLUTION_KEYS)
+        if unknown:
+            raise ValueError(f"unknown resolution bucket keys, refusing to under-count calls: {unknown}")
+        buckets = {name: 0 for name in ("complete", "incomplete", "unresolved")}
+        for key, value in self.resolution_counts.items():
+            buckets[_CORE_RESOLUTION_KEYS[key]] += int(value)
+        return CoreCoverage(
+            entry_points=int(self.entry_points),
+            frameworks_recognised=dict(self.frameworks_recognised),
+            calls=sum(buckets.values()),
+            complete=buckets["complete"],
+            incomplete=buckets["incomplete"],
+            unresolved=buckets["unresolved"],
+            unchecked_reasons=dict(self.unchecked_by_reason),
+        )
+
 
 @dataclass(frozen=True)
 class ProverResult:

src/openultrasast/semantic/obligations/trace.py

@@ -10,6 +10,12 @@ from dataclasses import asdict, dataclass, replace
 from enum import Enum
 from typing import Any
 
+from openultrasast.core.trace import Guard as CoreGuard
+from openultrasast.core.trace import Hop as CoreHop
+from openultrasast.core.trace import TraceRecord
+from openultrasast.core.trace import Witness as CoreWitness
+from openultrasast.core.verdict import Judgment as CoreJudgment
+
 from ..cfg import ControlContext
 from ..ir import FileIR, FunctionIR, trailing_name
 from .abstract_domain import (
@@ -147,6 +153,47 @@ class Trace:
         record: dict[str, Any] = json.loads(json.dumps(payload, sort_keys=True, default=_enum_json_value))
         return record
 
+    def to_core_record(self, *, entry: str, digests: Mapping[str, str], complete: bool = False) -> TraceRecord:
+        """Native conformance to the detection-core trace contract (detection-core task 8.1).
+
+        ``complete`` defaults to False: a Trace carries no per-hop completeness, so the
+        record never asserts resolved call targets it did not see. Guards are mapped from
+        every hop's before-call guards plus the operation-level guards.
+        """
+        state = self.abstract_state
+        if state is None:
+            judgment = CoreJudgment.INDETERMINATE
+            reasons: tuple[str, ...] = ("abstract_state_unset",)
+            witnesses: tuple[CoreWitness, ...] = ()
+            resource = "unknown"
+            authority = "unknown"
+        else:
+            judgment = CoreJudgment(state.path_judgment.judgment.value)
+            completeness = state.path_judgment.completeness
+            reasons = tuple(completeness.reasons) or (("incomplete",) if not completeness.is_complete else ())
+            witnesses = tuple(
+                CoreWitness(kind="witness", provenance=evidence, relation=match.relation.value)
+                for match in state.path_judgment.matches
+                for evidence in match.evidence
+            )
+            resource = ",".join(state.resource.values) or "unknown"
+            authority = ",".join(state.authority.values) or "unknown"
+        hops = tuple(CoreHop(function=hop.function, call_line=hop.call_line, complete=complete) for hop in self.hops)
+        hop_guards = tuple(guard for hop in self.hops for guard in hop.guards_before_call)
+        guards = tuple(CoreGuard(kind=guard.kind, effect=guard.effect) for guard in (*hop_guards, *self.guards_after))
+        return TraceRecord(
+            entry=entry,
+            hops=hops,
+            operation_kind=self.operation_kind,
+            resource=resource,
+            authority=authority,
+            witnesses=witnesses,
+            guards=guards,
+            completeness_reasons=reasons,
+            judgment=judgment,
+            digests=dict(digests),
+        )
+
     def text(self) -> str:
         """Emit entry, hops, operation, later guards, verdict and, when present, the abstract state."""
         record = self.record()