feat: confidence-gated Accept/Refine/Restart at the local_triage boundary (Closes #1339)

evolution/lib/stage_gate.py

@@ -0,0 +1,177 @@
+# -*- coding: utf-8 -*-
+"""Confidence-gated Accept/Refine/Restart branch for stage boundaries (#1339).
+
+Slice B of the AREX loop (#1330).  Slice A (#1338) gave every stage a uniform
+:class:`~evolution.lib.stage_result.StageResult`; this module is the gate that
+*acts* on the confidence carried in it.
+
+Three branches, decided at a stage boundary:
+
+* **Accept**  — ``confidence >= threshold``.  Proceed as normal.
+* **Refine**  — below threshold but the trajectory is recoverable: keep the
+  evidence already gathered and re-investigate only the gaps.
+* **Restart** — below threshold and the trajectory is too noisy to salvage:
+  discard it and reinitialize from the original problem.
+
+On recoverability
+-----------------
+The issue specifies the recoverable-vs-noisy call is "produced by the model
+running the stage".  A deterministic script has no model, so ``decide`` takes an
+explicit ``recoverable`` argument for callers that *do* have a judgement, and
+falls back to a conservative structural proxy when it is ``None``: a result with
+evidence pointers has something concrete to build on and is refinable; a result
+with none is indistinguishable from noise and restarts.
+
+The threshold defaults to 70 — deliberately conservative, per the issue's
+"conservative τ prevents runaway looping" criterion.  Note the interaction with
+:meth:`StageResult.wrap`, which assigns 50 to any result that has evidence but
+no self-assessed confidence: such a result lands in Refine, not Accept, so an
+un-assessed stage is never silently trusted.
+
+Pure Python, no external dependencies, no side effects on import — matching the
+rest of ``evolution/lib``.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from evolution.lib.stage_result import StageResult
+
+__all__ = [
+    "ACCEPT",
+    "REFINE",
+    "RESTART",
+    "DEFAULT_CONFIDENCE_THRESHOLD",
+    "GateDecision",
+    "decide",
+]
+
+ACCEPT = "accept"
+REFINE = "refine"
+RESTART = "restart"
+
+#: Conservative default τ.  See module docstring on the 50-confidence interaction.
+DEFAULT_CONFIDENCE_THRESHOLD = 70
+
+
+@dataclass
+class GateDecision:
+    """The branch taken at one stage boundary, with the reasoning that got there.
+
+    Attributes
+    ----------
+    branch
+        One of :data:`ACCEPT`, :data:`REFINE`, :data:`RESTART`.
+    stage
+        Name of the stage this decision was made for.
+    confidence
+        The confidence that was gated on.
+    threshold
+        The τ it was compared against.
+    reason
+        One line, human-readable — this is what gets logged.
+    retained_evidence
+        Evidence pointers carried forward.  Populated for Accept and Refine;
+        empty for Restart, which by definition discards the trajectory.
+    """
+
+    branch: str
+    stage: str
+    confidence: int
+    threshold: int
+    reason: str
+    retained_evidence: list[str]
+
+    @property
+    def proceeds(self) -> bool:
+        """True when the pipeline may consume the result as-is (Accept only)."""
+        return self.branch == ACCEPT
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return a JSON-serialisable dict representation."""
+        return {
+            "branch": self.branch,
+            "stage": self.stage,
+            "confidence": self.confidence,
+            "threshold": self.threshold,
+            "reason": self.reason,
+            "retained_evidence": list(self.retained_evidence),
+        }
+
+    def log_line(self) -> str:
+        """Single-line log record for the boundary (#1339 success criterion)."""
+        return (
+            f"[stage-gate] {self.stage}: {self.branch.upper()} "
+            f"(confidence={self.confidence}, threshold={self.threshold}) — {self.reason}"
+        )
+
+
+def decide(
+    stage_result: StageResult,
+    *,
+    threshold: int = DEFAULT_CONFIDENCE_THRESHOLD,
+    recoverable: bool | None = None,
+) -> GateDecision:
+    """Choose Accept / Refine / Restart for a completed stage.
+
+    Parameters
+    ----------
+    stage_result
+        The tuple emitted at this boundary (#1338).
+    threshold
+        τ — confidence at or above which the result is accepted.  Clamped to
+        0–100 to match ``StageResult``'s own clamping.
+    recoverable
+        Explicit recoverable-vs-noisy judgement from the model running the
+        stage.  When ``None``, falls back to the structural proxy described in
+        the module docstring (evidence present ⇒ recoverable).
+
+    Returns
+    -------
+    GateDecision
+        Always returned — the gate never raises, so a stage boundary cannot be
+        taken down by its own instrumentation.
+    """
+    threshold = max(0, min(100, threshold))
+    confidence = max(0, min(100, int(stage_result.confidence)))
+    evidence = list(stage_result.evidence_pointers)
+    stage = stage_result.stage or "unknown"
+
+    if confidence >= threshold:
+        return GateDecision(
+            branch=ACCEPT,
+            stage=stage,
+            confidence=confidence,
+            threshold=threshold,
+            reason=f"confidence at or above threshold; {len(evidence)} evidence pointer(s)",
+            retained_evidence=evidence,
+        )
+
+    is_recoverable = bool(evidence) if recoverable is None else bool(recoverable)
+
+    if is_recoverable:
+        return GateDecision(
+            branch=REFINE,
+            stage=stage,
+            confidence=confidence,
+            threshold=threshold,
+            reason=(
+                f"below threshold but recoverable — retaining {len(evidence)} "
+                f"evidence pointer(s) and re-investigating the gaps"
+            ),
+            retained_evidence=evidence,
+        )
+
+    return GateDecision(
+        branch=RESTART,
+        stage=stage,
+        confidence=confidence,
+        threshold=threshold,
+        reason=(
+            "below threshold with no salvageable evidence — discarding the "
+            "trajectory and reinitializing from the original problem"
+        ),
+        retained_evidence=[],
+    )

evolution/tests/test_stage_gate.py

@@ -0,0 +1,123 @@
+# -*- coding: utf-8 -*-
+"""Tests for the confidence gate at stage boundaries (issue #1339)."""
+
+from __future__ import annotations
+
+from evolution.lib.stage_gate import (
+    ACCEPT,
+    DEFAULT_CONFIDENCE_THRESHOLD,
+    REFINE,
+    RESTART,
+    GateDecision,
+    decide,
+)
+from evolution.lib.stage_result import StageResult
+
+
+def _result(confidence: int, evidence: list[str] | None = None) -> StageResult:
+    return StageResult(
+        result={"payload": True},
+        evidence_pointers=list(evidence or []),
+        confidence=confidence,
+        stage="local_triage",
+        timestamp="2026-07-27T00:00:00+00:00",
+    )
+
+
+class TestAcceptBranch:
+    def test_at_threshold_accepts(self):
+        d = decide(_result(DEFAULT_CONFIDENCE_THRESHOLD, ["a.json"]))
+        assert d.branch == ACCEPT
+        assert d.proceeds is True
+
+    def test_above_threshold_accepts(self):
+        assert decide(_result(95, ["a.json"])).branch == ACCEPT
+
+    def test_accept_without_evidence_still_accepts(self):
+        """High confidence is sufficient on its own — evidence is not required."""
+        assert decide(_result(90)).branch == ACCEPT
+
+    def test_accept_retains_evidence(self):
+        d = decide(_result(90, ["a.json", "b.json"]))
+        assert d.retained_evidence == ["a.json", "b.json"]
+
+
+class TestRefineBranch:
+    def test_below_threshold_with_evidence_refines(self):
+        d = decide(_result(50, ["a.json"]))
+        assert d.branch == REFINE
+        assert d.proceeds is False
+
+    def test_refine_carries_evidence_forward(self):
+        """Refine preserves reliable findings rather than discarding them."""
+        d = decide(_result(50, ["a.json", "b.json"]))
+        assert d.retained_evidence == ["a.json", "b.json"]
+
+    def test_wrap_default_confidence_lands_in_refine(self):
+        """StageResult.wrap assigns 50 when evidence exists but confidence is
+        unset. That must NOT be silently accepted — an un-assessed stage is not
+        a confident one."""
+        sr = StageResult.wrap(result={}, evidence_pointers=["a.json"], stage="s")
+        assert sr.confidence == 50
+        assert decide(sr).branch == REFINE
+
+    def test_explicit_recoverable_overrides_absent_evidence(self):
+        """A model judging the trajectory recoverable wins over the proxy."""
+        d = decide(_result(20), recoverable=True)
+        assert d.branch == REFINE
+
+
+class TestRestartBranch:
+    def test_below_threshold_without_evidence_restarts(self):
+        d = decide(_result(10))
+        assert d.branch == RESTART
+        assert d.proceeds is False
+
+    def test_restart_discards_evidence(self):
+        d = decide(_result(10))
+        assert d.retained_evidence == []
+
+    def test_explicit_not_recoverable_overrides_present_evidence(self):
+        """A model judging the trajectory too noisy wins over the proxy."""
+        d = decide(_result(20, ["a.json"]), recoverable=False)
+        assert d.branch == RESTART
+        assert d.retained_evidence == []
+
+
+class TestThresholdHandling:
+    def test_custom_threshold_respected(self):
+        assert decide(_result(60, ["a.json"]), threshold=50).branch == ACCEPT
+        assert decide(_result(60, ["a.json"]), threshold=80).branch == REFINE
+
+    def test_threshold_clamped(self):
+        assert decide(_result(100, ["a.json"]), threshold=500).branch == ACCEPT
+        assert decide(_result(0), threshold=-10).branch == ACCEPT
+
+    def test_confidence_clamped(self):
+        sr = StageResult(confidence=1000, stage="s", evidence_pointers=["a"])
+        assert decide(sr).confidence == 100
+
+    def test_default_threshold_is_conservative(self):
+        """A conservative tau is what stops runaway looping (#1339)."""
+        assert DEFAULT_CONFIDENCE_THRESHOLD == 70
+
+
+class TestDecisionRecord:
+    def test_to_dict_round_trip(self):
+        d = decide(_result(50, ["a.json"]))
+        as_dict = d.to_dict()
+        assert as_dict["branch"] == REFINE
+        assert as_dict["stage"] == "local_triage"
+        assert as_dict["retained_evidence"] == ["a.json"]
+
+    def test_log_line_names_stage_and_branch(self):
+        line = decide(_result(50, ["a.json"])).log_line()
+        assert "[stage-gate]" in line
+        assert "local_triage" in line
+        assert "REFINE" in line
+
+    def test_unknown_stage_label(self):
+        assert decide(StageResult(confidence=90)).stage == "unknown"
+
+    def test_is_a_gate_decision(self):
+        assert isinstance(decide(_result(90)), GateDecision)

scripts/evolution_local_triage.py

@@ -24,6 +24,7 @@
 from pathlib import Path
 
 try:
+    from evolution.lib.stage_gate import decide as _gate_decide
     from evolution.lib.stage_result import StageResult
 
     _HAS_STAGE_RESULT = True
@@ -179,6 +180,17 @@ def run_local_triage(evolution_dir: Path) -> dict:
         envelope = stage_result.to_dict()
         envelope.pop("result", None)
         output["stage_result"] = envelope
+
+        # Consume the tuple through the Accept/Refine/Restart gate (#1339).
+        # Advisory at this boundary: local triage is a read-only pre-pass whose
+        # output the analysis stage consumes, so the gate records which branch
+        # the boundary lands in rather than aborting the pass. Confidence 50
+        # (evidence present, no LLM verification) sits below the conservative
+        # default of 70, so a triage run with sidecars lands in `refine` and one
+        # with none lands in `restart` — both surfaced for the next stage to act
+        # on instead of being silently treated as a confident result.
+        decision = _gate_decide(stage_result)
+        output["stage_gate"] = decision.to_dict()
     return output
 
 
@@ -224,6 +236,12 @@ def main(argv: list[str] | None = None) -> int:
     print(f"  Sidecars read: {', '.join(output['sidecars_read'].keys())}")
     print(f"  Selected: {len(output['selected_for_implementation'])} issues")
     print(f"  Effort budget: {output['effort_budget']}")
+    gate = output.get("stage_gate")
+    if gate:
+        print(
+            f"[stage-gate] {gate['stage']}: {gate['branch'].upper()} "
+            f"(confidence={gate['confidence']}, threshold={gate['threshold']}) — {gate['reason']}"
+        )
     return 0