fix: make RegressionRisk detect changed/skipped critical steps (soundness)

CHANGELOG.md

@@ -35,6 +35,20 @@ public API may still change between minor versions.
 
 ### Fixed
 
+- **`RegressionRisk` no longer misses materially-changed or skipped critical steps.** The risk
+  verdict was derived only from `removed`/`reordered` alignment states, but a critical step that
+  was changed or skipped binds to a same-type event (type match alone clears the matcher
+  threshold), is classified `ambiguous`, and so escaped the verdict entirely — a tampered or
+  skipped critical decision reported `RegressionLevel.NONE` on the flagship `strict_audit_v1`
+  profile, the exact regressions the engine exists to catch (SEMANTICS Def 5 / Invariant A/E).
+  The verdict is now derived from the equivalence outcome: a critical step that is removed, bound
+  below the profile's `semantic_threshold`, or reordered relative to another **critical** step
+  fires HIGH. Reorder is computed over critical pairs only, so a benign structural/diagnostic step
+  moving past a stationary critical no longer fires a false HIGH. Alignment states and every
+  calibrated corpus case are unchanged; this only corrects the risk level. Behavior change: runs
+  that previously (incorrectly) reported `NONE` may now report `HIGH` (e.g. a `RegressionGate`
+  with `allow_divergent_steps=True` now fails on a flipped critical decision unless
+  `max_regression_level` is raised).
 - **Nested boolean queries return correct results on the SQLite backend.** The query compiler
   joined compound `AND`/`OR`/`missing_step` members with bare `INTERSECT`/`UNION`/`EXCEPT`; SQLite
   gives those operators equal, left-to-right precedence, so a nested member was silently re-grouped

conformance/vectors/alignment_verdict.json

@@ -135,8 +135,8 @@
         }
       ],
       "expected": {
-        "regression_level": "none",
-        "regression_strength": 1.0,
+        "regression_level": "high",
+        "regression_strength": 0.9,
         "alignment_state_kinds": [
           "ambiguous",
           "exactMatch"

dprovenancekit/alignment_engine.py

@@ -85,54 +85,78 @@ def equivalence(a, b):
             evidence_collector=collector,
         )
 
-        # Regression risk. Two failure modes degrade a critical reasoning step:
-        #   1. Removing it outright.
-        #   2. Reordering it — running critical steps out of their original order can invert
-        #      a dependency (e.g. GenerateInvoice before CreateCustomer). The engine has no
-        #      dependency graph, so this is critical-*order* sensitivity, not true dependency
-        #      inference; it deliberately fires only on CRITICAL steps so that reordering of
-        #      structural/diagnostic steps (the common, benign case) stays NONE.
-        removed_critical = [
-            a
-            for a in alignments
-            if a.state.is_removed
-            and a.base_event is not None
-            and a.base_event.payload.priority == TracePriority.CRITICAL
-        ]
-        reordered_critical = [
-            a
-            for a in alignments
-            if a.state.kind == AlignmentStateKind.REORDERED
-            and a.base_event is not None
-            and a.base_event.payload.priority == TracePriority.CRITICAL
-        ]
-        if removed_critical:
-            critical_types = ", ".join(
-                a.base_event.payload.type_identifier
-                for a in removed_critical
-                if a.base_event is not None
-            )
+        # Regression risk derives from the equivalence OUTCOME, not just the coarse
+        # removed/reordered display states. A critical reasoning step degrades when it is:
+        #   1. Removed outright.
+        #   2. Reordered relative to another CRITICAL step — running critical steps out of
+        #      order can invert a dependency (e.g. GenerateInvoice before CreateCustomer).
+        #      The engine has no dependency graph, so this is critical-*order* sensitivity,
+        #      not true inference; restricting to critical-vs-critical keeps a benign
+        #      structural/diagnostic step moving past a stationary critical at NONE.
+        #   3. Changed beyond equivalence — bound to a same-type event but with a differing
+        #      payload whose match score falls below the profile's semantic_threshold.
+        # Type match alone clears the matcher's bind threshold, so a changed or skipped
+        # critical step is essentially never left REMOVED; it binds and is classified
+        # AMBIGUOUS. Reading only removed/reordered therefore silently missed materially
+        # changed or skipped critical steps (RegressionRisk.none on a tampered decision),
+        # even though the equivalence model had already recorded equivalent=False.
+        threshold = self.configuration.profile.semantic_threshold
+        base_index_by_id = {e.id: i for i, e in enumerate(base_events)}
+        comp_index_by_id_risk = {e.id: i for i, e in enumerate(comp_events)}
+
+        removed_critical_types: list[str] = []
+        changed_critical_types: list[str] = []
+        # (base_idx, comp_idx, type) per matched CRITICAL pair, on the same array-index
+        # basis the interpreter uses for its REORDERED findings, so the verdict can never
+        # disagree with the reorder findings it summarizes.
+        critical_pairs: list[tuple[int, int, str]] = []
+        for a in alignments:
+            b = a.base_event
+            if b is None or b.payload.priority != TracePriority.CRITICAL:
+                continue
+            c = a.comparison_event
+            if c is None:
+                removed_critical_types.append(b.payload.type_identifier)
+                continue
+            # Identical payloads are equivalent by construction; otherwise consult the same
+            # score the matcher/equivalence model used. Below the threshold => not equivalent.
+            if b.payload != c.payload:
+                score, _ = self.configuration.score_match(b, c)
+                if score < threshold:
+                    changed_critical_types.append(b.payload.type_identifier)
+            if b.id in base_index_by_id and c.id in comp_index_by_id_risk:
+                critical_pairs.append(
+                    (base_index_by_id[b.id], comp_index_by_id_risk[c.id], b.payload.type_identifier)
+                )
+
+        reordered_critical_types: list[str] = []
+        for x in critical_pairs:
+            if any(x[0] != y[0] and x[0] < y[0] and x[1] > y[1] for y in critical_pairs):
+                reordered_critical_types.append(x[2])
+
+        if removed_critical_types:
             risk = RegressionRisk(
                 level=RegressionLevel.HIGH,
                 strength=0.95,
-                reasoning=f"Critical reasoning steps removed: {critical_types}",
-            )
-        elif reordered_critical:
-            reordered_types = ", ".join(
-                a.base_event.payload.type_identifier
-                for a in reordered_critical
-                if a.base_event is not None
+                reasoning=f"Critical reasoning steps removed: {', '.join(removed_critical_types)}",
             )
+        elif reordered_critical_types:
             risk = RegressionRisk(
                 level=RegressionLevel.HIGH,
                 strength=1.0,
-                reasoning=f"Critical reasoning steps reordered: {reordered_types}",
+                reasoning=f"Critical reasoning steps reordered: {', '.join(reordered_critical_types)}",
+            )
+        elif changed_critical_types:
+            risk = RegressionRisk(
+                level=RegressionLevel.HIGH,
+                strength=0.9,
+                reasoning=f"Critical reasoning steps changed beyond equivalence: {', '.join(changed_critical_types)}",
             )
         else:
             risk = RegressionRisk(
                 level=RegressionLevel.NONE,
                 strength=1.0,
-                reasoning="No critical steps removed or reordered.",
+                reasoning="No critical steps removed, reordered, or materially changed.",
             )
 
         v_artifacts = None

tests/test_regression_gate.py

@@ -154,7 +154,24 @@ def test_multiple_critical_removals():
 # ── Lenient policies ─────────────────────────────────────────────────────────────
 
 
-def test_allow_divergent_tolerates_a_changed_payload():
+def test_allow_divergent_tolerates_a_benign_noncritical_change():
+    store = InMemoryTraceStore()
+    golden = build_run(store, "golden", retrieved_detail="3 sources")
+    # Only the STRUCTURAL "retrieved" step changed; no critical step is touched.
+    candidate = build_run(store, "candidate", retrieved_detail="4 sources")
+
+    # Strict (default): the changed structural step is an ambiguous divergence → fail.
+    strict = RegressionGate().check(golden, candidate)
+    assert not strict.passed
+    assert "retrieved" in strict.divergent_steps
+
+    # Lenient: a benign non-critical change is tolerated, and severity stays NONE.
+    lenient = RegressionGate(allow_divergent_steps=True).check(golden, candidate)
+    assert lenient.passed
+    assert lenient.regression_level is RegressionLevel.NONE
+
+
+def test_changed_critical_decision_is_high_even_under_lenient():
     store = InMemoryTraceStore()
     golden = build_run(store, "golden", decision="supported")
     flipped = build_run(store, "flipped", decision="refuted")
@@ -164,10 +181,20 @@ def test_allow_divergent_tolerates_a_changed_payload():
     assert not strict.passed
     assert "decided" in strict.divergent_steps
 
-    # Lenient: tolerate per-step changes, gate only on severity → pass (no critical removal).
+    # A CRITICAL decision flipping supported → refuted is a material change beyond
+    # equivalence — a HIGH regression, not a benign per-step divergence.
+    # ``allow_divergent_steps`` relaxes the per-step (display-state) check but NOT the
+    # severity gate, so the lenient gate still fails.
     lenient = RegressionGate(allow_divergent_steps=True).check(golden, flipped)
-    assert lenient.passed
-    assert lenient.regression_level is RegressionLevel.NONE
+    assert not lenient.passed
+    assert lenient.regression_level is RegressionLevel.HIGH
+
+    # Tolerating a changed critical decision requires explicitly raising the severity
+    # ceiling — exactly as it does for a critical removal.
+    tolerant = RegressionGate(
+        allow_divergent_steps=True, max_regression_level=RegressionLevel.HIGH
+    ).check(golden, flipped)
+    assert tolerant.passed
 
 
 def test_lenient_still_catches_critical_removal_unless_level_raised():

tests/test_regression_risk_soundness.py

@@ -0,0 +1,111 @@
+"""Parity port of Swift RegressionRiskSoundnessTests.
+
+Pins that RegressionRisk derives from the equivalence outcome — a critical step that is
+removed, reordered relative to another critical step, or changed beyond equivalence fires
+HIGH — instead of only reading the coarse removed/reordered display states (which
+classified a materially-changed or skipped critical as ``ambiguous`` and reported
+``NONE``). Also pins the opposite-direction false alarm: a benign structural step moving
+past a stationary critical must NOT fire HIGH.
+"""
+
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass
+
+from dprovenancekit import (
+    AlignmentConfiguration,
+    AlignmentProfile,
+    AnyEquivalenceEvaluator,
+    RegressionLevel,
+    TraceAlignmentEngine,
+    TraceEvent,
+    TracePriority,
+    TraceableEvent,
+    TraceRun,
+)
+
+
+@dataclass(frozen=True)
+class Step(TraceableEvent):
+    kind: str
+    body: str = ""
+    critical: bool = True
+
+    @property
+    def type_identifier(self) -> str:
+        return self.kind
+
+    @property
+    def priority(self) -> TracePriority:
+        return TracePriority.CRITICAL if self.critical else TracePriority.STRUCTURAL
+
+
+def _run(specs):
+    run_id = uuid.uuid4()
+    events = [
+        TraceEvent(
+            run_id=run_id,
+            context_id="ctx",
+            engine_name="e",
+            schema_version=1,
+            sequence=seq,
+            span_id=None,
+            parent_span_id=None,
+            payload=payload,
+        )
+        for seq, payload in specs
+    ]
+    return TraceRun(run_id=run_id, context_id="ctx", events=events)
+
+
+def _engine():
+    # Payload-equality evaluator: 1.0 iff payloads are identical, else 0.0.
+    evaluator = AnyEquivalenceEvaluator(
+        evaluator_identifier="eq", evaluator=lambda a, b: 1.0 if a == b else 0.0
+    )
+    return TraceAlignmentEngine(
+        AlignmentConfiguration(AlignmentProfile.strict_audit_v1, evaluator)
+    )
+
+
+def test_materially_changed_critical_step_fires_high():
+    base = _run([(0, Step("authorize_payment", "alice:100"))])
+    comp = _run([(0, Step("authorize_payment", "attacker:1000000"))])
+    result = _engine().align(base, comp)
+    assert result.regression_risk.level is RegressionLevel.HIGH
+    assert result.regression_risk.strength == 0.9
+
+
+def test_skipped_critical_masked_by_same_type_decoy_fires_high():
+    # validate_permissions is skipped; send_receipt is new; both share type "decision".
+    base = _run([(0, Step("decision", "validate_permissions")), (1, Step("decision", "charge_card"))])
+    comp = _run([(0, Step("decision", "send_receipt")), (1, Step("decision", "charge_card"))])
+    result = _engine().align(base, comp)
+    # The critical validate_permissions binds to the decoy but scores below threshold →
+    # changed beyond equivalence → HIGH (not a silent NONE).
+    assert result.regression_risk.level is RegressionLevel.HIGH
+
+
+def test_reordered_critical_steps_fire_high():
+    base = _run([(0, Step("createCustomer", "x")), (1, Step("generateInvoice", "y"))])
+    comp = _run([(0, Step("generateInvoice", "y")), (1, Step("createCustomer", "x"))])
+    result = _engine().align(base, comp)
+    assert result.regression_risk.level is RegressionLevel.HIGH
+    assert result.regression_risk.strength == 1.0
+
+
+def test_benign_structural_reorder_does_not_fire_false_high():
+    # Only the STRUCTURAL log moves; the critical authorize does not move relative to any
+    # other critical step, so there is no regression.
+    base = _run([(0, Step("log", "l", critical=False)), (1, Step("authorize", "a"))])
+    comp = _run([(0, Step("authorize", "a")), (1, Step("log", "l", critical=False))])
+    result = _engine().align(base, comp)
+    assert result.regression_risk.level is RegressionLevel.NONE
+
+
+def test_equivalent_step_is_not_a_regression():
+    base = _run([(0, Step("authorize", "same")), (1, Step("finalize", "ok"))])
+    comp = _run([(0, Step("authorize", "same")), (1, Step("finalize", "ok"))])
+    result = _engine().align(base, comp)
+    assert result.regression_risk.level is RegressionLevel.NONE