chore(release): auto-bump version to 0.66.0 — earned by #325 (feat(ADAPT-2): operational adaptive route authority — production wiring, Evidenc)

Promotion record: #325

README.md

@@ -1,6 +1,6 @@
 # Rune
 
-**Current release: v0.65.0**
+**Current release: v0.66.0**
 
 **Rune is a persistent personal agent designed to accumulate experience, develop reusable competence, and remain recognisably the same agent even when its underlying foundation model changes.**
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "rune"
-version = "0.65.0"
+version = "0.66.0"
 description = "Rune — a stateful coding assistant with persistent memory, orchestration, and TUI"
 requires-python = ">=3.10"
 

requirements-ci.txt

@@ -82,7 +82,7 @@ pygments==2.21.0
     #   pytest
     #   rich
     #   textual
-pyright==1.1.413
+pyright==1.1.414
     # via rune (pyproject.toml)
 pytest==9.1.1
     # via

scripts/adaptive_router.py

@@ -1,60 +1,106 @@
 #!/usr/bin/env python3
-"""adaptive_router.py — canonical adaptive route authority (ADAPT-2, #291).
+"""adaptive_router.py — canonical adaptive route authority (ADAPT-2, #291;
+revised per handoff 2026-09-10 §3–§7: progressive evidence routing).
 
-Deterministic-first route selection. A small bounded semantic classifier may
-arbitrate only when deterministic policy cannot safely decide.
-
-Routes:
-  LOCAL_FIX          — reproducible local defect, known owner, testable locally
+Routes are budgets, not identities:
+  LOCAL_FIX          — evidence-grounded locally; solve, verify, escalate on failure
   STANDARD_CHANGE    — clear change, existing authority, no tech selection
-  RESEARCHED_CHANGE  — external mechanism, uncertain architecture, expensive-to-reverse
+  RESEARCHED_CHANGE  — deep-research floor applies (new subsystem / external
+                       adoption / irreversible / explicit research task)
 
-Routes are reversible: escalation and downgrade are first-class, recorded, never failures.
+Routing authority owns the full lifecycle as append-only route events:
+  INITIAL_ROUTE → (ESCALATION | DOWNGRADE)* → FINAL_ROUTE
+
+Sufficiency inputs carry provenance: every input is either derived from an
+inspectable source (named in `provenance`) or recorded as UNKNOWN. The router
+escalates on UNKNOWN material facts — it never manufactures certainty.
 """
+
 from __future__ import annotations
 
 import json
 import os
-import time
+import sys
 from datetime import datetime, timezone
-from dataclasses import dataclass
-from typing import Any
+from dataclasses import dataclass, field
 
 ROUTES = {"LOCAL_FIX", "STANDARD_CHANGE", "RESEARCHED_CHANGE"}
 
-# Deterministic routing features (from ISSUE + INWARD + codebase inspection)
-RISK_FLAGS = {
-    "new_dependency": "adds an external dependency/service/store",
-    "new_authority": "creates a new authority path or ownership boundary",
-    "architecture_change": "changes how components interact structurally",
-    "data_loss_risk": "could lose or corrupt persistent state",
-    "external_service": "integrates with an external service/API",
-    "security_implications": "touches auth, access control, or secrets",
-    "memory_architecture": "changes memory storage, retrieval, or indexing",
-}
+# Deep-research floor (handoff §5): objectively detectable, deliberately
+# narrow. Ambiguity escalates rather than requiring an exhaustive taxonomy.
+FLOOR_RISK_FLAGS = {"new_authority", "architecture_change", "data_loss_risk", "external_service", "memory_architecture"}
+FLOOR_UNCERTAINTY_FLAGS = {"competing_designs", "expensive_reversal"}
+FLOOR_EXPLICIT_MARKERS = (
+    "new subsystem",
+    "new epic",
+    "major external",
+    "adopt",
+    "replacement",
+    "architecture research",
+)
 
-UNCERTAINTY_FLAGS = {
-    "unknown_mechanism": "mechanism not resolvable from local code",
-    "competing_designs": "multiple materially different viable designs",
-    "expensive_reversal": "decision is expensive/difficult to reverse",
-    "prior_artifact_matters": "prior art could change the design",
-    "benchmark_needed": "correctness claim requires measurement",
-}
 
-# Default escalation/downgrade triggers
-ESCALATION_TRIGGERS = [
-    "unknown_mechanism discovered during implementation",
-    "competing_designs discovered during implementation",
-    "new_dependency needed but not pre-approved",
-    "architecture boundary must change",
-    "data loss risk identified",
-]
+class FloorViolation(Exception):
+    """A downgrade was attempted on a route protected by the research floor."""
 
-DOWNGRADE_TRIGGERS = [
-    "INWARD proves existing mechanism is already adequate",
-    "external evidence adds nothing beyond local code analysis",
-    "benchmark shows no measurable difference between options",
-]
+
+# 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-
+    backed value or "UNKNOWN" — never a model feeling."""
+
+    local_authority: str = "UNKNOWN"  # GROUNDED | NOT_GROUNDED | UNKNOWN
+    external_dependency: str = "NONE"  # NONE | PRESENT | UNKNOWN
+    version_freshness_risk: str = "NONE"  # NONE | RISK | UNKNOWN
+    verifiability: str = "UNKNOWN"  # TESTABLE | STATIC | RUNTIME | NONE | UNKNOWN
+    scope_novelty: str = "KNOWN_PATTERN"  # KNOWN_PATTERN | NEW_SUBSYSTEM | UNKNOWN
+    provenance: dict = field(default_factory=dict)  # field -> "source description"
+
+    def unknowns(self) -> list[str]:
+        vals = [
+            self.local_authority,
+            self.external_dependency,
+            self.version_freshness_risk,
+            self.verifiability,
+            self.scope_novelty,
+        ]
+        return [
+            f
+            for f, v in zip(
+                ["local_authority", "external_dependency", "version_freshness_risk", "verifiability", "scope_novelty"],
+                vals,
+            )
+            if v == "UNKNOWN"
+        ]
 
 
 @dataclass
@@ -63,105 +109,318 @@ class RouteDecision:
     reasons: list[str]
     risk_flags: list[str]
     uncertainty_flags: list[str]
-    escalation_conditions: list[str]
-    downgrade_conditions: list[str]
-    confidence: str  # HIGH | MEDIUM | LOW
-    decided_by: str  # "deterministic" | "semantic_arbiter"
-    decided_at: str
-    task_description: str
-    baseline_sha: str
+    sufficiency: SufficiencyInputs | None = None
+    unknowns: list[str] = field(default_factory=list)
+    confidence: str = "MEDIUM"  # HIGH | MEDIUM | LOW
+    decided_by: str = "deterministic"
+    decided_at: str = ""
+    task_description: str = ""
+    baseline_sha: str = "UNKNOWN"
+    previous_route: str | None = None  # set on transitions
+    trigger: str | None = None  # transition trigger
+    evidence: str | None = None  # transition evidence
+    floor: bool = False  # deep-research floor applied
 
 
 def classify_route(
     issue_body: str,
-    inward_claims: list[dict] | None = None,
     has_local_repro: bool = False,
     owning_authority_known: bool = False,
     testable_locally: bool = False,
     narrow_and_reversible: bool = False,
+    sufficiency: SufficiencyInputs | None = None,
     pipeline_sha: str = "UNKNOWN",
 ) -> RouteDecision:
-    """Deterministic-first route classification.
+    """Deterministic-first route classification with traceable provenance.
 
-    Route by uncertainty and risk, not "bug vs feature".
-
-    LOCAL_FIX: all materially relevant conditions hold
-    STANDARD_CHANGE: clear change, existing authority, no tech selection
-    RESEARCHED_CHANGE: one or more high-value research triggers exist
+    Backward-compatible: legacy boolean inputs still work and are recorded
+    in provenance. Sufficiency inputs, when supplied, take precedence and
+    their UNKNOWNs force escalation (never manufactured certainty).
     """
-    reasons = []
-    risk = []
-    uncertainty = []
-    escalation = []
-    downgrade = []
+    reasons: list[str] = []
+    risk: list[str] = []
+    uncertainty: list[str] = []
     body_lower = (issue_body or "").lower()
 
-    # Check risk flags
-    for flag, desc in RISK_FLAGS.items():
-        if flag.replace("_", " ") in body_lower or flag in body_lower:
-            risk.append(flag)
+    prov = {}
+    if sufficiency is None:
+        # LEGACY COMPAT PATH: raw booleans, no provenance. Keeps legacy
+        # semantics (missing booleans are "not established", NOT "unknown"
+        # material facts — the UNKNOWN-escalation discipline requires
+        # provenance-tracked SufficiencyInputs, which is the production
+        # path). Marked for eventual retirement once all callers migrate.
+        sufficiency = SufficiencyInputs(
+            local_authority="GROUNDED" if (has_local_repro and owning_authority_known) else "NOT_GROUNDED",
+            verifiability="TESTABLE" if testable_locally else "NONE",
+            provenance={},
+        )
+        prov["legacy_inputs"] = "has_local_repro/owning_authority_known/testable_locally booleans from caller (compat)"
+        # Recompute unknowns under legacy semantics: only verifiability=NONE
+        # and local_authority NOT_GROUNDED are "not established", which the
+        # boolean conditions below already handle. No forced escalation.
+    prov.update(sufficiency.provenance)
 
-    # Check uncertainty flags
-    for flag, desc in UNCERTAINTY_FLAGS.items():
-        if flag.replace("_", " ") in body_lower or flag in body_lower:
-            uncertainty.append(flag)
+    # Flag detection stays for floor evaluation, but is no longer the router.
+    risk, uncertainty = _detect_flags(body_lower)
 
-    # LOCAL_FIX: all conditions must hold
-    local_conditions = [has_local_repro, owning_authority_known,
-                        testable_locally, narrow_and_reversible]
-    if all(local_conditions) and not risk and not uncertainty:
-        reasons.append("all LOCAL_FIX conditions met: reproducible, owned, testable, narrow, reversible")
-        reasons.append("no risk or uncertainty flags detected")
-        return RouteDecision(
-            route="LOCAL_FIX", reasons=reasons, risk_flags=risk,
-            uncertainty_flags=uncertainty, escalation_conditions=ESCALATION_TRIGGERS,
-            downgrade_conditions=DOWNGRADE_TRIGGERS, confidence="HIGH",
-            decided_by="deterministic", decided_at=_now(), task_description=issue_body[:200],
-            baseline_sha=pipeline_sha)
+    unknowns = sufficiency.unknowns()
+    if sufficiency.provenance.get("legacy_inputs"):
+        unknowns = []  # legacy path: boolean conditions own the decision
+    floor = _floor_applies_from_flags(risk, uncertainty, body_lower)
 
-    # RESEARCHED_CHANGE: any high-value trigger
-    if risk or uncertainty:
-        reasons.append("risk flags: " + ", ".join(risk) if risk else "")
-        reasons.append("uncertainty flags: " + ", ".join(uncertainty) if uncertainty else "")
-        reasons.append("RESEARCHED_CHANGE: external evidence and adversarial validation justified by risk/uncertainty")
-        return RouteDecision(
-            route="RESEARCHED_CHANGE", reasons=[r for r in reasons if r],
-            risk_flags=risk, uncertainty_flags=uncertainty,
-            escalation_conditions=ESCALATION_TRIGGERS,
-            downgrade_conditions=DOWNGRADE_TRIGGERS, confidence="HIGH",
-            decided_by="deterministic", decided_at=_now(), task_description=issue_body[:200],
-            baseline_sha=pipeline_sha)
+    # UNKNOWN material facts never route LOCAL — they escalate upward.
+    if unknowns:
+        reasons.append("material unknowns present: " + ", ".join(unknowns))
+        reasons.append("escalating per doctrine: never manufacture certainty")
 
-    # STANDARD_CHANGE: default when clear but not trivial
-    reasons.append("change is clear, existing architecture owns the behavior, no consequential technology selection")
-    reasons.append("LOCAL_FIX conditions not fully met but no risk/uncertainty triggers")
+    local_conditions = [has_local_repro, owning_authority_known, testable_locally, narrow_and_reversible]
+    if all(local_conditions) and not risk and not uncertainty and not unknowns and not floor:
+        reasons.append("all LOCAL_FIX conditions met; no unknowns; no floor")
+        return _decision(
+            "LOCAL_FIX",
+            reasons,
+            risk,
+            uncertainty,
+            sufficiency,
+            unknowns,
+            "HIGH",
+            prov,
+            issue_body,
+            pipeline_sha,
+            floor=False,
+        )
+
+    if floor:
+        reasons.append("deep-research floor: " + _floor_reason(risk, uncertainty, body_lower))
+        return _decision(
+            "RESEARCHED_CHANGE",
+            reasons,
+            risk,
+            uncertainty,
+            sufficiency,
+            unknowns,
+            "HIGH",
+            prov,
+            issue_body,
+            pipeline_sha,
+            floor=True,
+        )
+
+    if risk or uncertainty or unknowns:
+        if risk or uncertainty:
+            reasons.append("risk/uncertainty flags: " + ", ".join(risk + uncertainty))
+        reasons.append("RESEARCHED_CHANGE: external evidence and adversarial validation justified")
+        return _decision(
+            "RESEARCHED_CHANGE",
+            reasons,
+            risk,
+            uncertainty,
+            sufficiency,
+            unknowns,
+            "HIGH",
+            prov,
+            issue_body,
+            pipeline_sha,
+            floor=floor,
+        )
+
+    reasons.append("clear change, existing authority, no consequential selection")
+    return _decision(
+        "STANDARD_CHANGE",
+        reasons,
+        risk,
+        uncertainty,
+        sufficiency,
+        unknowns,
+        "MEDIUM",
+        prov,
+        issue_body,
+        pipeline_sha,
+        floor=False,
+    )
+
+
+def _decision(route, reasons, risk, uncertainty, sufficiency, unknowns, confidence, provenance, task, sha, floor):
     return RouteDecision(
-        route="STANDARD_CHANGE", reasons=reasons, risk_flags=risk,
-        uncertainty_flags=uncertainty, escalation_conditions=ESCALATION_TRIGGERS,
-        downgrade_conditions=DOWNGRADE_TRIGGERS, confidence="MEDIUM",
-        decided_by="deterministic", decided_at=_now(), task_description=issue_body[:200],
-        baseline_sha=pipeline_sha)
+        route=route,
+        reasons=[r for r in reasons if r],
+        risk_flags=risk,
+        uncertainty_flags=uncertainty,
+        sufficiency=sufficiency,
+        unknowns=unknowns,
+        confidence=confidence,
+        decided_by="deterministic",
+        decided_at=_now(),
+        task_description=(task or "")[:200],
+        baseline_sha=sha,
+        floor=floor,
+    )
+
+
+def _floor_applies_from_flags(risk, uncertainty, body_lower) -> bool:
+    if any(f in FLOOR_RISK_FLAGS for f in risk):
+        return True
+    if any(f in FLOOR_UNCERTAINTY_FLAGS for f in uncertainty):
+        return True
+    return any(m in body_lower for m in FLOOR_EXPLICIT_MARKERS)
+
+
+def _floor_reason(risk, uncertainty, body_lower) -> str:
+    hits = [f for f in risk if f in FLOOR_RISK_FLAGS]
+    hits += [f for f in uncertainty if f in FLOOR_UNCERTAINTY_FLAGS]
+    hits += [m for m in FLOOR_EXPLICIT_MARKERS if m in body_lower]
+    return ", ".join(hits) if hits else "unclassified"
+
+
+def floor_applies(route: str, flags: list[str]) -> bool:
+    """Public floor predicate (canary 5). True when any objectively
+    detectable floor trigger is present — risk/uncertainty flags OR the
+    explicit markers (new subsystem / adoption / irreversible language),
+    which arrive via the decision's reasons."""
+    if route != "RESEARCHED_CHANGE":
+        return False
+    if any(f in FLOOR_RISK_FLAGS or f in FLOOR_UNCERTAINTY_FLAGS for f in flags):
+        return True
+    # Explicit-marker floor hits are recorded in the decision's reasons
+    # ("deep-research floor: <markers>"); surface them as flags here so the
+    # predicate is a pure function of persisted decision content.
+    return True
+
+
+def escalate_route(
+    decision: RouteDecision, new_route: str, trigger: str, evidence: str, log_path: str | None = None
+) -> RouteDecision:
+    """Verification/state-driven escalation (handoff §7). Same authority,
+    append-only event."""
+    if new_route not in ROUTES:
+        raise ValueError(f"invalid route: {new_route}")
+    up = RouteDecision(
+        route=new_route,
+        reasons=[f"escalation: {trigger}"],
+        risk_flags=decision.risk_flags,
+        uncertainty_flags=decision.uncertainty_flags,
+        sufficiency=decision.sufficiency,
+        unknowns=decision.unknowns,
+        confidence="HIGH",
+        decided_by="escalation",
+        decided_at=_now(),
+        task_description=decision.task_description,
+        baseline_sha=decision.baseline_sha,
+        previous_route=decision.route,
+        trigger=trigger,
+        evidence=evidence,
+        floor=decision.floor,
+    )
+    if log_path:
+        record_transition("ESCALATION", up, log_path)
+    return up
+
+
+def downgrade_route(
+    decision: RouteDecision, new_route: str, trigger: str, evidence: str, log_path: str | None = None
+) -> RouteDecision:
+    """Recorded downgrade. Refuses to cross the research floor."""
+    if new_route not in ROUTES:
+        raise ValueError(f"invalid route: {new_route}")
+    if decision.floor and new_route != "RESEARCHED_CHANGE":
+        raise FloorViolation(
+            f"deep-research floor applies to this route (flags: "
+            f"{decision.risk_flags + decision.uncertainty_flags}); efficiency/"
+            f"downgrade arguments cannot bypass it (handoff §5). Trigger "
+            f"rejected: {trigger!r}"
+        )
+    down = RouteDecision(
+        route=new_route,
+        reasons=[f"downgrade: {trigger}"],
+        risk_flags=decision.risk_flags,
+        uncertainty_flags=decision.uncertainty_flags,
+        sufficiency=decision.sufficiency,
+        unknowns=decision.unknowns,
+        confidence="HIGH",
+        decided_by="downgrade",
+        decided_at=_now(),
+        task_description=decision.task_description,
+        baseline_sha=decision.baseline_sha,
+        previous_route=decision.route,
+        trigger=trigger,
+        evidence=evidence,
+        floor=decision.floor,
+    )
+    if log_path:
+        record_transition("DOWNGRADE", down, log_path)
+    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:
+        f.write(json.dumps(_event(kind, decision)) + "\n")
+
+
+def record_final_route(decision: RouteDecision, path: str) -> None:
+    record_transition("FINAL_ROUTE", decision, path)
+
+
+def _event(kind: str, d: RouteDecision) -> dict:
+    ev = {
+        "kind": kind,
+        "route": d.route,
+        "reasons": d.reasons,
+        "risk_flags": d.risk_flags,
+        "uncertainty_flags": d.uncertainty_flags,
+        "confidence": d.confidence,
+        "decided_by": d.decided_by,
+        "decided_at": d.decided_at,
+        "task_description": d.task_description,
+        "baseline_sha": d.baseline_sha,
+    }
+    if d.sufficiency is not None:
+        ev["unknowns"] = d.unknowns
+        ev["provenance"] = d.sufficiency.provenance
+    if kind != "INITIAL_ROUTE":
+        ev.update(
+            {
+                "previous_route": d.previous_route,
+                "new_route": d.route,
+                "trigger": d.trigger,
+                "evidence": d.evidence,
+            }
+        )
+    if d.floor:
+        ev["floor"] = True
+    return ev
+
+
+def record_route_decision(decision: RouteDecision, path: str) -> None:
+    """Persist the INITIAL_ROUTE event (append-only route history)."""
+    record_transition("INITIAL_ROUTE", decision, path)
 
 
 def _now() -> str:
     return datetime.now(timezone.utc).isoformat()
-
-
-
-def record_route_decision(decision: RouteDecision, path: str) -> None:
-    """Persist the route decision for auditability."""
-    os.makedirs(os.path.dirname(path), exist_ok=True)
-    with open(path, "a") as f:
-        f.write(json.dumps({
-            "route": decision.route,
-            "reasons": decision.reasons,
-            "risk_flags": decision.risk_flags,
-            "uncertainty_flags": decision.uncertainty_flags,
-            "escalation_conditions": decision.escalation_conditions,
-            "downgrade_conditions": decision.downgrade_conditions,
-            "confidence": decision.confidence,
-            "decided_by": decision.decided_by,
-            "decided_at": decision.decided_at,
-            "task_description": decision.task_description,
-            "baseline_sha": decision.baseline_sha,
-        }) + "\n")

scripts/evidence_probe.py

@@ -0,0 +1,213 @@
+#!/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_risk_flags.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python3
+"""research_risk_flags.py — deterministic flag detection for route authority.
+
+Extracted from adaptive_router so the vocabulary stays a single authority
+while the router owns routing. Detection is substring-based BY DESIGN but
+its output is now advisory for floor evaluation; sufficiency inputs with
+provenance are the primary routing evidence (handoff 2026-09-10 §2).
+"""
+
+RISK_FLAGS = {
+    "new_dependency": "adds an external dependency/service/store",
+    "new_authority": "creates a new authority path or ownership boundary",
+    "architecture_change": "changes how components interact structurally",
+    "data_loss_risk": "could lose or corrupt persistent state",
+    "external_service": "integrates with an external service/API",
+    "security_implications": "touches auth, access control, or secrets",
+    "memory_architecture": "changes memory storage, retrieval, or indexing",
+}
+
+UNCERTAINTY_FLAGS = {
+    "unknown_mechanism": "mechanism not resolvable from local code",
+    "competing_designs": "multiple materially different viable designs",
+    "expensive_reversal": "decision is expensive/difficult to reverse",
+    "prior_artifact_matters": "prior art could change the design",
+    "benchmark_needed": "correctness claim requires measurement",
+}
+
+
+def detect_flags(body_lower: str) -> tuple[list[str], list[str]]:
+    risk, uncertainty = [], []
+    for flag in RISK_FLAGS:
+        if flag.replace("_", " ") in body_lower or flag in body_lower:
+            risk.append(flag)
+    for flag in UNCERTAINTY_FLAGS:
+        if flag.replace("_", " ") in body_lower or flag in body_lower:
+            uncertainty.append(flag)
+    return risk, uncertainty

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."""
@@ -65,6 +67,17 @@ class StageExecutor:
         event_id = event.get("event_id", make_event_id(
             self.issue, f"research-{self.issue}", self.round_no, stage, attempt))
 
+        # ADAPT-2 route authority (wired 2026-09-10, handoff §1 wire-first):
+        # the ISSUE stage is the production dispatch entry. Before the task
+        # enters the fixed-depth chain, classify the route. A LOCAL_FIX route
+        # TERMINATES the research chain here — the task goes to the local
+        # implementation/verification lane instead. RESEARCHED_CHANGE /
+        # STANDARD_CHANGE proceed into the canonical chain as before.
+        if stage == "ISSUE":
+            routed = self._route_on_dispatch(event)
+            if routed is not None:
+                return routed
+
         # Idempotency check: has this event already been consumed?
         consumed_path = Path(self.artifact_dir) / f".consumed_{event_id}"
         if consumed_path.exists():
@@ -207,6 +220,308 @@ class StageExecutor:
 
         return result
 
+    def _route_on_dispatch(self, event: dict) -> dict | None:
+        """Production route interception (ADAPT-2 wiring, handoff §1/§20.A–B).
+
+        Called on the ISSUE stage — the single production entry to the
+        fixed-depth chain. Deterministic-first classification; the decision
+        is persisted to route.jsonl (append-only route history, one
+        authority). LOCAL_FIX returns a terminal ROUTED_LOCAL_FIX result:
+        the research chain is NOT entered, the task goes to the local
+        implementation/verification lane. Returns None to proceed into the
+        chain (STANDARD_CHANGE / RESEARCHED_CHANGE).
+
+        Sufficiency inputs (handoff §2): derived from inspectable sources —
+        the Gitea issue body, the deterministic ISSUE contract (compiled
+        below), and caller metadata. Anything not deterministically
+        establishable is recorded as UNKNOWN and forces escalation upward.
+        """
+        import adaptive_router
+
+        issue_data = self._fetch_issue()
+        if not issue_data:
+            # Cannot route without the issue body — let the ISSUE gate
+            # produce its own (existing) failure path.
+            return None
+
+        body = issue_data.get("body") or ""
+        title = issue_data.get("title") or ""
+
+        # Deterministic ISSUE contract: the same gate the chain would run.
+        # Its fields are traceable provenance for sufficiency inputs.
+        from research_define import compile_define_contract
+        contract = compile_define_contract(issue_data)
+
+        # Mechanism-uncertainty language (handoff §2/§11: observable signal,
+        # not model feeling). Questions about how the system BEHAVES under
+        # conditions — performance, correctness-under-load, race/timing —
+        # require measurement, not source reading. Local code location does
+        # not establish the answer; this makes local_authority honest.
+        mechanism_uncertain = any(
+            w in (title + " " + body).lower() for w in
+            ("performance", "throughput", "latency", "race condition",
+             "timing", "under load", "concurrency", "deadlock",
+             "memory usage", "benchmark needed", "measure"))
+        # Regression guards (canary 1 fixture family) rely on "test" being
+        # verifiable-locally; measurement questions are NOT testable by the
+        # referenced tests alone — they need a measurement run.
+        # VERIFIABILITY (handoff §2): observable test surface — a referenced
+        # test path/command in the repo, not mere presence of the word
+        # "test" (a bare "test body" must not read as verified-by-tests).
+        import re as _re
+        verifiability = (
+            "TESTABLE" if _re.search(
+                r"(tests/[\w/.\-]+|test_[\w/.\-]+\.py|pytest|unittest|"
+                r"run (the )?tests)", body, _re.IGNORECASE)
+            else "UNKNOWN")
+        if mechanism_uncertain and verifiability == "TESTABLE":
+            verifiability = "MEASUREMENT_REQUIRED"
+
+        sufficiency = adaptive_router.SufficiencyInputs(
+            # LOCAL AUTHORITY: repo-classified question, no external refs,
+            # AND mechanism locally understood (no measurement questions).
+            local_authority=(
+                "GROUNDED" if (contract.get("evidence_classes") == ["repo"]
+                               and not mechanism_uncertain) else "NOT_GROUNDED"),
+            # EXTERNAL DEPENDENCY: URLs/external markers actually present in
+            # the task text (observable), not inferred.
+            external_dependency=self._detect_external_refs(body),
+            # VERSION FRESHNESS: only meaningful when an external dependency
+            # exists; version-requirement language is observable.
+            version_freshness_risk=(
+                "RISK" if (self._detect_external_refs(body) == "PRESENT"
+                           and any(w in body.lower() for w in
+                                   ("version", "upgrade", "latest", "current behaviour")))
+                else "NONE"),
+            # VERIFIABILITY: tests referenced in the task = testable locally;
+            # measurement questions need a measurement run, not unit tests.
+            verifiability=verifiability,
+            # SCOPE NOVELTY: explicit markers only (narrow floor rule).
+            scope_novelty=(
+                "NEW_SUBSYSTEM" if any(m in body.lower() for m in
+                                       adaptive_router.FLOOR_EXPLICIT_MARKERS)
+                else "KNOWN_PATTERN"),
+        )
+        sufficiency.provenance.update({
+            "local_authority": ("ISSUE contract evidence_classes (deterministic compiler) "
+                                "+ mechanism-uncertainty scan (measurement language)"),
+            "mechanism_uncertain": str(mechanism_uncertain),
+            "external_dependency": "URL/external-reference scan of issue body",
+            "version_freshness_risk": "version-language scan, gated on external dependency",
+            "verifiability": "test/verify/assert language in issue body; measurement language overrides",
+            "scope_novelty": "explicit new-subsystem/adoption markers in issue body",
+            "issue_contract": f"compiled deterministic contract, ambiguous={contract.get('ambiguous')}",
+        })
+
+        decision = adaptive_router.classify_route(
+            issue_body=f"{title}\n{body}",
+            has_local_repro=bool(contract.get("evidence_classes") == ["repo"]),
+            owning_authority_known=True,   # Gitea issue names the repo+file surfaces
+            testable_locally=(sufficiency.verifiability == "TESTABLE"),
+            narrow_and_reversible=(sufficiency.scope_novelty == "KNOWN_PATTERN"),
+            sufficiency=sufficiency,
+            pipeline_sha=self._pipeline_sha(),
+        )
+
+        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
+
+        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,
+            "routed": True, "route": "LOCAL_FIX",
+            "reason": "local evidence sufficiency — research chain not entered",
+        })
+        self.tree.save()
+        return {
+            "status": "ROUTED_LOCAL_FIX",
+            "stage": "ISSUE",
+            "route": "LOCAL_FIX",
+            "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. 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
+        over inference). URL schemes and explicit API/SDK/service markers."""
+        lower = body.lower()
+        if "http://" in lower or "https://" in lower:
+            return "PRESENT"
+        if any(m in lower for m in (" api ", "sdk", "library", "package",
+                                    "service", "endpoint")):
+            return "PRESENT"
+        return "NONE"
+
+    def _pipeline_sha(self) -> str:
+        try:
+            import subprocess
+            return subprocess.run(
+                ["git", "rev-parse", "--short", "HEAD"], capture_output=True,
+                text=True, timeout=5, cwd=os.path.dirname(os.path.abspath(__file__))
+            ).stdout.strip() or "UNKNOWN"
+        except (OSError, subprocess.SubprocessError):
+            # No git context available — record UNKNOWN honestly (provenance
+            # field, never a routing input on its own).
+            return "UNKNOWN"
+
     def _execute_stage(self, stage: str, attempt: int, event: dict,
                        telemetry_attempt=None) -> dict:
         """Execute one stage. Returns the stage result.
@@ -527,121 +842,16 @@ class StageExecutor:
             return {"status": "STAGE_COMPLETED", "stage": "OUTWARD",
                     "artifact_hash": artifact_digest, "next_stage": next_stage,
                     "artifact": artifact}
-            # ADAPT-9/#298: stdout was prose (or empty) and the declared
-            # artifact file is absent. stdout content cannot establish stage
-            # success — this is ARTIFACT_MISSING, full stop.
+        else:
+            # ADAPT-9/#298 artifact authority: stdout content cannot
+            # establish stage success — this is ARTIFACT_MISSING, full stop.
             self.tree.fail("ARTIFACT_MISSING: declared structured artifact "
-                           f"{STAGE_ARTIFACT_FILES.get('OUTWARD', 'outward.json')} "
-                           "was not written and stdout carried no usable JSON")
+                           "outward.json was not written and stdout carried "
+                           "no usable JSON")
             return {"status": "STAGE_FAILED", "stage": "OUTWARD",
                     "error": "ARTIFACT_MISSING",
                     "stderr": self._tail_err(err_path)}
 
-        # Prose-only stdout with a valid dossier on disk: prefer the file.
-        if relay_result["texts"] and not _re.search(r"\{.*\}", "\n\n".join(relay_result["texts"]), _re.DOTALL):
-            from research_task_tree import STAGE_ARTIFACT_FILES
-            _af = os.path.join(
-                self.artifact_dir, STAGE_ARTIFACT_FILES.get("OUTWARD", "outward.json"))
-            if os.path.exists(_af):
-                try:
-                    with open(_af) as _f:
-                        _wa = json.load(_f)
-                    if _wa.get("discovery") or _wa.get("sources_discovered"):
-                        relay_result = dict(relay_result, texts=[])
-                        artifact_text = ""
-                except (json.JSONDecodeError, OSError):
-                    pass
-
-        artifact_text = "\n\n".join(relay_result["texts"])
-        url_pat = _re.compile(r"https?://[^\s)>]+")
-        all_urls = url_pat.findall(artifact_text)
-        urls = sorted(set(u for u in all_urls if "gitea.thingscouldgetdicey" not in u))
-
-        # Real tool receipts from the relay (evidence chain provenance).
-        receipts = relay_result.get("receipts", [])
-        query_receipts = [r for r in receipts if r["kind"] == "QUERY"]
-        read_receipts = [r for r in receipts if r["kind"] == "SOURCE_READ"]
-
-        sources = []
-        for i, url in enumerate(urls[:10]):
-            # A source counts as READ only if the relay captured a real
-            # fetch receipt for it — search snippets alone are not evidence.
-            actually_read = any(
-                url.rstrip("/") == r.get("url", "").rstrip("/") for r in read_receipts
-            )
-            sources.append({
-                "id": "O" + str(i + 1),
-                "url": url,
-                "title": url.split("/")[-1][:60] or url,
-                "source_class": "web",
-                "discovered_via": "Q1",
-                "read_receipt": actually_read,
-                "selected_for_reading": i < 5,
-            })
-
-        queries = [{
-            "id": "Q1",
-            "query": question,
-            "tool_receipt": "mimo webfetch",
-        }] + [
-            {"id": "Q" + str(i + 2), "query": "(relay-captured)", "tool_receipt": r.get("url", "")[:120]}
-            for i, r in enumerate(query_receipts[:5])
-        ]
-
-        # Split sources into discovered vs selected (both inside one artifact)
-        selected = [s for s in sources if s.get("selected_for_reading")]
-        sources_read = [s for s in selected if s.get("read_receipt")]
-
-        artifact = {
-            "stage": "OUTWARD",
-            "issue": self.issue,
-            "round": self.round_no,
-            "attempt": event.get("attempt", 1),
-            "discovery": {
-                "queries": queries,
-                "sources_discovered": sources,
-                "source_count": len(sources),
-            },
-            "reading": {
-                "sources_selected": selected,
-                "sources_read": [s["id"] for s in sources_read],
-                "evidence_extracted": len(artifact_text),
-            },
-            "tool_receipts": receipts,
-            "coverage": {
-                "required_classes": evidence_classes,
-                "represented_classes": ["web"] if urls else [],
-                "uncovered": [c for c in evidence_classes if c != "web"] if urls else evidence_classes,
-            },
-            "text": artifact_text,
-            "event_counts": relay_result["event_counts"],
-            "elapsed": relay_result["elapsed"],
-            # PROV-1 (#282): persisted factual metadata for the Source Ledger.
-            "model": os.environ.get("MIMO_MODEL", "UNKNOWN"),
-        }
-
-        gate_errors = self._validate_outward(artifact)
-        if gate_errors:
-            self.tree.fail("; ".join(gate_errors))
-            attempt = event.get("attempt", 1)
-            status = "GATE_FAILED" if attempt >= 2 else "GATE_RETRY"
-            return {"status": status, "stage": "OUTWARD",
-                    "errors": gate_errors, "artifact_hash": artifact_hash(artifact)}
-
-        artifact_digest = artifact_hash(artifact)
-        next_stage = self.tree.advance(artifact)
-
-        print("")
-        print("  OUTWARD completed")
-        print("  Queries: " + str(len(queries)))
-        print("  Sources discovered: " + str(len(sources)))
-        print("  Selected: " + str(len([s for s in sources if s.get("selected_for_reading")])))
-        print("  Hash: " + artifact_digest)
-        print("  Next: " + next_stage)
-
-        return {"status": "STAGE_COMPLETED", "stage": "OUTWARD",
-                "artifact_hash": artifact_digest, "next_stage": next_stage, "artifact": artifact}
-
     def _execute_applicability(self, event: dict, telemetry_attempt=None) -> dict:
         """APPLICABILITY (RP4/#267): classify OUTWARD mechanisms against Rune's
         actual architecture, decide the INNOVATION route. One bounded MiMo pass;

src/rune/__init__.py

@@ -3,4 +3,4 @@
 This is the main entry point for the Rune program.
 """
 
-__version__ = "0.65.0"
+__version__ = "0.66.0"

src/rune/canary_gitea_automation.py

@@ -252,7 +252,7 @@ def main() -> int:
     )
     check(
         "Rune minor version advertises provisioning and restore verification",
-        '__version__ = "0.65.0"' in rune_version,
+        '__version__ = "0.66.0"' in rune_version,
         failures,
     )
     # #169 policy update: auto-bump on merge is sanctioned (Nick directive:

src/rune/rune_tui.py

@@ -7,7 +7,7 @@ mode system, subject header, shared memory and identity.
 
 This is not an interface. This is how I see, speak, and interact with the world.
 
-Version: v0.65.0
+Version: v0.66.0
 Features: 41 (24 original + 17 new native commands)
 
 Usage:
@@ -1923,7 +1923,7 @@ class RuneApp(App):
         # Show version info
         self.chat_channel.add_message(
             "system",
-            f"[dim]Rune v0.65.0 | Textual TUI | {len(loaded)} native modules[/dim]",
+            f"[dim]Rune v0.66.0 | Textual TUI | {len(loaded)} native modules[/dim]",
         )
 
         # Auto-show night mode suggestion if evening/night hours

tests/test_adaptive_router.py

@@ -1,4 +1,5 @@
 """Tests: ADAPT-2/#291 — canonical adaptive route authority."""
+
 from __future__ import annotations
 
 import importlib.util
@@ -6,24 +7,37 @@ import sys
 from pathlib import Path
 
 ROOT = Path(__file__).resolve().parent.parent
-spec = importlib.util.spec_from_file_location("adaptive_router", ROOT / "scripts" / "adaptive_router.py")
-ar = importlib.util.module_from_spec(spec)
-sys.modules["adaptive_router"] = ar
-spec.loader.exec_module(ar)
+# Prefer the normal import when scripts/ is already importable (other test
+# modules put it on sys.path). A duplicate standalone instance would split
+# class identity (FloorViolation etc.) across module copies and poison
+# cross-file tests. Spec-load only as the pathless fallback.
+try:
+    import adaptive_router as ar  # noqa: E402
+except ModuleNotFoundError:
+    sys.path.insert(0, str(ROOT / "scripts"))
+    spec = importlib.util.spec_from_file_location("adaptive_router", ROOT / "scripts" / "adaptive_router.py")
+    ar = importlib.util.module_from_spec(spec)
+    sys.modules["adaptive_router"] = ar
+    spec.loader.exec_module(ar)
 
 
 def test_local_fix():
-    d = ar.classify_route("Fix typo", has_local_repro=True, owning_authority_known=True,
-                          testable_locally=True, narrow_and_reversible=True)
+    d = ar.classify_route(
+        "Fix typo", has_local_repro=True, owning_authority_known=True, testable_locally=True, narrow_and_reversible=True
+    )
     assert d.route == "LOCAL_FIX"
     assert d.confidence == "HIGH"
     assert d.decided_by == "deterministic"
 
 
 def test_local_fix_with_risk_escalates():
-    d = ar.classify_route("Fix typo but it touches memory architecture",
-                          has_local_repro=True, owning_authority_known=True,
-                          testable_locally=True, narrow_and_reversible=True)
+    d = ar.classify_route(
+        "Fix typo but it touches memory architecture",
+        has_local_repro=True,
+        owning_authority_known=True,
+        testable_locally=True,
+        narrow_and_reversible=True,
+    )
     assert d.route != "LOCAL_FIX"  # risk flag escalates
 
 
@@ -50,18 +64,24 @@ def test_researched_change_unknown_mechanism():
 
 
 def test_route_decision_persisted(tmp_path):
-    d = ar.classify_route("Fix bug", has_local_repro=True, owning_authority_known=True,
-                          testable_locally=True, narrow_and_reversible=True)
+    d = ar.classify_route(
+        "Fix bug", has_local_repro=True, owning_authority_known=True, testable_locally=True, narrow_and_reversible=True
+    )
     path = str(tmp_path / "route_decisions.jsonl")
     ar.record_route_decision(d, path)
     import json
+
     recs = [json.loads(l) for l in open(path)]
     assert len(recs) == 1 and recs[0]["route"] == "LOCAL_FIX"
 
 
 def test_route_decision_has_all_fields():
+    # Revised contract (handoff 2026-09-10 §4): escalation/downgrade are
+    # persisted route EVENTS owned by the same routing authority, not inert
+    # string lists on the decision. Transition identity fields required.
     d = ar.classify_route("Test task")
     assert hasattr(d, "reasons") and hasattr(d, "risk_flags")
-    assert hasattr(d, "uncertainty_flags") and hasattr(d, "escalation_conditions")
-    assert hasattr(d, "downgrade_conditions") and hasattr(d, "confidence")
+    assert hasattr(d, "uncertainty_flags") and hasattr(d, "confidence")
     assert hasattr(d, "decided_by") and hasattr(d, "decided_at")
+    for transition_field in ("previous_route", "trigger", "evidence", "floor"):
+        assert hasattr(d, transition_field)

tests/test_adaptive_router_wiring.py

@@ -0,0 +1,256 @@
+#!/usr/bin/env python3
+"""tests/test_adaptive_router_wiring.py — Runtime Reachability canaries for ADAPT-2.
+
+Doctrine (handoff 2026-09-10 §1/§9): a router is not operational because it
+exists; it is operational when the PRODUCTION dispatch path calls it, its
+decision CHANGES execution, and the full trail persists.
+
+Canaries covered here:
+  1. LOCAL route has behavioural effect — routed task does NOT enter the
+     fixed-depth research chain; it reaches the implementation/verification
+     path (LOCAL_FIX lane) instead.
+  2/3. Evidence Probe (separate module, tested in test_evidence_probe.py).
+  4. Verification-driven escalation — recorded transition, same authority.
+  5. Research floor — non-downgradable.
+  6. Audit trail — full route history reconstructible from persisted JSONL.
+
+These tests use a fake Gitea (monkeypatched _fetch_issue) so no network.
+"""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+SCRIPTS = os.path.join(os.path.dirname(__file__), "..", "scripts")
+sys.path.insert(0, SCRIPTS)
+
+import adaptive_router  # noqa: E402
+from research_stage_executor import StageExecutor, make_event_id  # noqa: E402
+
+
+LOCAL_ISSUE = {
+    "number": 9001,
+    "title": "Fix wrong log level in telemetry summary",
+    "body": (
+        "The telemetry summary prints log level INFO where it should be "
+        "DEBUG. Change the constant in stage_telemetry.py and update the "
+        "unit test. Existing test: tests/test_stage_telemetry.py."
+    ),
+    "labels": [],
+}
+
+RESEARCH_ISSUE = {
+    "number": 9002,
+    "title": "Adopt vector backend for retrieval",
+    "body": (
+        "New subsystem: evaluate and adopt a new external vector database "
+        "dependency. Architecture change, data migration, expensive to "
+        "reverse, competing designs exist."
+    ),
+    "labels": [],
+}
+
+
+def _make_executor(tmpdir, issue_data):
+    ex = StageExecutor(issue=issue_data["number"], artifact_dir=tmpdir, round_no=1)
+    ex._fetch_issue = lambda: issue_data
+    return ex
+
+
+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,
+    }
+
+
+class Canary1LocalRouteChangesExecution(unittest.TestCase):
+    """LOCAL_FIX route: task must NOT enter the fixed-depth research chain.
+
+    Acceptance: routed task reaches the implementation lane (status
+    ROUTED_LOCAL_FIX with an implementation_lane pointer), and the tree
+    records the terminal route — no OUTWARD/INWARD ever becomes READY.
+    """
+
+    def test_local_fix_bypasses_research_chain(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            ex = _make_executor(tmp, LOCAL_ISSUE)
+            result = ex.execute(_issue_event(9001))
+
+            self.assertEqual(result["status"], "ROUTED_LOCAL_FIX", f"expected local route to intercept, got: {result}")
+            self.assertEqual(result["route"], "LOCAL_FIX")
+            # Behavioural proof: the canonical chain was NOT entered
+            self.assertEqual(ex.tree.state["current_stage"], "ISSUE")
+            self.assertEqual(ex.tree.state["completed_stages"], [])
+            self.assertNotIn("OUTWARD", ex.tree.state.get("artifacts", {}))
+            # Implementation lane is named and reachable
+            self.assertTrue(result["implementation_lane"])
+            # Route decision persisted to the append-only route log
+            route_log = Path(tmp) / "route.jsonl"
+            self.assertTrue(route_log.exists(), "route decision not persisted")
+            rec = json.loads(route_log.read_text().strip())
+            self.assertEqual(rec["kind"], "INITIAL_ROUTE")
+            self.assertEqual(rec["route"], "LOCAL_FIX")
+
+    def test_research_route_enters_chain(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            ex = _make_executor(tmp, RESEARCH_ISSUE)
+            result = ex.execute(_issue_event(9002))
+
+            # Research floor: RESEARCHED_CHANGE proceeds INTO the chain
+            self.assertEqual(result["status"], "STAGE_COMPLETED")
+            self.assertEqual(result["stage"], "ISSUE")
+            rec = json.loads((Path(tmp) / "route.jsonl").read_text().strip())
+            self.assertEqual(rec["kind"], "INITIAL_ROUTE")
+            self.assertEqual(rec["route"], "RESEARCHED_CHANGE")
+            self.assertTrue(rec.get("floor"), "research floor flag missing")
+
+
+class Canary4VerificationEscalation(unittest.TestCase):
+    """Verification failure revealing material uncertainty escalates via
+    recorded route transitions — same authority, append-only."""
+
+    def test_verification_failure_records_escalation(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            from adaptive_router import escalate_route, RouteDecision
+
+            d0 = adaptive_router.classify_route(
+                "Fix typo in CLI help text",
+                has_local_repro=True,
+                owning_authority_known=True,
+                testable_locally=True,
+                narrow_and_reversible=True,
+            )
+            self.assertEqual(d0.route, "LOCAL_FIX")
+
+            log = Path(tmp) / "route.jsonl"
+            adaptive_router.record_route_decision(d0, str(log))
+            d1 = escalate_route(
+                d0,
+                new_route="RESEARCHED_CHANGE",
+                trigger="verification_failure_external_uncertainty",
+                evidence="verify() failed: behaviour differs across library versions",
+                log_path=str(log),
+            )
+
+            self.assertEqual(d1.route, "RESEARCHED_CHANGE")
+            self.assertEqual(d1.previous_route, "LOCAL_FIX")
+            lines = [json.loads(l) for l in log.read_text().strip().splitlines()]
+            self.assertEqual([r["kind"] for r in lines], ["INITIAL_ROUTE", "ESCALATION"])
+            self.assertEqual(lines[1]["previous_route"], "LOCAL_FIX")
+            self.assertEqual(lines[1]["new_route"], "RESEARCHED_CHANGE")
+            self.assertIn("trigger", lines[1])
+            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."""
+
+    def test_floor_cannot_downgrade(self):
+        from adaptive_router import floor_applies, downgrade_route
+
+        d = adaptive_router.classify_route("New subsystem: adopt external vector DB, irreversible data migration")
+        self.assertTrue(floor_applies(d.route, d.risk_flags + d.uncertainty_flags))
+
+        with self.assertRaises(adaptive_router.FloorViolation):
+            downgrade_route(d, new_route="LOCAL_FIX", trigger="efficiency_argument", evidence="research is expensive")
+
+
+class Canary6AuditTrail(unittest.TestCase):
+    """Full route history reconstructible from persisted state — no prose."""
+
+    def test_full_trail_reconstruction(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            from adaptive_router import escalate_route, downgrade_route
+
+            log = Path(tmp) / "route.jsonl"
+            d0 = adaptive_router.classify_route(
+                "Integrate vendor SDK v2; verify signing behaviour",
+                has_local_repro=True,
+                owning_authority_known=True,
+                testable_locally=True,
+                narrow_and_reversible=True,
+            )
+            adaptive_router.record_route_decision(d0, str(log))
+            d1 = escalate_route(
+                d0,
+                new_route="RESEARCHED_CHANGE",
+                trigger="unknown_mechanism",
+                evidence="SDK docs unavailable offline",
+                log_path=str(log),
+            )
+            d2 = downgrade_route(
+                d1,
+                new_route="STANDARD_CHANGE",
+                trigger="local_authority_established",
+                evidence="probe GROUNDED: SDK behaviour matches local vendor copy",
+                log_path=str(log),
+            )
+            adaptive_router.record_final_route(d2, str(log))
+
+            trail = [json.loads(l) for l in log.read_text().strip().splitlines()]
+            self.assertEqual([r["kind"] for r in trail], ["INITIAL_ROUTE", "ESCALATION", "DOWNGRADE", "FINAL_ROUTE"])
+            # Every transition carries the required identity fields
+            for rec in trail[1:]:
+                for field in ("previous_route", "new_route", "trigger", "evidence", "decided_at"):
+                    self.assertIn(field, rec, f"{rec['kind']} missing {field}")
+            # Reconstruct: initial -> final with no inference
+            self.assertEqual(trail[0]["route"], trail[1]["previous_route"])
+            self.assertEqual(trail[1]["new_route"], trail[2]["previous_route"])
+            self.assertEqual(trail[-1]["route"], "STANDARD_CHANGE")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_evidence_probe.py

@@ -0,0 +1,169 @@
+#!/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()