feat(ADAPT-2): Targeted Evidence Probe + verification-driven escalation

Handoff 2026-09-10 §20.D–E, canaries 2–4:

- evidence_probe.py: bounded uncertainty-resolution operation OUTSIDE the
  canonical stage chain. Fixed 5-key schema (QUESTION/SOURCE/ANSWER/
  IMPLICATION/STATUS); GROUNDED requires a cited source; STILL_UNKNOWN/
  CONTRADICTED are honest findings, not schema violations. Executor-owned
  artifact path (ADAPT-9); no stdout rescue. Probe budget caps recursion
  (MAX_PROBES_PER_TASK=2) — unresolved unknowns escalate, never loop.
- Dispatch wiring: material unknowns on non-floored tasks are resolved by
  bounded probes BEFORE the route becomes final. GROUNDED probes fold into
  sufficiency inputs with citation provenance; failures record an
  ESCALATION route event (previous_route/trigger/evidence) and enter the
  canonical chain as RESEARCHED_CHANGE.
- Real probing is OPT-IN (MIMO_PROBE_ENABLED=1 or injected runner):
  dispatch stays deterministic by default — a worker call from the
  dispatch path is minutes of cognition, never an implicit side effect.
- escalate_on_verification_failure(): production hook for the local lane;
  refuses bare 'test failed' evidence (must name the material uncertainty).
- ROUTED_LOCAL_FIX contract message names the exact escalation call.
- Canaries: grounded probe returns to local lane without deep research;
  unresolved probe escalates with recorded transition; no probe recursion;
  verification-driven escalation persisted; bare evidence refused.
- Suite: 1185 passed (3 pre-existing env failures excluded)

Refs #291

scripts/adaptive_router.py

@@ -38,6 +38,34 @@ class FloorViolation(Exception):
     """A downgrade was attempted on a route protected by the research floor."""
 
 
+# Flag detection: resolved relative to THIS file (legacy tests load this
+# module standalone via importlib, with no package context on sys.path).
+def _load_detect_flags():
+    try:
+        from research_risk_flags import detect_flags
+        return detect_flags
+    except ModuleNotFoundError:
+        import importlib.util as _ilu
+        _spec = _ilu.spec_from_file_location(
+            "research_risk_flags",
+            os.path.join(os.path.dirname(os.path.abspath(__file__)),
+                         "research_risk_flags.py"))
+        _mod = _ilu.module_from_spec(_spec)
+        sys.modules["research_risk_flags"] = _mod
+        _spec.loader.exec_module(_mod)
+        return _mod.detect_flags
+
+
+_detect_flags = _load_detect_flags()
+
+
+def body_has_floor_trigger(body_lower: str) -> bool:
+    """Public floor predicate over raw task text (used by dispatch routing
+    before classification, e.g. to decide probe eligibility)."""
+    risk, uncertainty = _detect_flags(body_lower)
+    return _floor_applies_from_flags(risk, uncertainty, body_lower)
+
+
 @dataclass
 class SufficiencyInputs:
     """Evidence-sufficiency assessment (handoff §4). Each field is a source-
@@ -114,21 +142,7 @@ def classify_route(
     prov.update(sufficiency.provenance)
 
     # Flag detection stays for floor evaluation, but is no longer the router.
-    # Resolved relative to THIS file (legacy tests load this module standalone
-    # via importlib, with no package context and no scripts/ on sys.path).
-    try:
-        from research_risk_flags import detect_flags
-    except ModuleNotFoundError:
-        import importlib.util as _ilu
-        _spec = _ilu.spec_from_file_location(
-            "research_risk_flags",
-            os.path.join(os.path.dirname(os.path.abspath(__file__)),
-                         "research_risk_flags.py"))
-        _mod = _ilu.module_from_spec(_spec)
-        sys.modules["research_risk_flags"] = _mod
-        _spec.loader.exec_module(_mod)
-        detect_flags = _mod.detect_flags
-    risk, uncertainty = detect_flags(body_lower)
+    risk, uncertainty = _detect_flags(body_lower)
 
     unknowns = sufficiency.unknowns()
     if sufficiency.provenance.get("legacy_inputs"):
@@ -252,6 +266,30 @@ def downgrade_route(decision: RouteDecision, new_route: str, trigger: str,
     return down
 
 
+def escalate_on_verification_failure(decision: RouteDecision,
+                                     failure_evidence: str,
+                                     log_path: str) -> RouteDecision:
+    """Production hook for the local lane (handoff §7, canary 4).
+
+    A locally-routed task whose verification unexpectedly fails calls this
+    with the failure evidence. The transition is recorded in the same
+    append-only authority; the caller then runs the Targeted Evidence Probe
+    or deep research on the revealed uncertainty.
+
+    The evidence must name the material uncertainty — a bare "test failed"
+    does not establish that more cognition is required.
+    """
+    if not failure_evidence or len(failure_evidence.strip()) < 20:
+        raise ValueError(
+            "verification failure evidence must establish the material "
+            "uncertainty (what external/architectural fact is in doubt), "
+            "not merely report that a check failed")
+    return escalate_route(
+        decision, new_route="RESEARCHED_CHANGE",
+        trigger="verification_failure_external_uncertainty",
+        evidence=failure_evidence, log_path=log_path)
+
+
 def record_transition(kind: str, decision: RouteDecision, path: str) -> None:
     os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
     with open(path, "a") as f:

scripts/evidence_probe.py

@@ -0,0 +1,198 @@
+#!/usr/bin/env python3
+"""evidence_probe.py — Targeted Evidence Probe (handoff 2026-09-10 §5).
+
+A bounded uncertainty-resolution operation, OUTSIDE the canonical deep-research
+stage chain. It is not another research pipeline: it establishes ONE material
+unknown cheaply enough to return to normal execution.
+
+Contract:
+  QUESTION    — the exact unknown blocking the solution
+  SOURCE      — authoritative/current evidence actually consulted
+  ANSWER      — what the evidence establishes
+  IMPLICATION — what this changes about the implementation
+  STATUS      — GROUNDED | STILL_UNKNOWN | CONTRADICTED
+
+Discipline:
+  - The probe researches the uncertainty, not the entire task.
+  - STILL_UNKNOWN / CONTRADICTED escalate (route event) — they never cause
+    repeated probe loops (no recursion; one probe per unknown per attempt).
+  - The artifact file is the sole authority (ADAPT-9); the declared absolute
+    path is injected into the worker prompt, never derived by the worker.
+"""
+from __future__ import annotations
+
+import json
+import os
+import time
+from dataclasses import dataclass, field
+
+PROBE_STATUSES = {"GROUNDED", "STILL_UNKNOWN", "CONTRADICTED"}
+PROBE_FIELDS = ("question", "source", "answer", "implication", "status")
+
+# No recursive probing: a task may resolve at most this many unknowns via
+# probes before escalation to deep research is mandatory.
+MAX_PROBES_PER_TASK = 2
+
+
+class ProbeBudgetExceeded(Exception):
+    """Raised when probing would recurse past the bounded budget."""
+
+
+@dataclass
+class ProbeResult:
+    question: str
+    source: str
+    answer: str
+    implication: str
+    status: str
+    probe_id: str = ""
+    decided_at: str = ""
+    model: str = "UNKNOWN"
+
+    def to_dict(self) -> dict:
+        return {
+            "question": self.question, "source": self.source,
+            "answer": self.answer, "implication": self.implication,
+            "status": self.status, "probe_id": self.probe_id,
+            "decided_at": self.decided_at, "model": self.model,
+        }
+
+
+def build_probe_prompt(question: str, artifact_path: str) -> str:
+    """Bounded prompt: research THE UNKNOWN, write the 5-key JSON file."""
+    return f"""TARGETED EVIDENCE PROBE (bounded — not full research)
+
+You are resolving exactly ONE material unknown. Do NOT research the wider
+task, do NOT design anything, do NOT modify code.
+
+UNKNOWN:
+{question}
+
+Find authoritative, CURRENT evidence that resolves this unknown (official
+docs, source code of the dependency, release notes, standards). One or two
+focused lookups are enough — this is a probe, not a survey.
+
+OUTPUT: write JSON to
+  {artifact_path}  (EXACT path — write to this exact absolute path, no other)
+with EXACTLY these keys:
+  question:    the unknown you were asked to resolve (copy verbatim)
+  source:      what authoritative evidence you consulted (URLs/paths)
+  answer:      what the evidence establishes
+  implication: what this changes about the implementation
+  status:      "GROUNDED" (unknown resolved with cited evidence)
+             | "STILL_UNKNOWN" (evidence not found/insufficient)
+             | "CONTRADICTED" (evidence contradicts the task's assumption)
+
+MUST NOT: modify any file except the artifact path; continue into further
+research after the unknown is resolved; guess without a source."""
+
+
+def validate_probe(artifact: dict) -> list[str]:
+    """Schema gate. Returns gate errors (empty = pass)."""
+    errors = []
+    if not isinstance(artifact, dict):
+        return ["probe artifact is not a JSON object"]
+    for f in PROBE_FIELDS:
+        if f not in artifact:
+            errors.append(f"missing required field: {f}")
+        elif not isinstance(artifact[f], str):
+            errors.append(f"field {f} must be a string")
+        elif f != "source" and not artifact[f].strip():
+            errors.append(f"field {f} must be a non-empty string")
+    status = artifact.get("status")
+    if status is not None and status not in PROBE_STATUSES:
+        errors.append(f"status must be one of {sorted(PROBE_STATUSES)}, got {status!r}")
+    # A GROUNDED probe must cite its source (that is what makes it grounded).
+    # STILL_UNKNOWN / CONTRADICTED may have an empty source — no evidence
+    # found is the honest finding, not a schema violation.
+    if status == "GROUNDED" and len(artifact.get("source", "")) < 8:
+        errors.append("GROUNDED probe requires a concrete source citation")
+    extra = set(artifact) - set(PROBE_FIELDS) - {"probe_id", "decided_at", "model"}
+    if extra:
+        errors.append(f"unexpected extra fields: {sorted(extra)}")
+    return errors
+
+
+def count_probes(artifact_dir: str) -> int:
+    """Probes already run for this task (no-recursion budget)."""
+    path = os.path.join(artifact_dir, "probe.jsonl")
+    if not os.path.exists(path):
+        return 0
+    with open(path) as f:
+        return sum(1 for line in f if line.strip())
+
+
+def check_budget(artifact_dir: str) -> None:
+    if count_probes(artifact_dir) >= MAX_PROBES_PER_TASK:
+        raise ProbeBudgetExceeded(
+            f"probe budget exhausted ({MAX_PROBES_PER_TASK}); "
+            "unresolved unknowns must escalate to deep research")
+
+
+def record_probe(result: ProbeResult, artifact_dir: str) -> str:
+    """Persist the probe (probe.json = artifact, probe.jsonl = append-only)."""
+    os.makedirs(artifact_dir, exist_ok=True)
+    with open(os.path.join(artifact_dir, "probe.json"), "w") as f:
+        json.dump(result.to_dict(), f, indent=2)
+    with open(os.path.join(artifact_dir, "probe.jsonl"), "a") as f:
+        f.write(json.dumps(result.to_dict()) + "\n")
+    return os.path.join(artifact_dir, "probe.json")
+
+
+def run_probe(question: str, artifact_dir: str, runner=None,
+              model: str = "UNKNOWN") -> ProbeResult:
+    """Run one bounded probe. `runner(prompt, artifact_path)` is injectable
+    for tests; the default invokes the local MiMo worker (same channel as
+    canonical stages). The probe does NOT touch the research task tree."""
+    check_budget(artifact_dir)
+    artifact_path = os.path.abspath(
+        os.path.join(artifact_dir, "evidence_probe.json"))
+
+    if runner is None:
+        runner = _default_runner
+
+    runner(build_probe_prompt(question, artifact_path), artifact_path)
+
+    if not os.path.exists(artifact_path):
+        _missing_artifact()
+
+    with open(artifact_path) as f:
+        artifact = json.load(f)
+
+    errors = validate_probe(artifact)
+    if errors:
+        raise ValueError("probe gate failed: " + "; ".join(errors))
+
+    result = ProbeResult(
+        question=artifact["question"], source=artifact["source"],
+        answer=artifact["answer"], implication=artifact["implication"],
+        status=artifact["status"],
+        probe_id=f"probe-{int(time.time())}",
+        decided_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        model=model or os.environ.get("MIMO_MODEL", "UNKNOWN"),
+    )
+    record_probe(result, artifact_dir)
+    return result
+
+
+def _missing_artifact() -> ProbeResult:
+    raise ValueError(
+        "probe failed: worker did not write the declared artifact file "
+        "(ADAPT-9 authority — no stdout rescue for probes)")
+
+
+def _default_runner(prompt: str, artifact_path: str) -> None:
+    """Same worker channel as canonical stages: local MiMo, JSON format."""
+    import subprocess
+    mimo_bin = os.environ.get("MIMO_BIN", "/var/lib/gitea-desktop-runner/bin/mimo")
+    mimo_model = os.environ.get("MIMO_MODEL", "Verboo Code/glm-5.3-flash")
+    cmd = [mimo_bin, "run", "--agent", os.environ.get("MIMO_AGENT", "build"),
+           "-m", mimo_model, "--dangerously-skip-permissions",
+           "--format", "json", prompt]
+    result = subprocess.run(cmd, capture_output=True, text=True, timeout=900)
+    # stdout is observability only; the artifact file is the authority.
+    with open(os.path.join(os.path.dirname(artifact_path),
+                           "evidence_probe.stdout.log"), "w") as f:
+        f.write(result.stdout or "")
+    if result.returncode != 0:
+        raise ValueError(f"probe worker exited {result.returncode}")

scripts/research_stage_executor.py

@@ -50,13 +50,15 @@ class StageExecutor:
     """Executes exactly one research stage. No self-advancing."""
 
     def __init__(self, issue: int, artifact_dir: str, round_no: int = 1,
-                 allow_stdout_fallback: bool = False):
+                 allow_stdout_fallback: bool = False, probe_runner=None):
         self.tree = ResearchTaskTree(issue=issue, round_no=round_no, artifact_dir=artifact_dir)
         self.issue = issue
         self.artifact_dir = artifact_dir
         self.round_no = round_no
         self.allow_stdout_fallback = allow_stdout_fallback
         self.event_log = []
+        # Injectable probe runner for tests; None = real MiMo worker channel.
+        self.probe_runner = probe_runner
 
     def execute(self, event: dict) -> dict:
         """Execute exactly one stage based on the ResearchStageReady event."""
@@ -322,20 +324,120 @@ class StageExecutor:
         )
 
         route_log = os.path.join(self.artifact_dir, "route.jsonl")
+
+        # Targeted Evidence Probe (handoff §3/§5, wired per §20.D): material
+        # unknowns on a non-floored, non-novel task are resolved by BOUNDED
+        # probes before the route becomes final — cheapest defensible
+        # cognition first. STILL_UNKNOWN/CONTRADICTED escalate via recorded
+        # route events; no probe recursion (budget-capped).
+        floor = adaptive_router.body_has_floor_trigger((title + "\n" + body).lower())
+        probe_eligible = list(sufficiency.unknowns())
+        if (sufficiency.local_authority == "NOT_GROUNDED"
+                and sufficiency.scope_novelty == "KNOWN_PATTERN"):
+            probe_eligible.append("local_authority")
+
+        probes_run: list[dict] = []
+        # Real probing is OPT-IN (MIMO_PROBE_ENABLED=1): dispatch must stay
+        # deterministic by default — a real worker call from the dispatch
+        # path is minutes of cognition, never an implicit side effect.
+        # Without the flag, unresolved unknowns escalate directly to deep
+        # research (correct, just not probe-optimized).
+        probe_enabled = (self.probe_runner is not None
+                         or os.environ.get("MIMO_PROBE_ENABLED") == "1")
+        if probe_eligible and probe_enabled and not floor and not decision.floor:
+            from evidence_probe import run_probe, MAX_PROBES_PER_TASK, ProbeBudgetExceeded
+            if len(probe_eligible) <= MAX_PROBES_PER_TASK:
+                import dataclasses
+                pre_sufficiency = dataclasses.replace(sufficiency)
+                any_unresolved = False
+                for u in probe_eligible:
+                    try:
+                        probe = run_probe(
+                            self._unknown_question(u, title, body),
+                            self.artifact_dir,
+                            runner=self.probe_runner,
+                            model=os.environ.get("MIMO_MODEL", "UNKNOWN"))
+                    except (ValueError, ProbeBudgetExceeded) as pe:
+                        # Probe failure = unresolved unknown (fail loud, no loop)
+                        probes_run.append({"unknown": u, "status": "FAILED",
+                                           "error": str(pe)[:200]})
+                        any_unresolved = True
+                        continue
+                    probes_run.append(probe.to_dict())
+                    if probe.status == "GROUNDED":
+                        self._apply_grounded_probe(sufficiency, u, probe)
+                    else:
+                        any_unresolved = True
+                if probes_run:
+                    decision = adaptive_router.classify_route(
+                        issue_body=f"{title}\n{body}",
+                        has_local_repro=(sufficiency.local_authority == "GROUNDED"),
+                        owning_authority_known=True,
+                        testable_locally=(sufficiency.verifiability == "TESTABLE"),
+                        narrow_and_reversible=(sufficiency.scope_novelty == "KNOWN_PATTERN"),
+                        sufficiency=sufficiency,
+                        pipeline_sha=self._pipeline_sha())
+                    provisional_route = decision.route
+                    if any_unresolved and provisional_route != "RESEARCHED_CHANGE":
+                        # Recorded escalation (canary 3): probe evidence,
+                        # not vibes, moved the route. The pre-probe route is
+                        # computed from the ORIGINAL sufficiency snapshot so
+                        # the audit trail shows what the evidence at hand
+                        # actually supported.
+                        pre = adaptive_router.classify_route(
+                            issue_body=f"{title}\n{body}",
+                            has_local_repro=(pre_sufficiency.local_authority == "GROUNDED"),
+                            owning_authority_known=True,
+                            testable_locally=(pre_sufficiency.verifiability == "TESTABLE"),
+                            narrow_and_reversible=(
+                                pre_sufficiency.scope_novelty == "KNOWN_PATTERN"),
+                            sufficiency=pre_sufficiency,
+                            pipeline_sha=self._pipeline_sha())
+                        # Recompute the pre-probe route honestly from the
+                        # ORIGINAL sufficiency (pre-probe copy) for the trail.
+                        adaptive_router.record_route_decision(pre, route_log)
+                        decision = adaptive_router.escalate_route(
+                            pre, new_route="RESEARCHED_CHANGE",
+                            trigger="evidence_probe_unresolved",
+                            evidence="; ".join(
+                                f"{p.get('unknown') or p.get('question','?')[:40]}"
+                                f"={p.get('status')}" for p in probes_run),
+                            log_path=route_log)
+                        self.tree.state["probes"] = probes_run
+                        self.tree.save()
+                        return None  # enter canonical chain as RESEARCHED_CHANGE
+                    # All probes grounded: record route with probe provenance
+                    adaptive_router.record_route_decision(decision, route_log)
+                    if decision.route == "LOCAL_FIX":
+                        return self._terminal_local_route(decision, event, route_log,
+                                                          probes_run)
+                    return None  # STANDARD_CHANGE proceeds into chain
+            else:
+                # Too many unknowns for bounded probing → deep research floor
+                decision = adaptive_router.classify_route(
+                    issue_body=f"{title}\n{body}",
+                    has_local_repro=False, owning_authority_known=True,
+                    testable_locally=False, narrow_and_reversible=False,
+                    sufficiency=sufficiency, pipeline_sha=self._pipeline_sha())
+
         adaptive_router.record_route_decision(decision, route_log)
 
         if decision.route != "LOCAL_FIX":
             return None  # proceed into the canonical chain
 
-        # LOCAL_FIX: terminate the research chain. The task is handed to the
-        # local implementation/verification lane. The tree records the
-        # terminal route; no research stage ever becomes READY.
+        return self._terminal_local_route(decision, event, route_log, probes_run)
+
+    def _terminal_local_route(self, decision, event, route_log, probes_run) -> dict:
+        """LOCAL_FIX: terminate the research chain. The task is handed to the
+        local implementation/verification lane. The tree records the
+        terminal route; no research stage ever becomes READY."""
         self.tree.state["status"] = "ROUTED_LOCAL_FIX"
         self.tree.state["route"] = {
             "route": "LOCAL_FIX",
             "decided_at": decision.decided_at,
             "reasons": decision.reasons,
             "unknowns": decision.unknowns,
+            "probes": probes_run,
         }
         self.tree.state["stage_history"].append({
             "stage": "ISSUE", "round": self.round_no,
@@ -350,15 +452,52 @@ class StageExecutor:
             "event_id": event.get("event_id"),
             "reasons": decision.reasons,
             "unknowns": decision.unknowns,
+            "probes": probes_run,
             "implementation_lane": "local_fix",
             "route_log": route_log,
             "message": (
                 "LOCAL_FIX route: evidence sufficiency established locally — "
                 "fixed-depth research pipeline not entered. Proceed to local "
-                "implementation + verification; escalate on verification "
-                "failure (verification-driven escalation)."),
+                "implementation + verification. On unexpected verification "
+                "failure, call "
+                "adaptive_router.escalate_on_verification_failure(decision, "
+                "failure_evidence, route_log) — the recorded transition routes "
+                "the task to the Targeted Evidence Probe or deep research."),
         }
 
+    @staticmethod
+    def _unknown_question(field_name: str, title: str, body: str) -> str:
+        """Deterministic question templates per unknown type (handoff §5:
+        the probe researches the uncertainty, not the entire task)."""
+        if field_name == "verifiability":
+            return (f"For the task '{title}', what existing test, runtime check "
+                    f"or static check can conclusively verify the change? Identify "
+                    f"concrete test paths/commands in this repository, or state "
+                    f"that none exist.")
+        if field_name == "local_authority":
+            return (f"Is the behaviour required by '{title}' fully determined by "
+                    f"this repository's own source code and existing tests, or "
+                    f"does it depend on external/runtime behaviour that must be "
+                    f"verified separately? Cite the decisive evidence.")
+        if field_name == "version_freshness_risk":
+            return (f"For '{title}': does the relevant external dependency's "
+                    f"CURRENT version still guarantee the behaviour the task "
+                    f"assumes? Cite the version/changelog evidence.")
+        return f"What evidence resolves the unknown '{field_name}' in '{title}'?"
+
+    @staticmethod
+    def _apply_grounded_probe(sufficiency, field_name: str, probe) -> None:
+        """Fold a GROUNDED probe into the sufficiency inputs (provenance
+        carries the citation — never manufactured certainty)."""
+        if field_name == "verifiability":
+            sufficiency.verifiability = "TESTABLE"
+        elif field_name == "local_authority":
+            sufficiency.local_authority = "GROUNDED"
+        elif field_name == "version_freshness_risk":
+            sufficiency.version_freshness_risk = "NONE"
+        sufficiency.provenance[field_name] = (
+            f"evidence probe GROUNDED: {probe.source[:100]}")
+
     @staticmethod
     def _detect_external_refs(body: str) -> str:
         """Observable external-reference detection (handoff §2: provenance

tests/test_adaptive_router_wiring.py

@@ -137,6 +137,43 @@ class Canary4VerificationEscalation(unittest.TestCase):
             self.assertIn("evidence", lines[1])
 
 
+class Canary4ProductionVerificationEscalation(unittest.TestCase):
+    """The local lane's production escalation hook: verification failure
+    with named material uncertainty → persisted route transition."""
+
+    def test_escalate_on_verification_failure(self):
+        from adaptive_router import escalate_on_verification_failure
+        with tempfile.TemporaryDirectory() as tmp:
+            log = Path(tmp) / "route.jsonl"
+            d0 = adaptive_router.classify_route(
+                "Fix log level constant", has_local_repro=True,
+                owning_authority_known=True, testable_locally=True,
+                narrow_and_reversible=True)
+            self.assertEqual(d0.route, "LOCAL_FIX")
+            adaptive_router.record_route_decision(d0, str(log))
+
+            d1 = escalate_on_verification_failure(
+                d0,
+                "verify() failed: behaviour differs across platformdirs "
+                "versions — the assumed API contract does not hold on v4",
+                str(log))
+            self.assertEqual(d1.route, "RESEARCHED_CHANGE")
+            self.assertEqual(d1.trigger,
+                             "verification_failure_external_uncertainty")
+            events = [json.loads(l) for l in log.read_text().strip().splitlines()]
+            self.assertEqual([e["kind"] for e in events],
+                             ["INITIAL_ROUTE", "ESCALATION"])
+
+    def test_bare_failure_evidence_is_refused(self):
+        from adaptive_router import escalate_on_verification_failure
+        d0 = adaptive_router.classify_route(
+            "Fix typo", has_local_repro=True, owning_authority_known=True,
+            testable_locally=True, narrow_and_reversible=True)
+        with self.assertRaises(ValueError):
+            escalate_on_verification_failure(
+                d0, "test failed", "/tmp/unused-route.jsonl")
+
+
 class Canary5ResearchFloor(unittest.TestCase):
     """The floor is non-downgradable: efficiency arguments cannot bypass it."""
 

tests/test_evidence_probe.py

@@ -0,0 +1,166 @@
+#!/usr/bin/env python3
+"""tests/test_evidence_probe.py — Targeted Evidence Probe canaries (§20.D).
+
+Canary 2: otherwise-local task → material external unknown → probe GROUNDED
+          → returns to implementation path, does NOT launch full research.
+Canary 3: probe STILL_UNKNOWN / CONTRADICTED → recorded escalation →
+          RESEARCHED_CHANGE (no probe loops).
+Plus module gates: schema validation, budget (no recursion), file authority.
+"""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+SCRIPTS = os.path.join(os.path.dirname(__file__), "..", "scripts")
+sys.path.insert(0, SCRIPTS)
+
+import adaptive_router  # noqa: E402
+import evidence_probe  # noqa: E402
+from research_stage_executor import StageExecutor, make_event_id  # noqa: E402
+
+
+def _issue_event(issue_no, stage="ISSUE", attempt=1):
+    return {"event_type": "ResearchStageReady",
+            "event_id": make_event_id(issue_no, f"research-{issue_no}", 1, stage, attempt),
+            "issue": issue_no, "round": 1, "stage": stage, "attempt": attempt}
+
+
+# An otherwise-local task with ONE material external unknown: the issue
+# references an external library's behaviour but is a known-pattern change.
+AMBIGUOUS_ISSUE = {
+    "number": 9100,
+    "title": "Handle platformdirs config-dir resolution on new installs",
+    "body": ("Wire the config dir through the platformdirs library package. "
+             "Reference: https://platformdirs.readthedocs.io. Depends on "
+             "current behaviour of the library's version. Existing test: "
+             "tests/test_config_paths.py covers the local wiring."),
+    "labels": [],
+}
+
+GROUNDED_PROBE = {
+    "question": "probe",
+    "source": "https://platformdirs.readthedocs.io/changelog.html",
+    "answer": "Current version guarantees the assumed behaviour.",
+    "implication": "Local wiring is sufficient; no version guard needed.",
+    "status": "GROUNDED",
+}
+
+UNKNOWN_PROBE = {
+    "question": "probe",
+    "source": "",
+    "answer": "Could not find authoritative evidence.",
+    "implication": "Uncertainty remains.",
+    "status": "STILL_UNKNOWN",
+}
+
+
+def _runner_returning(payload):
+    def runner(prompt, artifact_path):
+        with open(artifact_path, "w") as f:
+            json.dump(dict(payload, question=payload.get("question", "probe")), f)
+    return runner
+
+
+def _make_executor(tmpdir, issue_data, runner):
+    ex = StageExecutor(issue=issue_data["number"], artifact_dir=tmpdir, round_no=1,
+                       probe_runner=runner)
+    ex._fetch_issue = lambda: issue_data
+    return ex
+
+
+class Canary2ProbeGroundedReturnsToLocal(unittest.TestCase):
+    def test_grounded_probe_resolves_unknown_without_deep_research(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            def runner(prompt, artifact_path):
+                # The prompt must name the EXACT artifact path (ADAPT-9)
+                self.assertIn(artifact_path, prompt)
+                with open(artifact_path, "w") as f:
+                    json.dump(dict(GROUNDED_PROBE), f)
+            ex = _make_executor(tmp, AMBIGUOUS_ISSUE, runner)
+            result = ex.execute(_issue_event(9100))
+
+            # The route must NOT be deep research: the probe grounded the
+            # unknown and the task returns to the local implementation path.
+            self.assertEqual(result["status"], "ROUTED_LOCAL_FIX",
+                             f"grounded probe should return to local lane: {result}")
+            # Exactly ONE probe ran — no loops.
+            self.assertEqual(len(result["probes"]), 1)
+            self.assertEqual(result["probes"][0]["status"], "GROUNDED")
+            # Probe persisted with citations
+            probe = json.loads((Path(tmp) / "probe.json").read_text())
+            self.assertEqual(probe["status"], "GROUNDED")
+            self.assertIn("https://", probe["source"])
+            # Route log: single INITIAL_ROUTE (no escalation event)
+            events = [json.loads(l) for l in
+                      (Path(tmp) / "route.jsonl").read_text().strip().splitlines()]
+            self.assertEqual([e["kind"] for e in events], ["INITIAL_ROUTE"])
+            self.assertEqual(events[0]["route"], "LOCAL_FIX")
+            self.assertIn("evidence probe GROUNDED",
+                          events[0].get("provenance", {}).get("local_authority", ""))
+
+
+class Canary3UnresolvedEscalates(unittest.TestCase):
+    def test_still_unknown_escalates_to_researched_change(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            ex = _make_executor(tmp, AMBIGUOUS_ISSUE,
+                                _runner_returning(dict(UNKNOWN_PROBE)))
+            result = ex.execute(_issue_event(9100))
+
+            # Escalation: enters the canonical chain (no terminal local route)
+            self.assertNotEqual(result.get("status"), "ROUTED_LOCAL_FIX")
+            events = [json.loads(l) for l in
+                      (Path(tmp) / "route.jsonl").read_text().strip().splitlines()]
+            kinds = [e["kind"] for e in events]
+            self.assertIn("ESCALATION", kinds)
+            esc = next(e for e in events if e["kind"] == "ESCALATION")
+            self.assertEqual(esc["new_route"], "RESEARCHED_CHANGE")
+            self.assertEqual(esc["trigger"], "evidence_probe_unresolved")
+            self.assertIn("STILL_UNKNOWN", esc["evidence"])
+            # Transition identity fields present
+            for field in ("previous_route", "trigger", "evidence"):
+                self.assertIn(field, esc)
+            # Probe recorded the failure honestly — in the PERSISTED state
+            # (the escalation path returns into the chain; the durable
+            # artifact is the tree, not the ephemeral return value).
+            self.assertEqual(ex.tree.state["probes"][0]["status"],
+                             "STILL_UNKNOWN")
+            probe = json.loads((Path(tmp) / "probe.json").read_text())
+            self.assertEqual(probe["status"], "STILL_UNKNOWN")
+
+    def test_no_probe_recursion_budget(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            for i in range(evidence_probe.MAX_PROBES_PER_TASK):
+                evidence_probe.run_probe(
+                    f"unknown {i}", tmp,
+                    runner=_runner_returning(dict(UNKNOWN_PROBE)))
+            with self.assertRaises(evidence_probe.ProbeBudgetExceeded):
+                evidence_probe.run_probe(
+                    "one too many", tmp,
+                    runner=_runner_returning(dict(UNKNOWN_PROBE)))
+
+
+class ProbeModuleGates(unittest.TestCase):
+    def test_schema_validation(self):
+        bad = {"question": "q", "source": "s", "answer": "a",
+               "implication": "i", "status": "MAYBE"}
+        errors = evidence_probe.validate_probe(bad)
+        self.assertTrue(any("status" in e for e in errors))
+        grounded_no_source = dict(GROUNDED_PROBE, source="x")
+        errors = evidence_probe.validate_probe(grounded_no_source)
+        self.assertTrue(any("source" in e for e in errors))
+        self.assertEqual(evidence_probe.validate_probe(dict(GROUNDED_PROBE)), [])
+
+    def test_missing_artifact_fails_loud(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            def silent_runner(prompt, artifact_path):
+                pass  # worker wrote nothing
+            with self.assertRaises(ValueError) as ctx:
+                evidence_probe.run_probe("q", tmp, runner=silent_runner)
+            self.assertIn("declared artifact", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()