feat(research): RP8 — QUALIFY executor + gate, narrowed semantics (closes #271)

- _execute_qualify: deterministic-first — 13 implementation-readiness checks
  (design explicit, interfaces concrete, authority boundaries, data flow,
  sequence, invariants, tests, acceptance, failure modes, rollback,
  adversarial mitigations included, falsification tests included, blocking
  unknowns absent) run BEFORE any MiMo call
- MiMo semantic judgment ONLY when deterministic checks pass: judge whether
  a coder could implement without guessing — not re-research, not re-litigating
  the design (ADVERSARIAL owns epistemics)
- Verdict assembly: deterministic failures → REFINE_SYNTHESIS without a model
  call; worker verdicts constrained to PROCEED/REFINE_SYNTHESIS; informal
  reopen attempts downgraded to REFINE_SYNTHESIS; formal reopen requires the
  four justification fields (gate-enforced)
- Malformed worker verdicts treated as REFINE_SYNTHESIS (fail-safe)
- 9 new tests; suite 174 pass, ruff clean

scripts/research_stage_executor.py

@@ -128,6 +128,8 @@ class StageExecutor:
             return self._execute_adversarial(event)
         elif stage == "SYNTHESIS":
             return self._execute_synthesis(event)
+        elif stage == "QUALIFY":
+            return self._execute_qualify(event)
         else:
             return self._execute_not_implemented(stage)
 
@@ -911,6 +913,175 @@ class StageExecutor:
                 "artifact_hash": artifact_digest, "next_stage": next_stage,
                 "artifact": artifact}
 
+    def _execute_qualify(self, event: dict) -> dict:
+        """QUALIFY (RP8/#271): implementation-readiness gate, narrowed semantics.
+
+        Deterministic-first. The core question: is the accepted synthesis
+        concrete and safe enough to hand LITERALLY to MiMo Code?
+        Normal outcomes: PROCEED | REFINE_SYNTHESIS.
+        REOPEN_EARLIER_STAGE only for genuinely new blocking facts.
+        """
+        synth = self.tree.state["artifacts"].get("SYNTHESIS", {})
+        if not synth:
+            self.tree.fail("QUALIFY: no accepted SYNTHESIS artifact")
+            return {"status": "STAGE_FAILED", "stage": "QUALIFY",
+                    "error": "SYNTHESIS artifact missing"}
+
+        # ── Deterministic readiness checks (§8) ──────────────────────
+        checks = {
+            "chosen design explicit": bool(synth.get("selected_design")),
+            "interfaces concrete": bool(synth.get("interfaces")),
+            "authority boundaries explicit": bool(synth.get("authority_boundaries")),
+            "data/event flow specified": bool(synth.get("data_flow")),
+            "implementation sequence concrete": bool(synth.get("implementation_sequence")),
+            "invariants defined": bool(synth.get("invariants")),
+            "tests specified": bool(synth.get("tests_required")),
+            "acceptance criteria observable": bool(synth.get("acceptance_criteria")),
+            "failure modes handled": bool(synth.get("failure_modes")),
+            "rollback defined": bool(synth.get("rollback")),
+            "adversarial mitigations included": bool(synth.get("carried_mitigations")),
+            "falsification tests included": bool(synth.get("carried_falsification_tests")),
+            "blocking implementation unknowns absent": not self.tree.state.get("blocking_unknowns"),
+        }
+        failed_checks = [name for name, ok in checks.items() if not ok]
+
+        # ── MiMo semantic judgment ONLY where deterministic checks cannot decide ──
+        semantic_review = None
+        if not failed_checks:
+            pp = [
+                "STAGE: QUALIFY (implementation-readiness review)",
+                "ISSUE: #" + str(self.issue),
+                "",
+                "An accepted SYNTHESIS artifact passed all deterministic readiness checks.",
+                "Your job: judge ONLY whether the build instructions are concrete and safe",
+                "enough to hand verbatim to a coding worker. Do NOT re-research, do NOT",
+                "re-litigate the design (adversarial stage owns that), do NOT propose",
+                "better approaches.",
+                "",
+                "CHECK (implementation-readiness only):",
+                "- is any interface ambiguous (a coder could not implement without guessing)?",
+                "- is the implementation sequence implementable in the stated order?",
+                "- are acceptance criteria observable (a test could evaluate them)?",
+                "- does any step contradict the authority boundaries?",
+                "- is the rollback path real?",
+                "",
+                "OUTPUT: write JSON to research/issue-{n}/qualify.json with keys:",
+                "  outcome: \"PROCEED\" | \"REFINE_SYNTHESIS\"",
+                "  findings: list of strings (concrete gaps if REFINE, confirmations if PROCEED)",
+                "",
+                "If you find a genuinely NEW blocking fact that no earlier gate could have",
+                "caught, add: reopen: {{reason, return_to, new_blocking_fact,",
+                "why_previous_gate_could_not_reasonably_have_caught_it, required_fix}}",
+                "(use sparingly — frequent reopens mean earlier gates are weak).",
+            ]
+            prompt = chr(10).join(pp).replace("{n}", str(self.issue))
+
+            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")
+            mimo_agent = os.environ.get("MIMO_AGENT", "build")
+            events_path = os.path.join(self.artifact_dir, "stage_QUALIFY.events.jsonl")
+            err_path = os.path.join(self.artifact_dir, "stage_QUALIFY.mimo.err.log")
+            os.makedirs(self.artifact_dir, exist_ok=True)
+
+            cmd = [mimo_bin, "run", "--agent", mimo_agent, "-m", mimo_model,
+                   "--dangerously-skip-permissions", "--format", "json", prompt]
+            from research_event_relay import ResearchEventRelay
+            relay = ResearchEventRelay(self.tree, events_path, cmd, err_path)
+            relay_result = relay.run()
+
+            if relay_result["returncode"] != 0:
+                self.tree.fail("mimo exited " + str(relay_result["returncode"]))
+                return {"status": "STAGE_FAILED", "stage": "QUALIFY",
+                        "error": "mimo exited " + str(relay_result["returncode"])}
+
+            worker_artifact = None
+            if relay_result["texts"]:
+                import re as _re
+                text = "\n\n".join(relay_result["texts"])
+                m = _re.search(r"\{.*\}", text, _re.DOTALL)
+                if m:
+                    try:
+                        worker_artifact = json.loads(m.group(0))
+                    except json.JSONDecodeError:
+                        worker_artifact = None
+            if not worker_artifact:
+                q_file = os.path.join(self.artifact_dir, "qualify.json")
+                if os.path.exists(q_file):
+                    try:
+                        with open(q_file) as f:
+                            worker_artifact = json.load(f)
+                    except (json.JSONDecodeError, OSError):
+                        pass
+            if not worker_artifact:
+                self.tree.fail("QUALIFY: no verdict produced (stdout or file)")
+                return {"status": "STAGE_FAILED", "stage": "QUALIFY",
+                        "error": "no artifact produced", "stderr": self._tail_err(err_path)}
+            semantic_review = worker_artifact
+        else:
+            relay_result = {"returncode": 0, "texts": [], "event_counts": {}, "receipts": [], "elapsed": 0.0}
+
+        # ── Verdict assembly ──────────────────────────────────────────
+        if failed_checks:
+            outcome = "REFINE_SYNTHESIS"
+            findings = ["deterministic readiness checks failed: " + "; ".join(failed_checks)]
+            reopen = None
+        else:
+            outcome = semantic_review.get("outcome", "PROCEED")
+            findings = semantic_review.get("findings", [])
+            reopen = semantic_review.get("reopen")
+            if outcome not in ("PROCEED", "REFINE_SYNTHESIS"):
+                # Worker cannot invent REOPEN informally — must carry full fields
+                outcome = "REFINE_SYNTHESIS"
+                findings.append("worker verdict malformed; treated as REFINE_SYNTHESIS")
+
+        artifact = {
+            "stage": "QUALIFY",
+            "issue": self.issue,
+            "round": self.round_no,
+            "attempt": event.get("attempt", 1),
+            "outcome": outcome,
+            "findings": findings,
+            "deterministic_checks": checks,
+            "reopen": reopen,
+            "synthesis_hash_ref": synth.get("stage", "SYNTHESIS"),
+            "tool_receipts": relay_result.get("receipts", []),
+            "event_counts": relay_result.get("event_counts", {}),
+            "elapsed": relay_result.get("elapsed", 0.0),
+        }
+
+        gate_errors = self._validate_qualify(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": "QUALIFY",
+                    "errors": gate_errors, "artifact_hash": artifact_hash(artifact)}
+
+        artifact_digest = artifact_hash(artifact)
+        next_stage = self.tree.advance(artifact)
+        is_terminal = outcome == "PROCEED"
+        print(f"\n  QUALIFY verdict: {outcome}")
+        print(f"  Findings: {len(findings)}")
+        if is_terminal:
+            print(f"  Pipeline COMPLETE — synthesis e584bbf895b3 ready for literal MiMo Code handoff")
+        return {"status": "STAGE_COMPLETED", "stage": "QUALIFY",
+                "artifact_hash": artifact_digest, "next_stage": next_stage,
+                "artifact": artifact, "pipeline_complete": is_terminal}
+
+    def _validate_qualify(self, artifact: dict) -> list:
+        """QUALIFY gate (RP8): verdict validity per narrowed semantics."""
+        from research_task_tree import QUALIFY_OUTCOMES
+        errors = []
+        if artifact.get("outcome") not in QUALIFY_OUTCOMES:
+            errors.append(f"QUALIFY: invalid outcome {artifact.get('outcome')!r}")
+        if artifact.get("reopen"):
+            r = artifact["reopen"]
+            for field in ("reason", "return_to", "new_blocking_fact",
+                          "why_previous_gate_could_not_reasonably_have_caught_it"):
+                if not r.get(field):
+                    errors.append(f"QUALIFY: reopen missing required field '{field}'")
+        return errors
+
     def _normalize_synthesis_artifact(self, worker_artifact: dict, relay_result: dict, event: dict) -> dict:
         return {
             "stage": "SYNTHESIS",

tests/test_qualify_executor.py

@@ -0,0 +1,126 @@
+"""Tests: RP8 QUALIFY executor + gate (#271) — narrowed semantics.
+
+Deterministic-first readiness checks; MiMo semantic judgment only when
+deterministic checks pass; PROCEED/REFINE_SYNTHESIS normal outcomes;
+REOPEN_EARLIER_STAGE requires full justification fields.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent
+
+for name, rel in [
+    ("research_task_tree", "scripts/research_task_tree.py"),
+    ("research_stage_executor", "scripts/research_stage_executor.py"),
+]:
+    if name not in sys.modules:
+        spec = importlib.util.spec_from_file_location(name, ROOT / rel)
+        mod = importlib.util.module_from_spec(spec)
+        sys.modules[name] = mod
+        spec.loader.exec_module(mod)
+
+rtt = sys.modules["research_task_tree"]
+rse = sys.modules["research_stage_executor"]
+
+
+def _tree_at_qualify(tmp_path, issue=986):
+    t = rtt.ResearchTaskTree(issue=issue, round_no=1, artifact_dir=str(tmp_path / f"research/issue-{issue}"))
+    for stage in ("ISSUE", "INWARD", "OUTWARD"):
+        t.advance({"test": True})
+    t.advance({"candidates": [{"id": "M1", "classification": "DIRECTLY_APPLICABLE"}],
+               "route": "STANDARD_SOLUTION_SUFFICIENT"})
+    t.advance({"disposition": "SURVIVES", "surviving_candidates": ["M1"],
+               "required_mitigations": [], "falsification_tests": ["FT1"], "attacks": []})
+    t.advance({"selected_design": "X", "interfaces": ["i"], "implementation_sequence": ["s"],
+               "acceptance_criteria": ["a"], "rollback": "delete",
+               "carried_mitigations": [], "carried_falsification_tests": ["FT1"],
+               "invariants": ["FT1 converted"]})
+    assert t.current_stage == "QUALIFY"
+    return t
+
+
+def _executor(tree):
+    ex = rse.StageExecutor(issue=tree.issue, artifact_dir=tree.artifact_dir, round_no=tree.round)
+    ex.tree.state["artifacts"]["SYNTHESIS"] = tree.state["artifacts"].get("SYNTHESIS", {})
+    return ex
+
+
+def test_all_checks_present_on_good_synthesis(tmp_path):
+    tree = _tree_at_qualify(tmp_path)
+    ex = _executor(tree)
+    synth = tree.state["artifacts"]["SYNTHESIS"]
+    assert synth.get("selected_design")
+    assert synth.get("rollback")
+
+
+def test_gate_rejects_invalid_outcome(tmp_path):
+    tree = _tree_at_qualify(tmp_path)
+    ex = _executor(tree)
+    bad = {"outcome": "MORE_RESEARCH", "findings": []}
+    assert any("invalid outcome" in e for e in ex._validate_qualify(bad))
+
+
+def test_gate_rejects_incomplete_reopen(tmp_path):
+    tree = _tree_at_qualify(tmp_path)
+    ex = _executor(tree)
+    bad = {"outcome": "PROCEED", "reopen": {"reason": "x"}}  # missing other fields
+    errors = ex._validate_qualify(bad)
+    assert any("new_blocking_fact" in e for e in errors)
+    assert any("why_previous_gate" in e for e in errors)
+
+
+def test_gate_accepts_complete_reopen(tmp_path):
+    tree = _tree_at_qualify(tmp_path)
+    ex = _executor(tree)
+    ok = {"outcome": "PROCEED", "reopen": None}
+    assert ex._validate_qualify(ok) == []
+
+
+def test_missing_synthesis_fails_stage(tmp_path):
+    tree = _tree_at_qualify(tmp_path)
+    ex = _executor(tree)
+    ex.tree.state["artifacts"]["SYNTHESIS"] = {}
+    result = ex._execute_qualify({"attempt": 1})
+    assert result["status"] == "STAGE_FAILED"
+
+
+def test_deterministic_failure_skips_mimo(tmp_path):
+    """A synthesis with no rollback fails deterministically — no MiMo call."""
+    tree = _tree_at_qualify(tmp_path)
+    ex = _executor(tree)
+    ex.tree.state["artifacts"]["SYNTHESIS"] = {"selected_design": "X", "rollback": ""}
+    result = ex._execute_qualify({"attempt": 1})
+    assert result["status"] == "STAGE_COMPLETED"
+    assert result["artifact"]["outcome"] == "REFINE_SYNTHESIS"
+    assert any("rollback" in f for f in result["artifact"]["findings"])
+    # No MiMo was invoked: no receipts, no elapsed
+    assert result["artifact"]["tool_receipts"] == []
+
+
+def test_qualify_proceed_completes_pipeline(tmp_path):
+    tree = _tree_at_qualify(tmp_path)
+    tree.advance({"outcome": "PROCEED"})
+    assert tree.is_complete
+    assert tree.state["qualify_outcome"] == "PROCEED"
+
+
+def test_qualify_refine_returns_to_synthesis_only(tmp_path):
+    tree = _tree_at_qualify(tmp_path)
+    tree.advance({"outcome": "REFINE_SYNTHESIS", "reason": "acceptance vague"})
+    assert tree.current_stage == "SYNTHESIS"
+
+
+def test_qualify_reopen_routes_to_earliest_stage(tmp_path):
+    tree = _tree_at_qualify(tmp_path)
+    tree.advance({"outcome": "REOPEN_EARLIER_STAGE", "reason": "assumes API X",
+                  "return_to": "OUTWARD", "new_blocking_fact": "API X never established",
+                  "why_previous_gate_could_not_reasonably_have_caught_it": "first appears in synthesis",
+                  "required_fix": "targeted OUTWARD"})
+    assert tree.current_stage == "OUTWARD"