Nick
43ef8e43e7
- _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
127 lines
4.8 KiB
Python
127 lines
4.8 KiB
Python
"""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"
|